Open Source Projects Europe

 *

 *

 *

 *

 */

 */

#ifndef _WORKQUEUE_H_INCLUDED_

#ifndef _WORKQUEUE_H_INCLUDED_

#define _WORKQUEUE_H_INCLUDED_

#define _WORKQUEUE_H_INCLUDED_

#include "libupnpp/config.h"

#include "libupnpp/config.h"

public:

public:

    /** Create a WorkQueue

    /** Create a WorkQueue

     * @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

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

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

     */

     */

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

    WorkQueue(const std::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_ok = (pthread_cond_init(&m_ccond, 0) == 0) &&

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

            (pthread_cond_init(&m_wcond, 0) == 0);

    }

    }

    ~WorkQueue()

        if (!m_worker_threads.empty())

        if (!m_worker_threads.empty())

            setTerminateAndWait();

            setTerminateAndWait();

    }

    }

    /** Start the worker threads.

    /** Start the worker threads.

     *

     *

     * @param nworkers number of threads copies to start.

     * @param nworkers number of threads copies to start.

     * @param start_routine thread function. It should loop

     * @param start_routine thread function. It should loop

     * @param arg initial parameter to thread function.

     * @param arg initial parameter to thread function.

     * @return true if ok.

     * @return true if ok.

     */

     */

    bool start(int nworkers, void *(*workproc)(void *), void *arg)

        PTMutexLocker lock(m_mutex);

        PTMutexLocker lock(m_mutex);

        for (int i = 0; i < nworkers; i++) {

            int err;

            int err;

            pthread_t thr;

            pthread_t thr;

            if ((err = pthread_create(&thr, 0, workproc, arg))) {

            if ((err = pthread_create(&thr, 0, workproc, arg))) {

                return false;

                return false;

            }

            }

    /** Add item to work queue, called from client.

    /** Add item to work queue, called from client.

     *

     *

     * Sleeps if there are already too many.

     * Sleeps if there are already too many.

     */

     */

    bool put(T t, bool flushprevious = false)

        PTMutexLocker lock(m_mutex);

        PTMutexLocker lock(m_mutex);

        if (!lock.ok() || !ok()) {

        if (!lock.ok() || !ok()) {

            return false;

            return false;

        }

        }

     * idle EXCEPT if the caller knows that no jobs are still being created.

     * idle EXCEPT if the caller knows that no jobs are still being created.

     * It would be possible to transform this into a safe call if some kind

     * It would be possible to transform this into a safe call if some kind

     * of suspend condition was set on the queue by waitIdle(), to be reset by

     * of suspend condition was set on the queue by waitIdle(), to be reset by

     * some kind of "resume" call. Not currently the case.

     * some kind of "resume" call. Not currently the case.

     */

     */

    bool waitIdle()

        PTMutexLocker lock(m_mutex);

        PTMutexLocker lock(m_mutex);

        if (!lock.ok() || !ok()) {

        if (!lock.ok() || !ok()) {

            return false;

            return false;

        }

        }

    /** Tell the workers to exit, and wait for them.

    /** Tell the workers to exit, and wait for them.

     *

     *

     * Does not bother about tasks possibly remaining on the queue, so

     * Does not bother about tasks possibly remaining on the queue, so

     * should be called after waitIdle() for an orderly shutdown.

     * should be called after waitIdle() for an orderly shutdown.

     */

     */

    void* setTerminateAndWait()

        PTMutexLocker lock(m_mutex);

        PTMutexLocker lock(m_mutex);

        if (m_worker_threads.empty()) {

        if (m_worker_threads.empty()) {

            // Already called ?

            // Already called ?

            return (void*)0;

            return (void*)0;

            m_worker_threads.erase(it);

            m_worker_threads.erase(it);

        }

        }

        // Reset to start state.

        // Reset to start state.

        m_workers_exited = m_clients_waiting = m_workers_waiting =

        m_workers_exited = m_clients_waiting = m_workers_waiting =

            m_tottasks = m_nowake = m_workersleeps = m_clientsleeps = 0;

        m_ok = true;

        m_ok = true;

        return statusall;

        return statusall;

    }

    }

    /** Take task from queue. Called from worker.

    /** Take task from queue. Called from worker.

     *

     *

     * Sleeps if there are not enough. Signal if we go to sleep on empty

     * Sleeps if there are not enough. Signal if we go to sleep on empty

     * queue: client may be waiting for our going idle.

     * queue: client may be waiting for our going idle.

     */

     */

    bool take(T* tp, size_t *szp = 0)

        PTMutexLocker lock(m_mutex);

        PTMutexLocker lock(m_mutex);

        if (!lock.ok() || !ok()) {

        if (!lock.ok() || !ok()) {

            return false;

            return false;

        }

        }

     * the queue is terminated by the client. Workers never exit normally,

     * the queue is terminated by the client. Workers never exit normally,

     * except when the queue is shut down (at which point m_ok is set to

     * except when the queue is shut down (at which point m_ok is set to

     * false by the shutdown code anyway). The thread must return/exit

     * false by the shutdown code anyway). The thread must return/exit

     * immediately after calling this.

     * immediately after calling this.

     */

     */

    void workerExit()

        PTMutexLocker lock(m_mutex);

        PTMutexLocker lock(m_mutex);

        m_workers_exited++;

        m_workers_exited++;

        m_ok = false;

        m_ok = false;

        pthread_cond_broadcast(&m_ccond);

        pthread_cond_broadcast(&m_ccond);

    }

    }

    size_t qsize()

        PTMutexLocker lock(m_mutex);

        PTMutexLocker lock(m_mutex);

        size_t sz = m_queue.size();

        size_t sz = m_queue.size();

        return sz;

        return sz;

    }

    }

private:

private:

    bool ok()

        bool isok = m_ok && m_workers_exited == 0 && !m_worker_threads.empty();

        bool isok = m_ok && m_workers_exited == 0 && !m_worker_threads.empty();

        return isok;

        return isok;

    }

    }

    long long nanodiff(const struct timespec& older,

    long long nanodiff(const struct timespec& older,

                       const struct timespec& newer)

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

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

            + newer.tv_nsec - older.tv_nsec;

    }

    }

    // Configuration

    // Configuration

    std::string m_name;

    std::string m_name;

    size_t m_high;

    size_t m_high;