tarcap_thread_pool.hpp

Go to the documentation of this file.

#pragma once
 
#include <condition_variable>
#include <mutex>
#include <thread>
#include <vector>
 
#include "logger/logging.hpp"
#include "network/tarcap/tarcap_version.hpp"
#include "priority_queue.hpp"
 
namespace taraxa::network::tarcap {
class PacketsHandler;
}  // namespace taraxa::network::tarcap
 
namespace taraxa {
class PbftManager;
}
 
namespace taraxa::network::threadpool {
 
class PacketsThreadPool {
 public:
  PacketsThreadPool(size_t workers_num = 10, const std::shared_ptr<PbftManager>& pbft_mgr = nullptr);
  ~PacketsThreadPool();
 
  PacketsThreadPool(const PacketsThreadPool&) = delete;
  PacketsThreadPool& operator=(const PacketsThreadPool&) = delete;
  PacketsThreadPool(PacketsThreadPool&&) = delete;
  PacketsThreadPool& operator=(PacketsThreadPool&&) = delete;
 
  std::optional<uint64_t> push(std::pair<tarcap::TarcapVersion, PacketData>&& packet_data);
 
  void startProcessing();
 
  void stopProcessing();
 
  void processPacket(size_t worker_id);
 
  void setPacketsHandlers(tarcap::TarcapVersion tarcap_version,
                          std::shared_ptr<tarcap::PacketsHandler> packets_handlers);
 
  std::tuple<size_t, size_t, size_t> getQueueSize() const;
 
 private:
  // Number of workers(threads)
  const size_t workers_num_;
 
  // Common packets handler - each taraxa capability haits own packets handler
  std::unordered_map<tarcap::TarcapVersion, std::shared_ptr<tarcap::PacketsHandler>> packets_handlers_;
 
  // If true, stop processing packets and join all workers threads
  std::atomic<bool> stopProcessing_{false};
 
  // How many packets were pushed into the queue, it also serves for creating packet unique id
  uint64_t packets_count_{0};
 
  // Queue of unprocessed packets
  PriorityQueue queue_;
 
  // Queue mutex
  std::mutex queue_mutex_;
 
  // Queue condition variable
  std::condition_variable cond_var_;
 
  // Vector of worker threads - should be initialized as the last member
  std::vector<std::thread> workers_;
 
  logger::Logger logger_;
};
 
}  // namespace taraxa::network::threadpool