mirror of https://github.com/trapexit/mergerfs.git
Browse Source
Merge pull request #1118 from trapexit/msghandling
Merge pull request #1118 from trapexit/msghandling
Add async message processingpull/1119/head
trapexit
2 years ago
committed by
GitHub
GitHub
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 1526 additions and 755 deletions
-
199README.md
-
27libfuse/Makefile
-
9libfuse/include/fuse.h
-
14libfuse/include/fuse_lowlevel.h
-
12libfuse/include/fuse_msgbuf.h
-
165libfuse/lib/bounded_queue.hpp
-
72libfuse/lib/fuse.c
-
19libfuse/lib/fuse_i.h
-
15libfuse/lib/fuse_ll.hpp
-
231libfuse/lib/fuse_loop_mt.c
-
266libfuse/lib/fuse_loop_mt.cpp
-
373libfuse/lib/fuse_lowlevel.c
-
123libfuse/lib/fuse_msgbuf.cpp
-
29libfuse/lib/fuse_msgbuf.hpp
-
3libfuse/lib/fuse_mt.c
-
102libfuse/lib/fuse_session.c
-
112libfuse/lib/thread_pool.hpp
-
161libfuse/lib/unbounded_queue.hpp
-
7src/config.cpp
-
3src/config.hpp
-
13src/option_parser.cpp
-
14tools/create-branches
@ -0,0 +1,12 @@ |
|||
#pragma once |
|||
|
|||
#include <stdint.h> |
|||
|
|||
typedef struct fuse_msgbuf_t fuse_msgbuf_t; |
|||
struct fuse_msgbuf_t |
|||
{ |
|||
char *mem; |
|||
uint32_t size; |
|||
uint32_t used; |
|||
int pipefd[2]; |
|||
}; |
|||
@ -0,0 +1,165 @@ |
|||
#pragma once
|
|||
|
|||
#include <condition_variable>
|
|||
#include <mutex>
|
|||
#include <queue>
|
|||
#include <utility>
|
|||
|
|||
|
|||
template<typename T> |
|||
class BoundedQueue |
|||
{ |
|||
public: |
|||
explicit |
|||
BoundedQueue(std::size_t max_size_, |
|||
bool block_ = true) |
|||
: _block(block), |
|||
_max_size(max_size_) |
|||
{ |
|||
if(_max_size == 0) |
|||
_max_size = 1; |
|||
} |
|||
|
|||
bool |
|||
push(const T& item_) |
|||
{ |
|||
{ |
|||
std::unique_lock guard(_queue_lock); |
|||
|
|||
_condition_push.wait(guard, [&]() { return _queue.size() < _max_size || !_block; }); |
|||
|
|||
if(_queue.size() == _max_size) |
|||
return false; |
|||
|
|||
_queue.push(item); |
|||
} |
|||
|
|||
_condition_pop.notify_one(); |
|||
|
|||
return true; |
|||
} |
|||
|
|||
bool |
|||
push(T&& item_) |
|||
{ |
|||
{ |
|||
std::unique_lock guard(_queue_lock); |
|||
|
|||
_condition_push.wait(guard, [&]() { return _queue.size() < _max_size || !_block; }); |
|||
|
|||
if(_queue.size() == _max_size) |
|||
return false; |
|||
|
|||
_queue.push(std::move(item_)); |
|||
} |
|||
_condition_pop.notify_one(); |
|||
return true; |
|||
} |
|||
|
|||
template<typename... Args> |
|||
bool |
|||
emplace(Args&&... args_) |
|||
{ |
|||
{ |
|||
std::unique_lock guard(_queue_lock); |
|||
|
|||
_condition_push.wait(guard, [&]() { return _queue.size() < _max_size || !_block; }); |
|||
|
|||
if(_queue.size() == _max_size) |
|||
return false; |
|||
|
|||
_queue.emplace(std::forward<Args>(args_)...); |
|||
} |
|||
|
|||
_condition_pop.notify_one(); |
|||
|
|||
return true; |
|||
} |
|||
|
|||
bool |
|||
pop(T& item_) |
|||
{ |
|||
{ |
|||
std::unique_lock guard(_queue_lock); |
|||
|
|||
_condition_pop.wait(guard, [&]() { return !_queue.empty() || !_block; }); |
|||
if(_queue.empty()) |
|||
return false; |
|||
|
|||
item_ = std::move(_queue.front()); |
|||
|
|||
_queue.pop(); |
|||
} |
|||
|
|||
_condition_push.notify_one(); |
|||
|
|||
return true; |
|||
} |
|||
|
|||
std::size_t |
|||
size() const |
|||
{ |
|||
std::lock_guard guard(_queue_lock); |
|||
|
|||
return _queue.size(); |
|||
} |
|||
|
|||
std::size_t |
|||
capacity() const |
|||
{ |
|||
return _max_size; |
|||
} |
|||
|
|||
bool |
|||
empty() const |
|||
{ |
|||
std::lock_guard guard(_queue_lock); |
|||
|
|||
return _queue.empty(); |
|||
} |
|||
|
|||
bool |
|||
full() const |
|||
{ |
|||
std::lock_guard lock(_queue_lock); |
|||
|
|||
return (_queue.size() == capacity()); |
|||
} |
|||
|
|||
void |
|||
block() |
|||
{ |
|||
std::lock_guard guard(_queue_lock); |
|||
_block = true; |
|||
} |
|||
|
|||
void |
|||
unblock() |
|||
{ |
|||
{ |
|||
std::lock_guard guard(_queue_lock); |
|||
_block = false; |
|||
} |
|||
|
|||
_condition_push.notify_all(); |
|||
_condition_pop.notify_all(); |
|||
} |
|||
|
|||
bool |
|||
blocking() const |
|||
{ |
|||
std::lock_guard guard(_queue_lock); |
|||
|
|||
return _block; |
|||
} |
|||
|
|||
private: |
|||
mutable std::mutex _queue_lock; |
|||
|
|||
private: |
|||
bool _block; |
|||
std::queue<T> _queue; |
|||
const std::size_t _max_size; |
|||
std::condition_variable _condition_push; |
|||
std::condition_variable _condition_pop; |
|||
}; |
|||
@ -0,0 +1,15 @@ |
|||
#pragma once
|
|||
|
|||
#include "fuse_msgbuf.h"
|
|||
|
|||
int fuse_receive_buf(struct fuse_session *se, |
|||
fuse_msgbuf_t *msgbuf); |
|||
void fuse_process_buf(void *data, |
|||
const fuse_msgbuf_t *msgbuf, |
|||
struct fuse_chan *ch); |
|||
|
|||
int fuse_receive_buf_splice(struct fuse_chan *ch, |
|||
fuse_msgbuf_t *msgbuf); |
|||
void fuse_process_buf_splice(struct fuse_chan *ch, |
|||
const fuse_msgbuf_t *msgbuf, |
|||
void *data); |
|||
@ -1,231 +0,0 @@ |
|||
/* |
|||
FUSE: Filesystem in Userspace |
|||
Copyright (C) 2001-2007 Miklos Szeredi <miklos@szeredi.hu> |
|||
|
|||
This program can be distributed under the terms of the GNU LGPLv2. |
|||
See the file COPYING.LIB. |
|||
*/ |
|||
|
|||
#include "fuse_i.h" |
|||
#include "fuse_kernel.h" |
|||
#include "fuse_lowlevel.h" |
|||
#include "fuse_misc.h" |
|||
|
|||
#include <errno.h> |
|||
#include <semaphore.h> |
|||
#include <signal.h> |
|||
#include <stdio.h> |
|||
#include <stdlib.h> |
|||
#include <string.h> |
|||
#include <sys/time.h> |
|||
#include <unistd.h> |
|||
#include <unistd.h> |
|||
|
|||
/* Environment var controlling the thread stack size */ |
|||
#define ENVNAME_THREAD_STACK "FUSE_THREAD_STACK" |
|||
|
|||
struct fuse_worker |
|||
{ |
|||
struct fuse_worker *prev; |
|||
struct fuse_worker *next; |
|||
pthread_t thread_id; |
|||
size_t bufsize; |
|||
char *buf; |
|||
struct fuse_mt *mt; |
|||
}; |
|||
|
|||
struct fuse_mt |
|||
{ |
|||
struct fuse_session *se; |
|||
struct fuse_worker main; |
|||
sem_t finish; |
|||
int exit; |
|||
int error; |
|||
}; |
|||
|
|||
static |
|||
void |
|||
list_add_worker(struct fuse_worker *w, |
|||
struct fuse_worker *next) |
|||
{ |
|||
struct fuse_worker *prev = next->prev; |
|||
w->next = next; |
|||
w->prev = prev; |
|||
prev->next = w; |
|||
next->prev = w; |
|||
} |
|||
|
|||
static void list_del_worker(struct fuse_worker *w) |
|||
{ |
|||
struct fuse_worker *prev = w->prev; |
|||
struct fuse_worker *next = w->next; |
|||
prev->next = next; |
|||
next->prev = prev; |
|||
} |
|||
|
|||
static int fuse_loop_start_thread(struct fuse_mt *mt); |
|||
|
|||
static |
|||
void* |
|||
fuse_do_work(void *data) |
|||
{ |
|||
struct fuse_worker *w = (struct fuse_worker *) data; |
|||
struct fuse_mt *mt = w->mt; |
|||
|
|||
while(!fuse_session_exited(mt->se)) |
|||
{ |
|||
int res; |
|||
struct fuse_buf fbuf; |
|||
|
|||
fbuf = (struct fuse_buf){ .mem = w->buf, |
|||
.size = w->bufsize }; |
|||
|
|||
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL); |
|||
res = fuse_session_receive(mt->se,&fbuf); |
|||
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); |
|||
if(res == -EINTR) |
|||
continue; |
|||
if(res <= 0) |
|||
{ |
|||
if(res < 0) |
|||
{ |
|||
mt->se->exited = 1; |
|||
mt->error = -1; |
|||
} |
|||
break; |
|||
} |
|||
|
|||
if(mt->exit) |
|||
return NULL; |
|||
|
|||
fuse_session_process(mt->se,&fbuf); |
|||
} |
|||
|
|||
sem_post(&mt->finish); |
|||
|
|||
return NULL; |
|||
} |
|||
|
|||
int fuse_start_thread(pthread_t *thread_id, void *(*func)(void *), void *arg) |
|||
{ |
|||
sigset_t oldset; |
|||
sigset_t newset; |
|||
int res; |
|||
pthread_attr_t attr; |
|||
char *stack_size; |
|||
|
|||
/* Override default stack size */ |
|||
pthread_attr_init(&attr); |
|||
stack_size = getenv(ENVNAME_THREAD_STACK); |
|||
if(stack_size && pthread_attr_setstacksize(&attr, atoi(stack_size))) |
|||
fprintf(stderr, "fuse: invalid stack size: %s\n", stack_size); |
|||
|
|||
/* Disallow signal reception in worker threads */ |
|||
sigfillset(&newset); |
|||
pthread_sigmask(SIG_BLOCK,&newset,&oldset); |
|||
res = pthread_create(thread_id, &attr, func, arg); |
|||
pthread_sigmask(SIG_SETMASK,&oldset,NULL); |
|||
pthread_attr_destroy(&attr); |
|||
if(res != 0) { |
|||
fprintf(stderr, "fuse: error creating thread: %s\n", |
|||
strerror(res)); |
|||
return -1; |
|||
} |
|||
|
|||
return 0; |
|||
} |
|||
|
|||
static int fuse_loop_start_thread(struct fuse_mt *mt) |
|||
{ |
|||
int res; |
|||
struct fuse_worker *w = malloc(sizeof(struct fuse_worker)); |
|||
if(!w) { |
|||
fprintf(stderr, "fuse: failed to allocate worker structure\n"); |
|||
return -1; |
|||
} |
|||
memset(w, 0, sizeof(struct fuse_worker)); |
|||
w->bufsize = fuse_chan_bufsize(mt->se->ch); |
|||
w->buf = calloc(w->bufsize,1); |
|||
w->mt = mt; |
|||
if(!w->buf) { |
|||
fprintf(stderr, "fuse: failed to allocate read buffer\n"); |
|||
free(w); |
|||
return -1; |
|||
} |
|||
|
|||
res = fuse_start_thread(&w->thread_id, fuse_do_work, w); |
|||
if(res == -1) { |
|||
free(w->buf); |
|||
free(w); |
|||
return -1; |
|||
} |
|||
list_add_worker(w, &mt->main); |
|||
|
|||
return 0; |
|||
} |
|||
|
|||
static void fuse_join_worker(struct fuse_worker *w) |
|||
{ |
|||
pthread_join(w->thread_id, NULL); |
|||
list_del_worker(w); |
|||
free(w->buf); |
|||
free(w); |
|||
} |
|||
|
|||
static int number_of_threads(void) |
|||
{ |
|||
#ifdef _SC_NPROCESSORS_ONLN |
|||
return sysconf(_SC_NPROCESSORS_ONLN); |
|||
#endif |
|||
|
|||
return 4; |
|||
} |
|||
|
|||
int |
|||
fuse_session_loop_mt(struct fuse_session *se_, |
|||
const int threads_) |
|||
{ |
|||
int i; |
|||
int err; |
|||
int threads; |
|||
struct fuse_mt mt; |
|||
struct fuse_worker *w; |
|||
|
|||
memset(&mt,0,sizeof(struct fuse_mt)); |
|||
mt.se = se_; |
|||
mt.error = 0; |
|||
mt.main.thread_id = pthread_self(); |
|||
mt.main.prev = mt.main.next = &mt.main; |
|||
sem_init(&mt.finish,0,0); |
|||
|
|||
threads = ((threads_ > 0) ? threads_ : number_of_threads()); |
|||
if(threads_ < 0) |
|||
threads /= -threads_; |
|||
if(threads == 0) |
|||
threads = 1; |
|||
|
|||
err = 0; |
|||
for(i = 0; (i < threads) && !err; i++) |
|||
err = fuse_loop_start_thread(&mt); |
|||
|
|||
if(!err) |
|||
{ |
|||
/* sem_wait() is interruptible */ |
|||
while(!fuse_session_exited(se_)) |
|||
sem_wait(&mt.finish); |
|||
|
|||
for(w = mt.main.next; w != &mt.main; w = w->next) |
|||
pthread_cancel(w->thread_id); |
|||
mt.exit = 1; |
|||
|
|||
while(mt.main.next != &mt.main) |
|||
fuse_join_worker(mt.main.next); |
|||
|
|||
err = mt.error; |
|||
} |
|||
|
|||
sem_destroy(&mt.finish); |
|||
fuse_session_reset(se_); |
|||
|
|||
return err; |
|||
} |
|||
@ -0,0 +1,266 @@ |
|||
#include "thread_pool.hpp"
|
|||
|
|||
#include "fuse_i.h"
|
|||
#include "fuse_kernel.h"
|
|||
#include "fuse_lowlevel.h"
|
|||
#include "fuse_misc.h"
|
|||
|
|||
#include "fuse_msgbuf.hpp"
|
|||
#include "fuse_ll.hpp"
|
|||
|
|||
#include <errno.h>
|
|||
#include <semaphore.h>
|
|||
#include <signal.h>
|
|||
#include <stdio.h>
|
|||
#include <stdlib.h>
|
|||
#include <string.h>
|
|||
#include <sys/time.h>
|
|||
#include <unistd.h>
|
|||
|
|||
#include <cassert>
|
|||
#include <vector>
|
|||
|
|||
|
|||
struct fuse_worker_data_t |
|||
{ |
|||
struct fuse_session *se; |
|||
sem_t finished; |
|||
std::function<void(fuse_worker_data_t*,fuse_msgbuf_t*)> msgbuf_processor; |
|||
std::function<fuse_msgbuf_t*(void)> msgbuf_allocator; |
|||
std::shared_ptr<ThreadPool> tp; |
|||
}; |
|||
|
|||
class WorkerCleanup |
|||
{ |
|||
public: |
|||
WorkerCleanup(fuse_worker_data_t *wd_) |
|||
: _wd(wd_) |
|||
{ |
|||
} |
|||
|
|||
~WorkerCleanup() |
|||
{ |
|||
fuse_session_exit(_wd->se); |
|||
sem_post(&_wd->finished); |
|||
} |
|||
|
|||
private: |
|||
fuse_worker_data_t *_wd; |
|||
}; |
|||
|
|||
static |
|||
bool |
|||
retriable_receive_error(const int err_) |
|||
{ |
|||
switch(err_) |
|||
{ |
|||
case -EINTR: |
|||
case -EAGAIN: |
|||
case -ENOENT: |
|||
return true; |
|||
default: |
|||
return false; |
|||
} |
|||
} |
|||
|
|||
static |
|||
bool |
|||
fatal_receive_error(const int err_) |
|||
{ |
|||
return (err_ < 0); |
|||
} |
|||
|
|||
static |
|||
void* |
|||
handle_receive_error(const int rv_, |
|||
fuse_msgbuf_t *msgbuf_) |
|||
{ |
|||
msgbuf_free(msgbuf_); |
|||
|
|||
fprintf(stderr, |
|||
"mergerfs: error reading from /dev/fuse - %s (%d)\n", |
|||
strerror(-rv_), |
|||
-rv_); |
|||
|
|||
return NULL; |
|||
} |
|||
|
|||
static |
|||
void* |
|||
fuse_do_work(void *data) |
|||
{ |
|||
fuse_worker_data_t *wd = (fuse_worker_data_t*)data; |
|||
fuse_session *se = wd->se; |
|||
auto &process_msgbuf = wd->msgbuf_processor; |
|||
auto &msgbuf_allocator = wd->msgbuf_allocator; |
|||
WorkerCleanup workercleanup(wd); |
|||
|
|||
while(!fuse_session_exited(se)) |
|||
{ |
|||
int rv; |
|||
fuse_msgbuf_t *msgbuf; |
|||
|
|||
msgbuf = msgbuf_allocator(); |
|||
|
|||
do |
|||
{ |
|||
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,NULL); |
|||
rv = se->receive_buf(se,msgbuf); |
|||
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE,NULL); |
|||
if(rv == 0) |
|||
return NULL; |
|||
if(retriable_receive_error(rv)) |
|||
continue; |
|||
if(fatal_receive_error(rv)) |
|||
return handle_receive_error(rv,msgbuf); |
|||
} while(false); |
|||
|
|||
process_msgbuf(wd,msgbuf); |
|||
} |
|||
|
|||
return NULL; |
|||
} |
|||
|
|||
int |
|||
fuse_start_thread(pthread_t *thread_id, |
|||
void *(*func)(void *), |
|||
void *arg) |
|||
{ |
|||
int res; |
|||
sigset_t oldset; |
|||
sigset_t newset; |
|||
|
|||
sigfillset(&newset); |
|||
pthread_sigmask(SIG_BLOCK,&newset,&oldset); |
|||
res = pthread_create(thread_id,NULL,func,arg); |
|||
pthread_sigmask(SIG_SETMASK,&oldset,NULL); |
|||
|
|||
if(res != 0) |
|||
{ |
|||
fprintf(stderr, |
|||
"fuse: error creating thread: %s\n", |
|||
strerror(res)); |
|||
return -1; |
|||
} |
|||
|
|||
return 0; |
|||
} |
|||
|
|||
static |
|||
int |
|||
calculate_thread_count(const int raw_thread_count_) |
|||
{ |
|||
int thread_count; |
|||
|
|||
thread_count = 4; |
|||
if(raw_thread_count_ == 0) |
|||
thread_count = std::thread::hardware_concurrency(); |
|||
else if(raw_thread_count_ < 0) |
|||
thread_count = (std::thread::hardware_concurrency() / -raw_thread_count_); |
|||
else if(raw_thread_count_ > 0) |
|||
thread_count = raw_thread_count_; |
|||
|
|||
if(thread_count <= 0) |
|||
thread_count = 1; |
|||
|
|||
return thread_count; |
|||
} |
|||
|
|||
static |
|||
void |
|||
calculate_thread_counts(int *read_thread_count_, |
|||
int *process_thread_count_) |
|||
{ |
|||
if((*read_thread_count_ == -1) && (*process_thread_count_ == -1)) |
|||
{ |
|||
int nproc; |
|||
|
|||
nproc = std::thread::hardware_concurrency(); |
|||
*read_thread_count_ = 2; |
|||
*process_thread_count_ = std::max(2,(nproc - 2)); |
|||
} |
|||
else |
|||
{ |
|||
*read_thread_count_ = ::calculate_thread_count(*read_thread_count_); |
|||
if(*process_thread_count_ != -1) |
|||
*process_thread_count_ = ::calculate_thread_count(*process_thread_count_); |
|||
} |
|||
} |
|||
|
|||
static |
|||
void |
|||
process_msgbuf_sync(fuse_worker_data_t *wd_, |
|||
fuse_msgbuf_t *msgbuf_) |
|||
{ |
|||
wd_->se->process_buf(wd_->se,msgbuf_); |
|||
msgbuf_free(msgbuf_); |
|||
} |
|||
|
|||
static |
|||
void |
|||
process_msgbuf_async(fuse_worker_data_t *wd_, |
|||
fuse_msgbuf_t *msgbuf_) |
|||
{ |
|||
const auto func = [=] { |
|||
process_msgbuf_sync(wd_,msgbuf_); |
|||
}; |
|||
|
|||
wd_->tp->enqueue_work(func); |
|||
} |
|||
|
|||
int |
|||
fuse_session_loop_mt(struct fuse_session *se_, |
|||
const int raw_read_thread_count_, |
|||
const int raw_process_thread_count_) |
|||
{ |
|||
int err; |
|||
int read_thread_count; |
|||
int process_thread_count; |
|||
fuse_worker_data_t wd = {0}; |
|||
std::vector<pthread_t> threads; |
|||
|
|||
read_thread_count = raw_read_thread_count_; |
|||
process_thread_count = raw_process_thread_count_; |
|||
::calculate_thread_counts(&read_thread_count,&process_thread_count); |
|||
|
|||
if(process_thread_count > 0) |
|||
{ |
|||
wd.tp = std::make_shared<ThreadPool>(process_thread_count); |
|||
wd.msgbuf_processor = process_msgbuf_async; |
|||
} |
|||
else |
|||
{ |
|||
wd.msgbuf_processor = process_msgbuf_sync; |
|||
} |
|||
|
|||
wd.msgbuf_allocator = ((se_->f->splice_read) ? msgbuf_alloc : msgbuf_alloc_memonly); |
|||
|
|||
wd.se = se_; |
|||
sem_init(&wd.finished,0,0); |
|||
|
|||
err = 0; |
|||
for(int i = 0; i < read_thread_count; i++) |
|||
{ |
|||
pthread_t thread_id; |
|||
err = fuse_start_thread(&thread_id,fuse_do_work,&wd); |
|||
assert(err == 0); |
|||
threads.push_back(thread_id); |
|||
} |
|||
|
|||
if(!err) |
|||
{ |
|||
/* sem_wait() is interruptible */ |
|||
while(!fuse_session_exited(se_)) |
|||
sem_wait(&wd.finished); |
|||
|
|||
for(const auto &thread_id : threads) |
|||
pthread_cancel(thread_id); |
|||
|
|||
for(const auto &thread_id : threads) |
|||
pthread_join(thread_id,NULL); |
|||
} |
|||
|
|||
sem_destroy(&wd.finished); |
|||
|
|||
return err; |
|||
} |
|||
@ -0,0 +1,123 @@ |
|||
/*
|
|||
ISC License |
|||
|
|||
Copyright (c) 2022, Antonio SJ Musumeci <trapexit@spawn.link> |
|||
|
|||
Permission to use, copy, modify, and/or distribute this software for any |
|||
purpose with or without fee is hereby granted, provided that the above |
|||
copyright notice and this permission notice appear in all copies. |
|||
|
|||
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES |
|||
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF |
|||
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR |
|||
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES |
|||
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN |
|||
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF |
|||
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. |
|||
*/ |
|||
|
|||
#include "fuse_msgbuf.h"
|
|||
|
|||
#include <assert.h>
|
|||
#include <fcntl.h>
|
|||
#include <unistd.h>
|
|||
|
|||
#include <mutex>
|
|||
#include <stack>
|
|||
|
|||
|
|||
static std::size_t g_BUFSIZE = (1024 * 1024 * 2); |
|||
|
|||
static std::mutex g_MUTEX; |
|||
static std::stack<fuse_msgbuf_t*> g_MSGBUF_STACK; |
|||
|
|||
static |
|||
__attribute__((destructor)) |
|||
void |
|||
msgbuf_destroy() |
|||
{ |
|||
// TODO: cleanup?
|
|||
} |
|||
|
|||
std::size_t |
|||
msgbuf_bufsize() |
|||
{ |
|||
return g_BUFSIZE; |
|||
} |
|||
|
|||
void |
|||
msgbuf_bufsize(const std::size_t size_) |
|||
{ |
|||
g_BUFSIZE = size_; |
|||
} |
|||
|
|||
fuse_msgbuf_t* |
|||
msgbuf_alloc() |
|||
{ |
|||
int rv; |
|||
fuse_msgbuf_t *msgbuf; |
|||
|
|||
g_MUTEX.lock(); |
|||
if(g_MSGBUF_STACK.empty()) |
|||
{ |
|||
g_MUTEX.unlock(); |
|||
|
|||
msgbuf = (fuse_msgbuf_t*)malloc(sizeof(fuse_msgbuf_t)); |
|||
if(msgbuf == NULL) |
|||
return NULL; |
|||
|
|||
rv = pipe(msgbuf->pipefd); |
|||
assert(rv == 0); |
|||
rv = fcntl(msgbuf->pipefd[0],F_SETPIPE_SZ,g_BUFSIZE); |
|||
assert(rv > 0); |
|||
msgbuf->mem = (char*)malloc(rv); |
|||
msgbuf->size = rv; |
|||
msgbuf->used = 0; |
|||
} |
|||
else |
|||
{ |
|||
msgbuf = g_MSGBUF_STACK.top(); |
|||
g_MSGBUF_STACK.pop(); |
|||
g_MUTEX.unlock(); |
|||
} |
|||
|
|||
return msgbuf; |
|||
} |
|||
|
|||
fuse_msgbuf_t* |
|||
msgbuf_alloc_memonly() |
|||
{ |
|||
fuse_msgbuf_t *msgbuf; |
|||
|
|||
g_MUTEX.lock(); |
|||
if(g_MSGBUF_STACK.empty()) |
|||
{ |
|||
g_MUTEX.unlock(); |
|||
|
|||
msgbuf = (fuse_msgbuf_t*)malloc(sizeof(fuse_msgbuf_t)); |
|||
if(msgbuf == NULL) |
|||
return NULL; |
|||
|
|||
msgbuf->pipefd[0] = -1; |
|||
msgbuf->pipefd[1] = -1; |
|||
msgbuf->mem = (char*)malloc(g_BUFSIZE); |
|||
msgbuf->size = g_BUFSIZE; |
|||
msgbuf->used = 0; |
|||
} |
|||
else |
|||
{ |
|||
msgbuf = g_MSGBUF_STACK.top(); |
|||
g_MSGBUF_STACK.pop(); |
|||
g_MUTEX.unlock(); |
|||
} |
|||
|
|||
return msgbuf; |
|||
} |
|||
|
|||
void |
|||
msgbuf_free(fuse_msgbuf_t *msgbuf_) |
|||
{ |
|||
std::lock_guard<std::mutex> lck(g_MUTEX); |
|||
|
|||
g_MSGBUF_STACK.push(msgbuf_); |
|||
} |
|||
@ -0,0 +1,29 @@ |
|||
/*
|
|||
ISC License |
|||
|
|||
Copyright (c) 2022, Antonio SJ Musumeci <trapexit@spawn.link> |
|||
|
|||
Permission to use, copy, modify, and/or distribute this software for any |
|||
purpose with or without fee is hereby granted, provided that the above |
|||
copyright notice and this permission notice appear in all copies. |
|||
|
|||
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES |
|||
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF |
|||
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR |
|||
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES |
|||
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN |
|||
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF |
|||
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. |
|||
*/ |
|||
|
|||
#pragma once
|
|||
|
|||
#include "fuse_msgbuf.h"
|
|||
|
|||
void msgbuf_bufsize(const uint32_t size); |
|||
std::size_t msgbuf_bufsize(); |
|||
|
|||
fuse_msgbuf_t* msgbuf_alloc(); |
|||
fuse_msgbuf_t* msgbuf_alloc_memonly(); |
|||
|
|||
void msgbuf_free(fuse_msgbuf_t *msgbuf); |
|||
@ -0,0 +1,112 @@ |
|||
#pragma once
|
|||
|
|||
#include "unbounded_queue.hpp"
|
|||
|
|||
#include <tuple>
|
|||
#include <atomic>
|
|||
#include <vector>
|
|||
#include <thread>
|
|||
#include <memory>
|
|||
#include <future>
|
|||
#include <utility>
|
|||
#include <functional>
|
|||
#include <type_traits>
|
|||
|
|||
|
|||
class ThreadPool |
|||
{ |
|||
public: |
|||
explicit |
|||
ThreadPool(const std::size_t thread_count_ = std::thread::hardware_concurrency()) |
|||
: _queues(thread_count_), |
|||
_count(thread_count_) |
|||
{ |
|||
auto worker = [this](std::size_t i) |
|||
{ |
|||
while(true) |
|||
{ |
|||
Proc f; |
|||
|
|||
for(std::size_t n = 0; n < (_count * K); ++n) |
|||
{ |
|||
if(_queues[(i + n) % _count].try_pop(f)) |
|||
break; |
|||
} |
|||
|
|||
if(!f && !_queues[i].pop(f)) |
|||
break; |
|||
|
|||
f(); |
|||
} |
|||
}; |
|||
|
|||
_threads.reserve(thread_count_); |
|||
for(std::size_t i = 0; i < thread_count_; ++i) |
|||
_threads.emplace_back(worker, i); |
|||
} |
|||
|
|||
~ThreadPool() |
|||
{ |
|||
for(auto& queue : _queues) |
|||
queue.unblock(); |
|||
for(auto& thread : _threads) |
|||
thread.join(); |
|||
} |
|||
|
|||
template<typename F> |
|||
void |
|||
enqueue_work(F&& f_) |
|||
{ |
|||
auto i = _index++; |
|||
|
|||
for(std::size_t n = 0; n < (_count * K); ++n) |
|||
{ |
|||
if(_queues[(i + n) % _count].try_push(f_)) |
|||
return; |
|||
} |
|||
|
|||
_queues[i % _count].push(std::move(f_)); |
|||
} |
|||
|
|||
template<typename F> |
|||
[[nodiscard]] |
|||
std::future<typename std::result_of<F()>::type> |
|||
enqueue_task(F&& f_) |
|||
{ |
|||
using TaskReturnType = typename std::result_of<F()>::type; |
|||
using Promise = std::promise<TaskReturnType>; |
|||
|
|||
auto i = _index++; |
|||
auto promise = std::make_shared<Promise>(); |
|||
auto future = promise->get_future(); |
|||
auto work = [=]() { |
|||
auto rv = f_(); |
|||
promise->set_value(rv); |
|||
}; |
|||
|
|||
for(std::size_t n = 0; n < (_count * K); ++n) |
|||
{ |
|||
if(_queues[(i + n) % _count].try_push(work)) |
|||
return future; |
|||
} |
|||
|
|||
_queues[i % _count].push(std::move(work)); |
|||
|
|||
return future; |
|||
} |
|||
|
|||
private: |
|||
using Proc = std::function<void(void)>; |
|||
using Queue = UnboundedQueue<Proc>; |
|||
using Queues = std::vector<Queue>; |
|||
Queues _queues; |
|||
|
|||
private: |
|||
std::vector<std::thread> _threads; |
|||
|
|||
private: |
|||
const std::size_t _count; |
|||
std::atomic_uint _index; |
|||
|
|||
static const unsigned int K = 2; |
|||
}; |
|||
@ -0,0 +1,161 @@ |
|||
#pragma once
|
|||
|
|||
#include <condition_variable>
|
|||
#include <mutex>
|
|||
#include <queue>
|
|||
#include <utility>
|
|||
|
|||
|
|||
template<typename T> |
|||
class UnboundedQueue |
|||
{ |
|||
public: |
|||
explicit |
|||
UnboundedQueue(bool block_ = true) |
|||
: _block(block_) |
|||
{ |
|||
} |
|||
|
|||
void |
|||
push(const T& item_) |
|||
{ |
|||
{ |
|||
std::lock_guard<std::mutex> guard(_queue_lock); |
|||
_queue.push(item_); |
|||
} |
|||
_condition.notify_one(); |
|||
} |
|||
|
|||
void |
|||
push(T&& item_) |
|||
{ |
|||
{ |
|||
std::lock_guard<std::mutex> guard(_queue_lock); |
|||
_queue.push(std::move(item_)); |
|||
} |
|||
|
|||
_condition.notify_one(); |
|||
} |
|||
|
|||
template<typename... Args> |
|||
void |
|||
emplace(Args&&... args_) |
|||
{ |
|||
{ |
|||
std::lock_guard<std::mutex> guard(_queue_lock); |
|||
_queue.emplace(std::forward<Args>(args_)...); |
|||
} |
|||
|
|||
_condition.notify_one(); |
|||
} |
|||
|
|||
bool |
|||
try_push(const T& item_) |
|||
{ |
|||
{ |
|||
std::unique_lock<std::mutex> lock(_queue_lock, std::try_to_lock); |
|||
if(!lock) |
|||
return false; |
|||
_queue.push(item_); |
|||
} |
|||
|
|||
_condition.notify_one(); |
|||
|
|||
return true; |
|||
} |
|||
|
|||
bool |
|||
try_push(T&& item_) |
|||
{ |
|||
{ |
|||
std::unique_lock<std::mutex> lock(_queue_lock, std::try_to_lock); |
|||
if(!lock) |
|||
return false; |
|||
_queue.push(std::move(item_)); |
|||
} |
|||
|
|||
_condition.notify_one(); |
|||
|
|||
return true; |
|||
} |
|||
|
|||
//TODO: push multiple T at once
|
|||
|
|||
bool |
|||
pop(T& item_) |
|||
{ |
|||
std::unique_lock<std::mutex> guard(_queue_lock); |
|||
|
|||
_condition.wait(guard, [&]() { return !_queue.empty() || !_block; }); |
|||
if(_queue.empty()) |
|||
return false; |
|||
|
|||
item_ = std::move(_queue.front()); |
|||
_queue.pop(); |
|||
|
|||
return true; |
|||
} |
|||
|
|||
bool |
|||
try_pop(T& item_) |
|||
{ |
|||
std::unique_lock<std::mutex> lock(_queue_lock, std::try_to_lock); |
|||
if(!lock || _queue.empty()) |
|||
return false; |
|||
|
|||
item_ = std::move(_queue.front()); |
|||
_queue.pop(); |
|||
|
|||
return true; |
|||
} |
|||
|
|||
std::size_t |
|||
size() const |
|||
{ |
|||
std::lock_guard<std::mutex> guard(_queue_lock); |
|||
|
|||
return _queue.size(); |
|||
} |
|||
|
|||
bool |
|||
empty() const |
|||
{ |
|||
std::lock_guard<std::mutex> guard(_queue_lock); |
|||
|
|||
return _queue.empty(); |
|||
} |
|||
|
|||
void |
|||
block() |
|||
{ |
|||
std::lock_guard<std::mutex> guard(_queue_lock); |
|||
_block = true; |
|||
} |
|||
|
|||
void |
|||
unblock() |
|||
{ |
|||
{ |
|||
std::lock_guard<std::mutex> guard(_queue_lock); |
|||
_block = false; |
|||
} |
|||
|
|||
_condition.notify_all(); |
|||
} |
|||
|
|||
bool |
|||
blocking() const |
|||
{ |
|||
std::lock_guard<std::mutex> guard(_queue_lock); |
|||
|
|||
return _block; |
|||
} |
|||
|
|||
private: |
|||
mutable std::mutex _queue_lock; |
|||
|
|||
private: |
|||
bool _block; |
|||
std::queue<T> _queue; |
|||
std::condition_variable _condition; |
|||
}; |
|||
@ -0,0 +1,14 @@ |
|||
#!/bin/sh |
|||
set -x |
|||
|
|||
BASEPATH="/tmp" |
|||
|
|||
for x in $(seq -w 2) |
|||
do |
|||
FILEPATH="${BASEPATH}/mergerfs-${x}.img" |
|||
MOUNTPOINT="${BASEPATH}/mergerfs-${x}" |
|||
truncate -s 1G "${FILEPATH}" |
|||
mkdir -p "${MOUNTPOINT}" |
|||
mkfs.ext4 -m0 -L "mergerfs${x}" "${FILEPATH}" |
|||
sudo mount "${FILEPATH}" "${MOUNTPOINT}" |
|||
done |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue