diff --git a/thread/ThreadPool.cc b/thread/ThreadPool.cc index dcaa19f..4e663a3 100644 --- a/thread/ThreadPool.cc +++ b/thread/ThreadPool.cc @@ -47,11 +47,17 @@ void ThreadPool::start(int numThreads) void ThreadPool::stop() { - running_ = false; + printf("starting to stop the pool.\n"); + { + MutexLockGuard lock(mutex_); + running_ = false; + } cond_.notifyAll(); for_each(threads_.begin(), threads_.end(), boost::bind(&muduo::Thread::join, _1)); + threads_.clear(); + printf("stop the pool successfully.\n"); } void ThreadPool::run(const Task& task) @@ -59,13 +65,13 @@ void ThreadPool::run(const Task& task) if (threads_.empty()) { task(); + return; } - else { MutexLockGuard lock(mutex_); queue_.push_back(task); - cond_.notify(); } + cond_.notify(); } ThreadPool::Task ThreadPool::take() @@ -117,3 +123,33 @@ void ThreadPool::runInThread() } } + + +// // bind cores +// const int targetCore = bindcores_[i]; +// cpu_set_t cpuset; +// CPU_ZERO(&cpuset); +// CPU_SET(targetCore, &cpuset); +// int ret = pthread_setaffinity_np(thread_[i], sizeof(cpu_set_t), &cpuset); // or std::thread::native_handle() +// // or pthread_self() +// if (ret != 0) +// { +// printf("[ThreadPool] thread %zu bind core %d failed\n", i, targetCore); +// } +// else +// { +// printf("[ThreadPool] thread %zu bind core %d success\n", i, targetCore); +// } + +// // 线程优先级 +// sched_param sp; +// sp.sched_priority = sched_get_priority_max(SCHED_FIFO); +// ret = pthread_setschedparam(thread_[i], SCHED_FIFO, &sp); +// if (ret != 0) +// { +// printf("[ThreadPool] thread %zu set priority %d failed\n", i, sp.sched_priority); +// } +// else +// { +// printf("[ThreadPool] thread %zu set priority %d success\n", i, sp.sched_priority); +// } diff --git a/thread/ThreadPool.h b/thread/ThreadPool.h index 042caa4..c322f19 100644 --- a/thread/ThreadPool.h +++ b/thread/ThreadPool.h @@ -17,6 +17,7 @@ #include #include +#include namespace muduo { @@ -34,6 +35,26 @@ class ThreadPool : boost::noncopyable void run(const Task& f); + template + inline auto run_future(Func&& func, Args&&... args) -> std::future::type> + { + if (threads_.empty()) { + func(args...); + } + using ret_type = typename std::result_of::type; + auto task = std::make_shared>(std::bind(std::forward(func), std::forward(args)...)); + auto ret = task->get_future(); + do { + MutexLockGuard lock(mutex_); + if (!running_) { + throw std::runtime_error("enqueue on stopped ThreadPool"); + } + queue_.push_back([task]() { (*task)(); }); + } while (0); + cond_.notify(); + return std::move(ret); + } + private: void runInThread(); Task take(); diff --git a/thread/test/Factory_cpp20.cc b/thread/test/Factory_cpp20.cc new file mode 100644 index 0000000..19e3ae8 --- /dev/null +++ b/thread/test/Factory_cpp20.cc @@ -0,0 +1,276 @@ +#include + +#include "../Mutex.h" + +#include +#include + +#include + +using std::string; + +class Stock : boost::noncopyable +{ + public: + Stock(const string& name) + : name_(name) + { + printf(" Stock[%p] %s\n", this, name_.c_str()); + } + + ~Stock() + { + printf("~Stock[%p] %s\n", this, name_.c_str()); + } + + const string& key() const { return name_; } + + private: + string name_; +}; + +namespace version1 +{ + +// questionable code +class StockFactory : boost::noncopyable +{ + public: + + std::shared_ptr get(const string& key) + { + muduo::MutexLockGuard lock(mutex_); + std::shared_ptr& pStock = stocks_[key]; + if (!pStock) + { + pStock.reset(new Stock(key)); + } + return pStock; + } + + + private: + mutable muduo::MutexLock mutex_; + std::map > stocks_; +}; + +} + +namespace version2 +{ + +class StockFactory : boost::noncopyable +{ + public: + std::shared_ptr get(const string& key) + { + std::shared_ptr pStock; + muduo::MutexLockGuard lock(mutex_); + std::weak_ptr& wkStock = stocks_[key]; + pStock = wkStock.lock(); + if (!pStock) + { + pStock.reset(new Stock(key)); + wkStock = pStock; + } + return pStock; + } + + private: + mutable muduo::MutexLock mutex_; + std::map > stocks_; +}; + +} + +namespace version3 +{ + +class StockFactory : boost::noncopyable +{ + public: + + std::shared_ptr get(const string& key) + { + std::shared_ptr pStock; + muduo::MutexLockGuard lock(mutex_); + std::weak_ptr& wkStock = stocks_[key]; + pStock = wkStock.lock(); + if (!pStock) + { + pStock.reset(new Stock(key), std::bind_front(&StockFactory::deleteStock, this)); + wkStock = pStock; + } + return pStock; + } + + private: + + void deleteStock(Stock* stock) + { + printf("deleteStock[%p]\n", stock); + if (stock) + { + muduo::MutexLockGuard lock(mutex_); + stocks_.erase(stock->key()); // This is wrong, see removeStock below for correct implementation. + } + delete stock; // sorry, I lied + } + mutable muduo::MutexLock mutex_; + std::map > stocks_; +}; + +} + +namespace version4 +{ + +class StockFactory : public std::enable_shared_from_this, + boost::noncopyable +{ + public: + + std::shared_ptr get(const string& key) + { + std::shared_ptr pStock; + muduo::MutexLockGuard lock(mutex_); + std::weak_ptr& wkStock = stocks_[key]; + pStock = wkStock.lock(); + if (!pStock) + { + pStock.reset(new Stock(key), + std::bind_front(&StockFactory::deleteStock, shared_from_this())); + wkStock = pStock; + } + return pStock; + } + + private: + + void deleteStock(Stock* stock) + { + printf("deleteStock[%p]\n", stock); + if (stock) + { + muduo::MutexLockGuard lock(mutex_); + stocks_.erase(stock->key()); // This is wrong, see removeStock below for correct implementation. + } + delete stock; // sorry, I lied + } + mutable muduo::MutexLock mutex_; + std::map > stocks_; +}; + +} + +class StockFactory : public std::enable_shared_from_this, + boost::noncopyable +{ + public: + std::shared_ptr get(const string& key) + { + std::shared_ptr pStock; + muduo::MutexLockGuard lock(mutex_); + std::weak_ptr& wkStock = stocks_[key]; + pStock = wkStock.lock(); + if (!pStock) + { + pStock.reset(new Stock(key), + std::bind_front(&StockFactory::weakDeleteCallback, + std::weak_ptr(shared_from_this()))); + wkStock = pStock; + } + return pStock; + } + + private: + static void weakDeleteCallback(const std::weak_ptr& wkFactory, + Stock* stock) + { + printf("weakDeleteStock[%p]\n", stock); + std::shared_ptr factory(wkFactory.lock()); + if (factory) + { + factory->removeStock(stock); + } + else + { + printf("factory died.\n"); + } + delete stock; // sorry, I lied + } + + void removeStock(Stock* stock) + { + if (stock) + { + muduo::MutexLockGuard lock(mutex_); + auto it = stocks_.find(stock->key()); + if (it != stocks_.end() && it->second.expired()) + { + stocks_.erase(stock->key()); + } + } + } + + private: + mutable muduo::MutexLock mutex_; + std::map > stocks_; +}; + +void testLongLifeFactory() +{ + std::shared_ptr factory(new StockFactory); + { + std::shared_ptr stock = factory->get("NYSE:IBM"); + std::shared_ptr stock2 = factory->get("NYSE:IBM"); + assert(stock == stock2); + // stock destructs here + } + // factory destructs here +} + +void testShortLifeFactory() +{ + std::shared_ptr stock; + { + std::shared_ptr factory(new StockFactory); + stock = factory->get("NYSE:IBM"); + std::shared_ptr stock2 = factory->get("NYSE:IBM"); + assert(stock == stock2); + // factory destructs here + } + // stock destructs here +} + +int main() +{ + version1::StockFactory sf1; + version2::StockFactory sf2; + version3::StockFactory sf3; + std::shared_ptr sf4(new version3::StockFactory); + std::shared_ptr sf5(new StockFactory); + + { + std::shared_ptr s1 = sf1.get("stock1"); + } + + { + std::shared_ptr s2 = sf2.get("stock2"); + } + + { + std::shared_ptr s3 = sf3.get("stock3"); + } + + { + std::shared_ptr s4 = sf4->get("stock4"); + } + + { + std::shared_ptr s5 = sf5->get("stock5"); + } + + testLongLifeFactory(); + testShortLifeFactory(); +} diff --git a/thread/test/ThreadPool_test.cc b/thread/test/ThreadPool_test.cc index f8ba48c..a253b93 100644 --- a/thread/test/ThreadPool_test.cc +++ b/thread/test/ThreadPool_test.cc @@ -17,8 +17,9 @@ void printString(const std::string& str) int main() { muduo::ThreadPool pool("MainThreadPool"); - pool.start(5); + + pool.start(5); pool.run(print); pool.run(print); for (int i = 0; i < 100; ++i) @@ -32,5 +33,22 @@ int main() pool.run(boost::bind(&muduo::CountDownLatch::countDown, &latch)); latch.wait(); pool.stop(); + + printf("Testing run_future\n"); + pool.start(5); + pool.run(print); + pool.run(print); + std::vector> futures; + for (int i = 0; i < 100; ++i) + { + char buf[32]; + snprintf(buf, sizeof buf, "task %d", i); + futures.emplace_back(std::move(pool.run_future(boost::bind(printString, std::string(buf))))); + } + for (auto& future : futures) + { + future.get(); + } + pool.stop(); }