mlMultiThreadedPageRequestProcessor.h

Go to the documentation of this file.

/*************************************************************************************
**
** Copyright 2009, MeVis Medical Solutions AG
**
** The user may use this file in accordance with the license agreement provided with
** the Software or, alternatively, in accordance with the terms contained in a
** written agreement between the user and MeVis Medical Solutions AG.
**
** For further information use the contact form at https://www.mevislab.de/contact
**
**************************************************************************************/
 
#ifndef ML_MULTI_THREADED_PAGE_REQUEST_PROCESSOR_H
#define ML_MULTI_THREADED_PAGE_REQUEST_PROCESSOR_H
 
#include "mlInitSystemML.h"
 
#include "mlPageRequestProcessorBase.h"
#include "mlAtomicCounter.h"
#include "mlProcessingTimeLine.h"
 
#include "ThirdPartyWarningsDisable.h"
#include <boost/thread.hpp>
#include "ThirdPartyWarningsRestore.h"
#include <mlMutex.h>
#include <mlWaitCondition.h>
 
#define ML_HOST_START_SINGLETHREADED_TIMEOUT 0.0005
#define ML_HOST_PAGEREQUEST_QUEUE_LIMIT 2
#define ML_HOST_PAGEREQUEST_PRODUCER_BATCH_SIZE 2
#define ML_HOST_START_SINGLETHREADED_PAGEREQUEST_LIMIT 10
#define ML_HOST_MINIMAL_GUITHREAD_SLEEP_MS 10
 
ML_START_NAMESPACE
 
//-------------------------------------------------------------------------------------------
//-------------------------------------------------------------------------------------------
class MLEXPORT MultiThreadedPageRequestProcessor : public PageRequestProcessorBase
{
public:
  MultiThreadedPageRequestProcessor(Host* host = nullptr);
  ~MultiThreadedPageRequestProcessor() override;
 
  bool supportsMultiThreading() override { return true; }
 
  void setNumWorkerThreads(std::size_t threads) override;
 
  void addRootTileRequest(TileRequest* tileRequest) override;
 
  void processAll() override;
 
  void process(double timeBudget) override;
 
  void launchMultiThreading();
 
  bool needsProcessing() override;
 
  void append(PageRequest* request) override;
 
  void enterProcessingScope() override;
  void leaveProcessingScope() override;
 
  bool processingWasSingleThreaded() const override { return _processingWasSingleThreaded; }
 
  void setAllowLateMultiThreadingStart(bool flag) { _allowLateMultiThreadingStart = flag;  }
 
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 addNewCursors();
 
  void runWorkerThreadLoop(WorkerThreadState* state);
 
  void notifyAllThreads();
 
  void launchWorkerThreads();
 
  int getCurrentNumberOfPageRequestsInQueues();
 
  void wakeupProducerIfWaitingForQueue();
 
  void handleErrorsInGUIThread() override;
 
  void stopProcessingOnAllThreads();
 
  void moveRequestsFromGUIQueuesToOtherQueues();
 
  void getAllPageRequestQueues(std::vector< std::deque<PageRequest*>* >& queues) override;
 
  void startCreatingWorkSingleThreaded();
 
  bool _shutdown;
 
  bool _producerWorking;
  int  _workerThreadsWorking;
  int _maxPageRequestsInQueues;
  bool _producerWaitingForQueue;
 
  bool _processingAllowed;
 
  bool _useMultiThreading;
  bool _processingWasSingleThreaded;
  bool _allowLateMultiThreadingStart;
 
  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;
  };
 
  std::size_t _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