mirror of https://github.com/trapexit/mergerfs.git
Antonio SJ Musumeci
10 months ago
22 changed files with 1076 additions and 171 deletions
-
49README.md
-
86man/mergerfs.1
-
4src/config.cpp
-
6src/config.hpp
-
13src/fs_devid.hpp
-
4src/fs_dirfd.hpp
-
13src/fs_inode.cpp
-
6src/fs_inode.hpp
-
75src/fuse_readdir.cpp
-
33src/fuse_readdir.hpp
-
34src/fuse_readdir_base.hpp
-
194src/fuse_readdir_cor.cpp
-
48src/fuse_readdir_cor.hpp
-
173src/fuse_readdir_cosr.cpp
-
39src/fuse_readdir_cosr.hpp
-
75src/fuse_readdir_factory.cpp
-
18src/fuse_readdir_factory.hpp
-
30src/fuse_readdir_plus.cpp
-
43src/fuse_readdir_seq.cpp
-
19src/fuse_readdir_seq.hpp
-
161src/unbounded_queue.hpp
-
124src/unbounded_thread_pool.hpp
@ -0,0 +1,34 @@ |
|||
/*
|
|||
Copyright (c) 2023, 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.h"
|
|||
|
|||
|
|||
namespace FUSE |
|||
{ |
|||
class ReadDirBase |
|||
{ |
|||
public: |
|||
ReadDirBase() {}; |
|||
virtual ~ReadDirBase() {}; |
|||
|
|||
public: |
|||
virtual int operator()(fuse_file_info_t const *ffi, |
|||
fuse_dirents_t *buf) = 0; |
|||
}; |
|||
} |
|||
@ -0,0 +1,194 @@ |
|||
/*
|
|||
Copyright (c) 2023, 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_readdir_cor.hpp"
|
|||
|
|||
#include "config.hpp"
|
|||
#include "dirinfo.hpp"
|
|||
#include "errno.hpp"
|
|||
#include "fs_closedir.hpp"
|
|||
#include "fs_devid.hpp"
|
|||
#include "fs_inode.hpp"
|
|||
#include "fs_opendir.hpp"
|
|||
#include "fs_path.hpp"
|
|||
#include "fs_readdir.hpp"
|
|||
#include "hashset.hpp"
|
|||
#include "ugid.hpp"
|
|||
|
|||
#include "fuse_dirents.h"
|
|||
|
|||
|
|||
FUSE::ReadDirCOR::ReadDirCOR(unsigned concurrency_) |
|||
: _tp(concurrency_) |
|||
{ |
|||
|
|||
} |
|||
|
|||
FUSE::ReadDirCOR::~ReadDirCOR() |
|||
{ |
|||
|
|||
} |
|||
|
|||
namespace l |
|||
{ |
|||
static |
|||
inline |
|||
uint64_t |
|||
dirent_exact_namelen(const struct dirent *d_) |
|||
{ |
|||
#ifdef _D_EXACT_NAMLEN
|
|||
return _D_EXACT_NAMLEN(d_); |
|||
#elif defined _DIRENT_HAVE_D_NAMLEN
|
|||
return d_->d_namlen; |
|||
#else
|
|||
return strlen(d_->d_name); |
|||
#endif
|
|||
} |
|||
|
|||
static |
|||
inline |
|||
int |
|||
readdir(std::string basepath_, |
|||
HashSet &names_, |
|||
std::mutex &names_mutex_, |
|||
fuse_dirents_t *buf_, |
|||
std::mutex &dirents_mutex_) |
|||
{ |
|||
int rv; |
|||
int err; |
|||
DIR *dh; |
|||
dev_t dev; |
|||
std::string filepath; |
|||
|
|||
dh = fs::opendir(basepath_); |
|||
if(dh == NULL) |
|||
return -errno; |
|||
|
|||
dev = fs::devid(dh); |
|||
|
|||
rv = 0; |
|||
err = 0; |
|||
for(struct dirent *de = fs::readdir(dh); de && !rv; de = fs::readdir(dh)) |
|||
{ |
|||
std::uint64_t namelen; |
|||
|
|||
namelen = l::dirent_exact_namelen(de); |
|||
|
|||
{ |
|||
std::lock_guard<std::mutex> lk(names_mutex_); |
|||
rv = names_.put(de->d_name,namelen); |
|||
if(rv == 0) |
|||
continue; |
|||
} |
|||
|
|||
filepath = fs::path::make(basepath_,de->d_name); |
|||
de->d_ino = fs::inode::calc(filepath, |
|||
DTTOIF(de->d_type), |
|||
dev, |
|||
de->d_ino); |
|||
|
|||
{ |
|||
std::lock_guard<std::mutex> lk(dirents_mutex_); |
|||
rv = fuse_dirents_add(buf_,de,namelen); |
|||
if(rv == 0) |
|||
continue; |
|||
} |
|||
|
|||
err = -ENOMEM; |
|||
} |
|||
|
|||
fs::closedir(dh); |
|||
|
|||
return err; |
|||
} |
|||
|
|||
static |
|||
std::vector<int> |
|||
concurrent_readdir(ThreadPool &tp_, |
|||
const Branches::CPtr &branches_, |
|||
const char *dirname_, |
|||
fuse_dirents_t *buf_) |
|||
{ |
|||
HashSet names; |
|||
std::mutex names_mutex; |
|||
std::mutex dirents_mutex; |
|||
std::vector<int> rv; |
|||
std::vector<std::future<int>> futures; |
|||
|
|||
for(auto const &branch : *branches_) |
|||
{ |
|||
auto func = [&]() |
|||
{ |
|||
std::string basepath = fs::path::make(branch.path,dirname_); |
|||
|
|||
return l::readdir(basepath,names,names_mutex,buf_,dirents_mutex); |
|||
}; |
|||
|
|||
auto rv = tp_.enqueue_task(func); |
|||
|
|||
futures.emplace_back(std::move(rv)); |
|||
} |
|||
|
|||
for(auto &future : futures) |
|||
rv.push_back(future.get()); |
|||
|
|||
return rv; |
|||
} |
|||
|
|||
static |
|||
int |
|||
calc_rv(std::vector<int> rvs_) |
|||
{ |
|||
for(auto rv : rvs_) |
|||
{ |
|||
if(rv == 0) |
|||
return 0; |
|||
} |
|||
|
|||
if(rvs_.empty()) |
|||
return -ENOENT; |
|||
|
|||
return rvs_[0]; |
|||
} |
|||
|
|||
static |
|||
int |
|||
readdir(ThreadPool &tp_, |
|||
const Branches::CPtr &branches_, |
|||
const char *dirname_, |
|||
fuse_dirents_t *buf_) |
|||
{ |
|||
std::vector<int> rvs; |
|||
|
|||
fuse_dirents_reset(buf_); |
|||
|
|||
rvs = l::concurrent_readdir(tp_,branches_,dirname_,buf_); |
|||
|
|||
return l::calc_rv(rvs); |
|||
} |
|||
} |
|||
|
|||
int |
|||
FUSE::ReadDirCOR::operator()(fuse_file_info_t const *ffi_, |
|||
fuse_dirents_t *buf_) |
|||
{ |
|||
Config::Read cfg; |
|||
DirInfo *di = reinterpret_cast<DirInfo*>(ffi_->fh); |
|||
const fuse_context *fc = fuse_get_context(); |
|||
const ugid::Set ugid(fc->uid,fc->gid); |
|||
|
|||
return l::readdir(_tp,cfg->branches,di->fusepath.c_str(),buf_); |
|||
} |
|||
@ -0,0 +1,173 @@ |
|||
/*
|
|||
Copyright (c) 2023, 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_readdir_cosr.hpp"
|
|||
|
|||
#include "config.hpp"
|
|||
#include "dirinfo.hpp"
|
|||
#include "errno.hpp"
|
|||
#include "fs_closedir.hpp"
|
|||
#include "fs_devid.hpp"
|
|||
#include "fs_dirfd.hpp"
|
|||
#include "fs_inode.hpp"
|
|||
#include "fs_opendir.hpp"
|
|||
#include "fs_path.hpp"
|
|||
#include "fs_readdir.hpp"
|
|||
#include "fs_stat.hpp"
|
|||
#include "hashset.hpp"
|
|||
#include "ugid.hpp"
|
|||
|
|||
#include "fuse_dirents.h"
|
|||
|
|||
|
|||
FUSE::ReadDirCOSR::ReadDirCOSR(unsigned concurrency_) |
|||
: _tp(concurrency_) |
|||
{ |
|||
|
|||
} |
|||
|
|||
FUSE::ReadDirCOSR::~ReadDirCOSR() |
|||
{ |
|||
|
|||
} |
|||
|
|||
namespace l |
|||
{ |
|||
static |
|||
inline |
|||
uint64_t |
|||
dirent_exact_namelen(const struct dirent *d_) |
|||
{ |
|||
#ifdef _D_EXACT_NAMLEN
|
|||
return _D_EXACT_NAMLEN(d_); |
|||
#elif defined _DIRENT_HAVE_D_NAMLEN
|
|||
return d_->d_namlen; |
|||
#else
|
|||
return strlen(d_->d_name); |
|||
#endif
|
|||
} |
|||
|
|||
static |
|||
inline |
|||
std::vector<std::future<DIR*>> |
|||
opendir(ThreadPool &tp_, |
|||
const Branches::CPtr &branches_, |
|||
char const *dirname_) |
|||
{ |
|||
std::vector<std::future<DIR*>> futures; |
|||
|
|||
for(auto const &branch : *branches_) |
|||
{ |
|||
auto func = [&branch,dirname_]() |
|||
{ |
|||
std::string basepath = fs::path::make(branch.path,dirname_); |
|||
|
|||
return fs::opendir(basepath); |
|||
}; |
|||
|
|||
auto rv = tp_.enqueue_task(func); |
|||
|
|||
futures.emplace_back(std::move(rv)); |
|||
} |
|||
|
|||
return futures; |
|||
} |
|||
|
|||
static |
|||
inline |
|||
int |
|||
readdir(std::vector<std::future<DIR*>> &dh_futures_, |
|||
char const *dirname_, |
|||
fuse_dirents_t *buf_) |
|||
{ |
|||
int err; |
|||
HashSet names; |
|||
std::string fullpath; |
|||
|
|||
err = 0; |
|||
for(auto &dh_future : dh_futures_) |
|||
{ |
|||
int rv; |
|||
DIR *dh; |
|||
dev_t dev; |
|||
|
|||
dh = dh_future.get(); |
|||
if(dh == NULL) |
|||
continue; |
|||
|
|||
dev = fs::devid(dh); |
|||
|
|||
rv = 0; |
|||
for(struct dirent *de = fs::readdir(dh); de && !rv; de = fs::readdir(dh)) |
|||
{ |
|||
std::uint64_t namelen; |
|||
|
|||
namelen = l::dirent_exact_namelen(de); |
|||
|
|||
rv = names.put(de->d_name,namelen); |
|||
if(rv == 0) |
|||
continue; |
|||
|
|||
fullpath = fs::path::make(dirname_,de->d_name); |
|||
de->d_ino = fs::inode::calc(fullpath, |
|||
DTTOIF(de->d_type), |
|||
dev, |
|||
de->d_ino); |
|||
|
|||
rv = fuse_dirents_add(buf_,de,namelen); |
|||
if(rv == 0) |
|||
continue; |
|||
|
|||
err = -ENOMEM; |
|||
} |
|||
|
|||
fs::closedir(dh); |
|||
} |
|||
|
|||
return err; |
|||
} |
|||
|
|||
static |
|||
inline |
|||
int |
|||
readdir(ThreadPool &tp_, |
|||
const Branches::CPtr &branches_, |
|||
const char *dirname_, |
|||
fuse_dirents_t *buf_) |
|||
{ |
|||
int rv; |
|||
std::vector<std::future<DIR*>> dh_futures; |
|||
|
|||
fuse_dirents_reset(buf_); |
|||
|
|||
dh_futures = l::opendir(tp_,branches_,dirname_); |
|||
rv = l::readdir(dh_futures,dirname_,buf_); |
|||
|
|||
return rv; |
|||
} |
|||
} |
|||
|
|||
int |
|||
FUSE::ReadDirCOSR::operator()(fuse_file_info_t const *ffi_, |
|||
fuse_dirents_t *buf_) |
|||
{ |
|||
Config::Read cfg; |
|||
DirInfo *di = reinterpret_cast<DirInfo*>(ffi_->fh); |
|||
const fuse_context *fc = fuse_get_context(); |
|||
const ugid::Set ugid(fc->uid,fc->gid); |
|||
|
|||
return l::readdir(_tp,cfg->branches,di->fusepath.c_str(),buf_); |
|||
} |
|||
@ -0,0 +1,39 @@ |
|||
/*
|
|||
ISC License |
|||
|
|||
Copyright (c) 2023, 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_readdir_base.hpp"
|
|||
#include "unbounded_thread_pool.hpp"
|
|||
|
|||
// concurrent open, sequential read
|
|||
namespace FUSE |
|||
{ |
|||
class ReadDirCOSR final : public FUSE::ReadDirBase |
|||
{ |
|||
public: |
|||
ReadDirCOSR(unsigned concurrency); |
|||
~ReadDirCOSR(); |
|||
|
|||
int operator()(fuse_file_info_t const *ffi, |
|||
fuse_dirents_t *buf); |
|||
|
|||
private: |
|||
ThreadPool _tp; |
|||
}; |
|||
} |
|||
@ -0,0 +1,75 @@ |
|||
/*
|
|||
ISC License |
|||
|
|||
Copyright (c) 2023, 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_readdir_factory.hpp"
|
|||
|
|||
#include "fuse_readdir_cor.hpp"
|
|||
#include "fuse_readdir_cosr.hpp"
|
|||
#include "fuse_readdir_seq.hpp"
|
|||
|
|||
#include <cassert>
|
|||
#include <cmath>
|
|||
#include <cstdio>
|
|||
#include <cstdlib>
|
|||
|
|||
|
|||
namespace l |
|||
{ |
|||
static |
|||
void |
|||
read_cfg(std::string const cfgstr_, |
|||
std::string &type_, |
|||
int &concurrency_) |
|||
{ |
|||
char type[16]; |
|||
int concurrency; |
|||
|
|||
concurrency = 0; |
|||
std::sscanf(cfgstr_.c_str(),"%15[a-z]:%u",type,&concurrency); |
|||
|
|||
if(concurrency == 0) |
|||
concurrency = std::thread::hardware_concurrency(); |
|||
else if(concurrency < 0) |
|||
concurrency = (std::thread::hardware_concurrency() / std::abs(concurrency)); |
|||
|
|||
if(concurrency == 0) |
|||
concurrency = 1; |
|||
|
|||
type_ = type; |
|||
concurrency_ = concurrency; |
|||
} |
|||
} |
|||
|
|||
std::shared_ptr<FUSE::ReadDirBase> |
|||
FUSE::ReadDirFactory::make(std::string const cfgstr_) |
|||
{ |
|||
int concurrency; |
|||
std::string type; |
|||
|
|||
l::read_cfg(cfgstr_,type,concurrency); |
|||
assert(concurrency); |
|||
|
|||
if(type == "seq") |
|||
return std::make_shared<FUSE::ReadDirSeq>(); |
|||
if(type == "cosr") |
|||
return std::make_shared<FUSE::ReadDirCOSR>(concurrency); |
|||
if(type == "cor") |
|||
return std::make_shared<FUSE::ReadDirCOR>(concurrency); |
|||
|
|||
return {}; |
|||
} |
|||
@ -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,124 @@ |
|||
#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; |
|||
} |
|||
|
|||
public: |
|||
std::vector<pthread_t> |
|||
threads() |
|||
{ |
|||
std::vector<pthread_t> rv; |
|||
|
|||
for(auto &thread : _threads) |
|||
rv.push_back(thread.native_handle()); |
|||
|
|||
return rv; |
|||
} |
|||
|
|||
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; |
|||
}; |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue