Threadpool

threadpoo.cpp

//  Created by Christian Kellner on 22/07/2014.
//  Copyright (c) 2014 G-Node. All rights reserved.
//

#include <iostream>
#include <thread>
#include <vector>
#include <mutex>
#include <utility>

class threadpool {
public:
    
    threadpool(size_t poolsize)
    : fun(nullptr), shutdown(false) {
        for (size_t i = 0; i < poolsize; i++) {
            threads.emplace_back(runloop, std::ref(*this), i);
        }
    }
    
    ~threadpool() {
        {
            //not really neccessary, but whhhyyyy nooooot
            std::unique_lock<std::mutex> lock(guard);
            shutdown = true;
        }
        
        cond.notify_all();
        
        for (auto& t: threads) {
            t.join();
        }
        
        std::cerr << "TP destoryed" << std::endl;
    }
    
    void launch(size_t n, std::function<void(size_t)> &&f) {
        
        //assign function
        {
            std::unique_lock<std::mutex> lock(guard);
            fun = std::forward<std::function<void(size_t)>>(f);
            worksize = n;
            workitem = 0;
        }
        
        cond.notify_all();
        
        {
            std::unique_lock<std::mutex> lock(guard);
            cond.wait(lock, [this]{ return !(workitem < worksize); });
            //when we are arrive here, all work should be done
        }
        
    }
    
private:
    
    static void runloop(threadpool &pool, size_t id) {
        
        std::unique_lock<std::mutex> lock(pool.guard);
        while (1) {
            pool.cond.wait(lock, [&pool]{
                return pool.shutdown || pool.workitem < pool.worksize;
            });
            
            if (pool.shutdown) {
                return;
            }
            
            size_t workid = -1;
            while (pool.workitem < pool.worksize) {
                workid = pool.workitem++;
                lock.unlock();
                pool.fun(workid);
                lock.lock();
            }
            
            //we hold the mutex here
            if (workid + 1 == pool.worksize) {
                pool.cond.notify_one();
            }
        }
    }
    
    std::vector<std::thread> threads;
    std::function<void(size_t)> fun;
    std::mutex guard;
    std::condition_variable cond;
    size_t worksize;
    size_t workitem;
    bool   shutdown;
};

int main(int argc, const char * argv[]) {
    std::mutex lock;
    
    {
        threadpool pool(8);
        std::mutex *lptr = &lock;
        pool.launch(4, [lptr](size_t id) {
            lptr->lock();
            std::cout << "W" << id << std::endl;
            std::cout.flush();
            lptr->unlock();
        });
    }
    
    std::chrono::milliseconds dura( 2000 );
    std::this_thread::sleep_for( dura );
    
    return 0;
}