priority_queue.hpp

Go to the documentation of this file.

#pragma once
 
#include <libdevcore/RLP.h>
 
#include <array>
#include <utility>
 
#include "logger/logger.hpp"
#include "network/tarcap/tarcap_version.hpp"
#include "network/threadpool/packets_blocking_mask.hpp"
#include "packets_queue.hpp"
 
namespace taraxa {
class PbftManager;
}
 
namespace taraxa::network::threadpool {
 
class PriorityQueue {
 public:
  PriorityQueue(size_t tp_workers_count, const std::shared_ptr<PbftManager>& pbft_mgr = nullptr,
                const addr_t& node_addr = {});
  ~PriorityQueue() = default;
 
  PriorityQueue(const PriorityQueue&) = delete;
  PriorityQueue& operator=(const PriorityQueue&) = delete;
  PriorityQueue(PriorityQueue&&) = delete;
  PriorityQueue& operator=(PriorityQueue&&) = delete;
 
  void pushBack(std::pair<tarcap::TarcapVersion, PacketData>&& packet);
 
  std::optional<std::pair<tarcap::TarcapVersion, PacketData>> pop();
 
  bool empty() const;
 
  void updateDependenciesStart(const PacketData& packet);
 
  void updateDependenciesFinish(const PacketData& packet, std::mutex& queue_mutex, std::condition_variable& cond_var);
 
  size_t getPrirotityQueueSize(PacketData::PacketPriority priority) const;
 
  bool isNonBlockingPacket(SubprotocolPacketType packet_type) const;
 
  bool updateBlockingDependencies(const PacketData& packet, bool unblock_processing = false);
 
 private:
  bool canBorrowThread();
 
 private:
  // Declare logger instances
  LOG_OBJECTS_DEFINE
 
  // Queues that group packets by it's priority.
  // All packets with PacketPriority::High go to packets_queues_[PacketPriority::High], etc...
  // TODO: make packets_queues_ const
  std::array<PacketsQueue, PacketData::PacketPriority::Count> packets_queues_{PacketsQueue(), PacketsQueue(),
                                                                              PacketsQueue()};
 
  // Mask with all packets types that are currently blocked for processing in another threads due to dependencies, e.g.
  // syncing packets must be processed synchronously one by one, etc...
  PacketsBlockingMask blocked_packets_mask_;
 
  // How many workers can process packets from all the queues at the same time
  const size_t MAX_TOTAL_WORKERS_COUNT;
 
  // How many workers are currently processing packets from all the queues at the same time
  std::atomic<size_t> act_total_workers_count_;
};
 
}  // namespace taraxa::network::threadpool