MeVis/Foundation/Sources/ML/include/host/mlMultiThreadedPageRequestProcessor.h

Go to the documentation of this file.

// **InsertLicense** code
//-------------------------------------------------------------------------
//-------------------------------------------------------------------------
#ifndef __mlMultiThreadedPageRequestProcessor_H
#define __mlMultiThreadedPageRequestProcessor_H

#include "mlInitSystemML.h"

#include "mlPageRequestProcessorBase.h"
#include "mlAtomicCounter.h"
#include "mlProcessingTimeLine.h"

#include <boost/thread.hpp>
#include <mlMutex.h>
#include <mlWaitCondition.h>

ML_START_NAMESPACE

//-------------------------------------------------------------------------------------------
//-------------------------------------------------------------------------------------------
class MLEXPORT MultiThreadedPageRequestProcessor : public PageRequestProcessorBase
{
public:
  MultiThreadedPageRequestProcessor(Host* host = NULL);
  ~MultiThreadedPageRequestProcessor();

  virtual bool supportsMultiThreading() { return true; }

  virtual void setNumWorkerThreads(int threads);

  virtual void addRootTileRequest(TileRequest* tileRequest);

  void processAll();

  void process(double timeBudget);

  bool needsProcessing();

  virtual void append(PageRequest* request);

  virtual void enterProcessingScope();
  virtual void leaveProcessingScope();

private:
  struct WorkerThreadState;

  bool processNextGUIRequest(int millisecondsToWait);

  void processNextWorkerRequest(WorkerThreadState* state, PerThreadStorage& perThreadStorage);

  MLErrorCode processRequest(PageRequest* request, bool isGUIThread, PerThreadStorage& perThreadStorage);

  static void rootTileRequestFinishedCB(void* data, TileRequest* request);

  void rootTileRequestFinished(TileRequest* request);

  void runPageRequestProducerLoop();

  void runWorkerThreadLoop(WorkerThreadState* state);
  
  void notifyAllThreads();

  void launchWorkerThreads();
  
  int getCurrentNumberOfPageRequestsInQueues();

  void wakeupProducerIfWaitingForQueue();
 
  virtual void handleErrorsInGUIThread();

  void stopProcessingOnAllThreads();

  virtual void getAllPageRequestQueues(std::vector< std::deque<PageRequest*>* >& queues);
  
  bool _shutdown;

  bool _producerWorking;
  int  _workerThreadsWorking;
  int _maxPageRequestsInQueues;
  bool _producerWaitingForQueue;

  bool _processingAllowed;

  WaitCondition _stoppedProcessing;

  WaitCondition _guiThreadWakeup;
  WaitCondition _producerWakeup;
  WaitCondition _workerWakeup;
  WaitCondition _ioThreadWakeup;

  Mutex _mutex;

  std::vector<PageRequestCursor*> _newCursors;

  std::deque<PageRequest*> _parallelWorkQueue;
  std::deque<PageRequest*> _ioWorkQueue;

  struct WorkerThreadState
  {
    WorkerThreadState();
    ~WorkerThreadState();

    boost::thread* _thread;

    bool _working;
    int  _workerThreadId;
    ProcessingTimeLine::TimeLine* _timeLine;

    WaitCondition* _wakeup;
    std::deque<PageRequest*>* _queue;

    bool _shutdown;

    TimeCounter _profilingTimer;
  };

  int _numDesiredWorkerThreads;

  std::vector<WorkerThreadState*> _workerThreads;

  WorkerThreadState* _ioThread;

  boost::thread* _producerThread;

  ProcessingTimeLine::TimeLine* _producerThreadTimeLine;

  struct ProfilingInfo {
    ProfilingInfo() {
      count = 0;
      elapsedTime = 0;
    }

    int count;

    double elapsedTime;
  };

  std::map<PagedImage*, ProfilingInfo> _profilingData;

  void addThreadTimesToMLProfiling();

};

ML_END_NAMESPACE

#endif