Open Source Projects Europe

/*

 *   This program is free software; you can redistribute it and/or modify

 *   it under the terms of the GNU General Public License as published by

 *   the Free Software Foundation; either version 2 of the License, or

 *   (at your option) any later version.

 *

 *   This program is distributed in the hope that it will be useful,

 *   but WITHOUT ANY WARRANTY; without even the implied warranty of

 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 *   GNU General Public License for more details.

 *

 *   You should have received a copy of the GNU General Public License

 *   along with this program; if not, write to the

 *   Free Software Foundation, Inc.,

 *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.

 */

#ifndef _WORKQUEUE_H_INCLUDED_

#define _WORKQUEUE_H_INCLUDED_

#include "pthread.h"

#include <string>

#include <queue>

using std::queue;

using std::string;

/**

 * A WorkQueue manages the synchronisation around a queue of work items,

 * where a single client thread queues tasks and a single worker takes

 * and executes them. The goal is to introduce some level of

 * parallelism between the successive steps of a previously single

 * threaded pipe-line (data extraction / data preparation / index

 * update).

 *

 * There is no individual task status return. In case of fatal error,

 * the client or worker sets an end condition on the queue. A second

 * queue could conceivably be used for returning individual task

 * status.

 */

template <class T> class WorkQueue {

public:

    WorkQueue(int hi = 0, int lo = 1)

  : m_high(hi), m_low(lo), m_size(0), m_worker_up(false),

    m_worker_waiting(false), m_jobcnt(0), m_lenacc(0)

    {

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

      (pthread_mutex_init(&m_mutex, 0) == 0);

    }

    ~WorkQueue() 

    {

  if (m_worker_up)

      setTerminateAndWait();

    }

    /** Start the worker thread. The start_routine will loop

     *  taking and executing tasks. */

    bool start(void *(*start_routine)(void *), void *arg)

    {

  bool status = pthread_create(&m_worker_thread, 0, 

                   start_routine, arg) == 0;

  if (status)

      m_worker_up = true;

  return status;

    }

    /**

     * Add item to work queue. Sleep if there are already too many.

     * Called from client.

     */

    bool put(T t)

    {

  if (!ok() || pthread_mutex_lock(&m_mutex) != 0) 

      return false;

  while (ok() && m_high > 0 && m_queue.size() >= m_high) {

      // Keep the order: we test ok() AFTER the sleep...

      if (pthread_cond_wait(&m_cond, &m_mutex) || !ok()) {

      pthread_mutex_unlock(&m_mutex);

      return false;

      }

  }

  m_queue.push(t);

  ++m_size;

  pthread_cond_broadcast(&m_cond);

  pthread_mutex_unlock(&m_mutex);

  return true;

    }

    /** Wait until the queue is empty and the worker is

     *  back waiting for task. Called from the client when it needs to

     *  perform work that couldn't be done in parallel with the

     *  worker's tasks.

     */

    bool waitIdle()

    {

  if (!ok() || pthread_mutex_lock(&m_mutex) != 0) 

      return false;

  // We're done when the queue is empty AND the worker is back

  // for a task (has finished the last)

  while (ok() && (m_queue.size() > 0 || !m_worker_waiting)) {

      if (pthread_cond_wait(&m_cond, &m_mutex)) {

      pthread_mutex_unlock(&m_mutex);

      return false;

      }

  }

  pthread_mutex_unlock(&m_mutex);

  return ok();

    }

    /** Tell the worker to exit, and wait for it. There may still

  be tasks on the queue. */

    void* setTerminateAndWait()

    {

  if (!m_worker_up)

      return (void *)0;

  pthread_mutex_lock(&m_mutex);

  m_ok = false;

  pthread_cond_broadcast(&m_cond);

  pthread_mutex_unlock(&m_mutex);

  void *status;

  pthread_join(m_worker_thread, &status);

  m_worker_up = false;

  return status;

    }

    /** Remove task from queue. Sleep if there are not enough. Signal if we go

  to sleep on empty queue: client may be waiting for our going idle */

    bool take(T* tp)

    {

  if (!ok() || pthread_mutex_lock(&m_mutex) != 0)

      return false;

  while (ok() && m_queue.size() < m_low) {

      m_worker_waiting = true;

      if (m_queue.empty())

      pthread_cond_broadcast(&m_cond);

      if (pthread_cond_wait(&m_cond, &m_mutex) || !ok()) {

      pthread_mutex_unlock(&m_mutex);

      m_worker_waiting = false;

      return false;

      }

      m_worker_waiting = false;

  }

  ++m_jobcnt;

  m_lenacc += m_size;

  *tp = m_queue.front();

  m_queue.pop();

  --m_size;

  pthread_cond_broadcast(&m_cond);

  pthread_mutex_unlock(&m_mutex);

  return true;

    }

    /** Take note of the worker exit. This would normally happen after an

  unrecoverable error */

    void workerExit()

    {

  if (!ok() || pthread_mutex_lock(&m_mutex) != 0)

      return;

  m_ok = false;

  pthread_cond_broadcast(&m_cond);

  pthread_mutex_unlock(&m_mutex);

    }

    /** Debug only: as the size is returned while the queue is unlocked, there

     *  is no warranty on its consistency. Not that we use the member size, not 

     *  the container size() call which would need locking.

     */

    size_t size() {return m_size;}

private:

    bool ok() {return m_ok && m_worker_up;}

    size_t m_high;

    size_t m_low; 

    size_t m_size;

    bool m_worker_up;

    bool m_worker_waiting;

    int m_jobcnt;

    int m_lenacc;

    pthread_t m_worker_thread;

    queue<T> m_queue;

    pthread_cond_t m_cond;

    pthread_mutex_t m_mutex;

    bool m_ok;

};

#endif /* _WORKQUEUE_H_INCLUDED_ */