Initial commit

include/pipedispenser.cuh

@@ -0,0 +1,139 @@
+/*
+ * pipedispenser.cuh
+ *
+ *  Created on: 2013-03-01
+ *      Author: spoon
+ */
+
+#ifndef PIPEDISPENSER_CUH_
+#define PIPEDISPENSER_CUH_
+
+#include <pthread.h>
+#include <set>
+#include <algorithm>
+#include <iterator>
+#include <util.cuh>
+
+class PipeDispenser {
+protected:
+    int _numPipes;
+    seti _pipes;
+    pthread_mutex_t *_mutex;
+    void lock() {
+        pthread_mutex_lock(_mutex);
+    }
+
+    void unlock() {
+        pthread_mutex_unlock(_mutex);
+    }
+public:
+    PipeDispenser(const seti& pipes) {
+        _pipes.insert(pipes.begin(), pipes.end());
+        _mutex = (pthread_mutex_t*)(malloc(sizeof (pthread_mutex_t)));
+        pthread_mutex_init(_mutex, NULL);
+    }
+
+    virtual ~PipeDispenser() {
+        pthread_mutex_destroy(_mutex);
+        free(_mutex);
+    }
+
+    virtual int getPipe(const seti& interested) = 0;
+    int getPipe(int interested) {
+        seti tmp;
+        tmp.insert(interested);
+        return getPipe(tmp);
+    }
+    virtual void freePipe(int pipe) = 0;
+};
+
+/*
+ * This one blocks until there is a free pipe to return.
+ */
+class PipeDispenserBlocking : public PipeDispenser {
+protected:
+    pthread_cond_t *_cv;
+
+    void wait() {
+        pthread_cond_wait(_cv, _mutex);
+    }
+
+    void broadcast() {
+        pthread_cond_broadcast(_cv);
+    }
+
+    int getAvailablePipes(const seti& interested, intv& available) {
+        available.clear();
+        std::set_intersection(_pipes.begin(), _pipes.end(), interested.begin(), interested.end(), std::back_inserter(available));
+        return available.size();
+    }
+public:
+    PipeDispenserBlocking(const seti& pipes) : PipeDispenser(pipes) {
+        _cv = (pthread_cond_t*)(malloc(sizeof (pthread_cond_t)));
+        pthread_cond_init(_cv, NULL);
+    }
+
+    ~PipeDispenserBlocking() {
+        pthread_cond_destroy(_cv);
+        free(_cv);
+    }
+
+    int getPipe(const seti& interested) {
+        lock();
+        intv avail;
+        while (getAvailablePipes(interested, avail) == 0) {
+            wait();
+        }
+        int pipe = avail[0];
+        _pipes.erase(pipe);
+        unlock();
+        return pipe;
+    }
+
+    void freePipe(int pipe) {
+        lock();
+        _pipes.insert(pipe);
+        broadcast();
+        unlock();
+    }
+};
+
+/*
+ * This one returns the least-occupied pipe.
+ */
+class PipeDispenserNonBlocking : public PipeDispenser  {
+protected:
+    std::map<int,int> _pipeUsers;
+
+public:
+    PipeDispenserNonBlocking(const seti& pipes) : PipeDispenser(pipes) {
+        for (seti::iterator it = pipes.begin(); it != pipes.end(); ++it) {
+            _pipeUsers[*it] = 0;
+        }
+    }
+
+    int getPipe(const seti& interested) {
+        lock();
+        int pipe = -1, users = 1 << 30;
+        for (seti::iterator it = _pipes.begin(); it != _pipes.end(); ++it) {
+            if (interested.count(*it) > 0 && _pipeUsers[*it] < users) {
+                pipe = *it;
+                users = _pipeUsers[*it];
+            }
+        }
+        if (pipe >= 0) {
+            _pipeUsers[pipe]++;
+        }
+        unlock();
+        return pipe;
+    }
+
+    void freePipe(int pipe) {
+        lock();
+        _pipeUsers[pipe]--;
+        unlock();
+    }
+};
+
+
+#endif /* PIPEDISPENSER_CUH_ */