src/asyncrpcqueue.h

   1 // Copyright (c) 2016 The Zcash developers
   2 // Distributed under the MIT software license, see the accompanying
   3 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
   4
   5 #ifndef ASYNCRPCQUEUE_H
   6 #define ASYNCRPCQUEUE_H
   7
   8 #include "asyncrpcoperation.h"
   9
  10 #include <iostream>
  11 #include <string>
  12 #include <chrono>
  13 #include <queue>
  14 #include <unordered_map>
  15 #include <vector>
  16 #include <future>
  17 #include <thread>
  18 #include <utility>
  19 #include <memory>
  20
  21
  22 typedef std::unordered_map<AsyncRPCOperationId, std::shared_ptr<AsyncRPCOperation> > AsyncRPCOperationMap;
  23
  24
  25 class AsyncRPCQueue {
  26 public:
  27     static shared_ptr<AsyncRPCQueue> sharedInstance();
  28
  29     AsyncRPCQueue();
  30     virtual ~AsyncRPCQueue();
  31
  32     // We don't want queue to be copied or moved around
  33     AsyncRPCQueue(AsyncRPCQueue const&) = delete;             // Copy construct
  34     AsyncRPCQueue(AsyncRPCQueue&&) = delete;                  // Move construct
  35     AsyncRPCQueue& operator=(AsyncRPCQueue const&) = delete;  // Copy assign
  36     AsyncRPCQueue& operator=(AsyncRPCQueue &&) = delete;      // Move assign
  37
  38     void addWorker();
  39     size_t getNumberOfWorkers() const;
  40     bool isClosed() const;
  41     bool isFinishing() const;
  42     void close(); // close queue and cancel all operations
  43     void finish(); // close queue but finishing existing operations
  44     void closeAndWait(); // block thread until all threads have terminated.
  45     void finishAndWait(); // block thread until existing operations have finished, threads terminated
  46     void cancelAllOperations(); // mark all operations in the queue as cancelled
  47     size_t getOperationCount() const;
  48     std::shared_ptr<AsyncRPCOperation> getOperationForId(AsyncRPCOperationId) const;
  49     std::shared_ptr<AsyncRPCOperation> popOperationForId(AsyncRPCOperationId);
  50     void addOperation(const std::shared_ptr<AsyncRPCOperation> &ptrOperation);
  51     std::vector<AsyncRPCOperationId> getAllOperationIds() const;
  52
  53 private:
  54     // addWorker() will spawn a new thread on run())
  55     void run(size_t workerId);
  56     void wait_for_worker_threads();
  57
  58     // Why this is not a recursive lock: http://www.zaval.org/resources/library/butenhof1.html
  59     mutable std::mutex lock_;
  60     std::condition_variable condition_;
  61     std::atomic<bool> closed_;
  62     std::atomic<bool> finish_;
  63     AsyncRPCOperationMap operation_map_;
  64     std::queue <AsyncRPCOperationId> operation_id_queue_;
  65     std::vector<std::thread> workers_;
  66 };
  67
  68 #endif
  69
  70