Open Source Projects Europe

#include <time.h>

#include <time.h>

#include <string>

#include <string>

#include <queue>

#include <queue>

#include <unordered_map>

#include <unordered_map>

#include "ptmutex.hxx"

#include "ptmutex.hxx"

/// Store per-worker-thread data. Just an initialized timespec, and

/// Store per-worker-thread data. Just an initialized timespec, and

/// used at the moment.

/// used at the moment.

class WQTData {

class WQTData {

public:

public:

     * @param name for message printing

     * @param name for message printing

     * @param hi number of tasks on queue before clients blocks. Default 0

     * @param hi number of tasks on queue before clients blocks. Default 0

     *    meaning no limit. hi == -1 means that the queue is disabled.

     *    meaning no limit. hi == -1 means that the queue is disabled.

     * @param lo minimum count of tasks before worker starts. Default 1.

     * @param lo minimum count of tasks before worker starts. Default 1.

     */

     */

    WorkQueue(const string& name, size_t hi = 0, size_t lo = 1)

        : m_name(name), m_high(hi), m_low(lo),

        : m_name(name), m_high(hi), m_low(lo),

          m_workers_exited(0), m_clients_waiting(0), m_workers_waiting(0),

          m_workers_exited(0), m_clients_waiting(0), m_workers_waiting(0),

          m_tottasks(0), m_nowake(0), m_workersleeps(0), m_clientsleeps(0)

          m_tottasks(0), m_nowake(0), m_workersleeps(0), m_clientsleeps(0)

    {

    {

        m_ok = (pthread_cond_init(&m_ccond, 0) == 0) &&

        m_ok = (pthread_cond_init(&m_ccond, 0) == 0) &&

        }

        }

        // Perform the thread joins and compute overall status

        // Perform the thread joins and compute overall status

        // Workers return (void*)1 if ok

        // Workers return (void*)1 if ok

        void *statusall = (void*)1;

        void *statusall = (void*)1;

        unordered_map<pthread_t,  WQTData>::iterator it;

        while (!m_worker_threads.empty()) {

        while (!m_worker_threads.empty()) {

            void *status;

            void *status;

            it = m_worker_threads.begin();

            it = m_worker_threads.begin();

            pthread_join(it->first, &status);

            pthread_join(it->first, &status);

            if (status == (void *)0)

            if (status == (void *)0)

        return (newer.tv_sec - older.tv_sec) * 1000000000LL

        return (newer.tv_sec - older.tv_sec) * 1000000000LL

            + newer.tv_nsec - older.tv_nsec;

            + newer.tv_nsec - older.tv_nsec;

    }

    }

    // Configuration

    // Configuration

    string m_name;

    size_t m_high;

    size_t m_high;

    size_t m_low;

    size_t m_low;

    // Status

    // Status

    // Worker threads having called exit

    // Worker threads having called exit

    unsigned int m_workers_exited;

    unsigned int m_workers_exited;

    bool m_ok;

    bool m_ok;

    // Per-thread data. The data is not used currently, this could be

    // Per-thread data. The data is not used currently, this could be

    // a set<pthread_t>

    // a set<pthread_t>

    unordered_map<pthread_t, WQTData> m_worker_threads;

    // Synchronization

    // Synchronization

    queue<T> m_queue;

    pthread_cond_t m_ccond;

    pthread_cond_t m_ccond;

    pthread_cond_t m_wcond;

    pthread_cond_t m_wcond;

    PTMutexInit m_mutex;

    PTMutexInit m_mutex;

    // Client/Worker threads currently waiting for a job

    // Client/Worker threads currently waiting for a job

    unsigned int m_clients_waiting;

    unsigned int m_clients_waiting;

    unsigned int m_nowake;

    unsigned int m_nowake;

    unsigned int m_workersleeps;

    unsigned int m_workersleeps;

    unsigned int m_clientsleeps;

    unsigned int m_clientsleeps;

};

};

#endif /* _WORKQUEUE_H_INCLUDED_ */

#endif /* _WORKQUEUE_H_INCLUDED_ */