From e26434979adc33bd949566ba7bf02dff64e80a3e Mon Sep 17 00:00:00 2001 From: Chris Dyer Date: Tue, 2 Oct 2012 00:19:43 -0400 Subject: cdec cleanup, remove bayesian stuff, parsing stuff --- gi/pyp-topics/src/workers.hh | 275 ------------------------------------------- 1 file changed, 275 deletions(-) delete mode 100644 gi/pyp-topics/src/workers.hh (limited to 'gi/pyp-topics/src/workers.hh') diff --git a/gi/pyp-topics/src/workers.hh b/gi/pyp-topics/src/workers.hh deleted file mode 100644 index 95b18947..00000000 --- a/gi/pyp-topics/src/workers.hh +++ /dev/null @@ -1,275 +0,0 @@ -/** - Basic thread-pool tools using Boost.Thread. - (Jan Botha, 7/2010) - - --Simple usage-- - Use SimpleWorker. - Example, call a function that returns an int in a new thread: - typedef boost::function JobType; - JobType job = boost::bind(funcname); - //or boost::bind(&class::funcname, this) for a member function - SimpleWorker worker(job); - int result = worker.getResult(); //blocks until result is ready - - --Extended usage-- - Use WorkerPool, which uses Queuemt (a synchronized queue) and Worker. - Example: - (same context and typedef - WorkerPool pool(num_threads); - JobType job = ... - pool.addJob(job); - ... - pool.get_result(); //blocks until all workers are done, returns the some of their results. - - Jobs added to a WorkerPool need to be the same type. A WorkerPool instance should not be reused (e.g. adding jobs) after calling get_result(). -*/ - -#ifndef WORKERS_HH -#define WORKERS_HH - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include -#include "timing.h" - -/** Implements a synchronized queue*/ -template -class Queuemt -{ - -public: - boost::condition_variable_any cond; - const bool& running; - - Queuemt() { } - Queuemt(const bool& running) : running(running), maxsize(0), qsize(0) - { - } - - ~Queuemt() { - } - - J pop() - { - J job; - { - boost::unique_lock qlock(q_mutex); - while (running && qsize == 0) - cond.wait(qlock); - - if (qsize > 0) - { - job = q.front(); - q.pop(); - --qsize; - } - } - if (job) - cond.notify_one(); - return job; - - } - - void push(J job) - { - { - boost::unique_lock lock(q_mutex); - q.push(job); - ++qsize; - } - if (qsize > maxsize) - maxsize = qsize; - - cond.notify_one(); - } - - int getMaxsize() - { - return maxsize; - } - - int size() - { - return qsize; - } - -private: - boost::shared_mutex q_mutex; - std::queue q; - int maxsize; - volatile int qsize; -}; - - -template -class Worker -{ -typedef boost::packaged_task PackagedTask; -public: - Worker(Queuemt& queue, int id, int num_workers) : - q(queue), tasktime(0.0), id(id), num_workers(num_workers) - { - PackagedTask task(boost::bind(&Worker::run, this)); - future = task.get_future(); - boost::thread t(boost::move(task)); - } - - R run() //this is called upon thread creation - { - R wresult = 0; - while (isRunning()) - { - J job = q.pop(); - - if (job) - { - timer.Reset(); - wresult += job(); - tasktime += timer.Elapsed(); - } - } - return wresult; - } - - R getResult() - { - if (!future.is_ready()) - future.wait(); - assert(future.is_ready()); - return future.get(); - } - - double getTaskTime() - { - return tasktime; - } - -private: - - Queuemt& q; - - boost::unique_future future; - - bool isRunning() - { - return q.running || q.size() > 0; - } - - Timer timer; - double tasktime; - int id; - int num_workers; -}; - -template -class WorkerPool -{ -typedef boost::packaged_task PackagedTask; -typedef Worker WJR; -typedef boost::ptr_vector WorkerVector; -public: - - WorkerPool(int num_workers) - { - q.reset(new Queuemt(running)); - running = true; - for (int i = 0; i < num_workers; ++i) - workers.push_back( new Worker(*q, i, num_workers) ); - } - - ~WorkerPool() - { - } - - R get_result() - { - running = false; - q->cond.notify_all(); - R tmp = 0; - double tasktime = 0.0; - for (typename WorkerVector::iterator it = workers.begin(); it != workers.end(); it++) - { - R res = it->getResult(); - tmp += res; - //std::cerr << "tasktime: " << it->getTaskTime() << std::endl; - tasktime += it->getTaskTime(); - } -// std::cerr << " maxQ = " << q->getMaxsize() << std::endl; - return tmp; - } - - void addJob(J job) - { - q->push(job); - } - -private: - - WorkerVector workers; - - boost::shared_ptr > q; - - bool running; -}; - -/////////////////// -template -class SimpleWorker -{ -typedef boost::packaged_task PackagedTask; -public: - SimpleWorker(J& job) : job(job), tasktime(0.0) - { - PackagedTask task(boost::bind(&SimpleWorker::run, this)); - future = task.get_future(); - boost::thread t(boost::move(task)); - } - - R run() //this is called upon thread creation - { - R wresult = 0; - - assert(job); - timer.Reset(); - wresult = job(); - tasktime = timer.Elapsed(); - std::cerr << tasktime << " s" << std::endl; - return wresult; - } - - R getResult() - { - if (!future.is_ready()) - future.wait(); - assert(future.is_ready()); - return future.get(); - } - - double getTaskTime() - { - return tasktime; - } - -private: - - J job; - - boost::unique_future future; - - Timer timer; - double tasktime; - -}; - - - -#endif -- cgit v1.2.3