worker.cpp

Go to the documentation of this file.

// Copyright (C) 2009, 2010,2013 Lutz Foucar
// Copyright (C) 2010 Jochen Küpper

/**
 * @file worker.cpp file contains definition of class Worker and Workers
 *
 * @author Lutz Foucar
 */

#include <stdexcept>
#include <QtCore/QMutexLocker>

#include "worker.h"

#include "ratemeter.h"
#include "processor_manager.h"
#include "log.h"


using namespace cass;
using namespace std;

Worker::Worker(RingBuffer<CASSEvent> &ringbuffer,
               Ratemeter &ratemeter,
               QObject *parent)
  : PausableThread(lmf::PausableThread::_run,parent),
    _ringbuffer(ringbuffer),
    _process(ProcessorManager::reference()),
    _ratemeter(ratemeter)
{}

void Worker::runthis()
{
  _status = lmf::PausableThread::running;
  RingBuffer<CASSEvent>::iter_type rbItem;
  while(_control != _quit)
  {
    pausePoint();
    /** ensure that one has retrieved a valid event and that the id is non zero number*/
    if ((rbItem = _ringbuffer.nextToProcess(1000)) != _ringbuffer.end())
    {
      /** @note we need to check the element id in here, because if it is done
       *        in the outer if statement (as before) a valid ringbuffer element
       *        would never be returned to the buffer, as this only happens
       *        within the true if clause.
       */
      if(rbItem->element->id())
        _process(*rbItem->element);
      _ringbuffer.doneProcessing(rbItem);
      _ratemeter.count();
    }
  }
}








// ============define static members (do not touch)==============
Workers::shared_pointer Workers::_instance;
QMutex Workers::_mutex;

Workers::shared_pointer Workers::instance(RingBuffer<CASSEvent> &ringbuffer,
                                          Ratemeter &ratemeter,
                                          QObject *parent)
{
  QMutexLocker lock(&_mutex);
  if(!_instance)
    _instance = Workers::shared_pointer(new Workers(ringbuffer, ratemeter, parent));
  return _instance;
}

Workers::shared_pointer::element_type& Workers::reference()
{
  QMutexLocker lock(&_mutex);
  if (!_instance)
    throw logic_error("Workers::reference(): The instance has not yet been created");
  return *_instance;
}
//===============================================================


//-----------------------the wrapper for more than 1 worker--------------------
Workers::Workers(RingBuffer<CASSEvent> &ringbuffer,
                 Ratemeter &ratemeter,
                 QObject *parent)
  : _workers(NbrOfWorkers),
    _rb(ringbuffer)
{
  for (size_t i=0;i<_workers.size();++i)
    _workers[i] = Worker::shared_pointer(new Worker(ringbuffer,ratemeter,parent));
}

void Workers::start()
{
  for (size_t i=0;i<_workers.size();++i)
    _workers[i]->start();
}

void Workers::pause()
{
  for (size_t i=0;i<_workers.size();++i)
    _workers[i]->pause();
  for (size_t i=0;i<_workers.size();++i)
    _workers[i]->waitUntilPaused();
}

void Workers::resume()
{
  for (size_t i=0;i<_workers.size();++i)
    _workers[i]->resume();
}

void Workers::end()
{
  for (size_t i=0;i<_workers.size();++i)
    _workers[i]->end();
  for (size_t i=0;i<_workers.size();++i)
    _workers[i]->wait();
  ProcessorManager::instance()->aboutToQuit();
}

void Workers::rethrowException()
{
  for (size_t i=0;i<_workers.size();++i)
    _workers[i]->rethrowException();
}

bool Workers::running()const
{
  bool running(true);
  for (size_t i=0;i<_workers.size();++i)
    running = running && _workers[i]->isRunning();
  return running;
}