CTT - Terminate all worker threads at once

Author: Damir Zainullin

src/core/ipfixprobe.cpp

@@ -448,7 +448,9 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 					conf.iqueue_size,
 					conf.max_pkts,
 					input_res,
-					input_stats),
+					input_stats,
+					pipeline_idx,
+					&conf.finished_workers),
 				input_res,
 				input_stats},
 			   {storagePlugin, storage_process_plugins}};

src/core/ipfixprobe.hpp

@@ -325,6 +325,7 @@ struct ipxp_conf_t {
 	} active;
 
 	std::vector<WorkPipeline> pipelines;
+	FinishedWorkers finished_workers;
 	std::vector<OutputWorker> outputs;
 
 	std::vector<std::atomic<InputStats>*> input_stats;

src/core/workers.cpp

@@ -43,7 +43,9 @@ void input_storage_worker(
 	size_t queue_size,
 	uint64_t pkt_limit,
 	std::promise<WorkerResult>* out,
-	std::atomic<InputStats>* out_stats)
+	std::atomic<InputStats>* out_stats,
+	size_t worker_id,
+	FinishedWorkers* finished_workers)
 {
 	struct timespec start_cache;
 	struct timespec end_cache;
@@ -54,6 +56,7 @@ void input_storage_worker(
 	InputPlugin::Result ret;
 	InputStats stats = {0, 0, 0, 0, 0};
 	WorkerResult res = {false, ""};
+	finished_workers->mark_in_progress(worker_id);
 
 	PacketBlock block(queue_size);
 
@@ -63,10 +66,13 @@ void input_storage_worker(
 	const clockid_t clk_id = CLOCK_MONOTONIC;
 #endif
 
-	while (storagePlugin->requires_input()) {
+	while (!finished_workers->all_finished()) {
 		if (terminate_input) {
 			storagePlugin->terminate_input();
 		}
+		if (!storagePlugin->requires_input()) {
+			finished_workers->mark_finished(worker_id);
+		}
 		block.cnt = 0;
 		block.bytes = 0;
 

src/core/workers.hpp

@@ -71,13 +71,46 @@ struct OutputWorker {
 	ipx_ring_t* queue;
 };
 
+class FinishedWorkers {
+	constexpr static size_t MAX_WORKERS = 64;
+	enum class WorkerStatus : uint8_t {
+		IN_PROGRESS = 0,
+		FINISHED = 1
+	};
+	std::array<WorkerStatus, MAX_WORKERS> workers_status;
+
+public:
+	FinishedWorkers() noexcept
+	{
+		workers_status.fill(WorkerStatus::FINISHED);
+	}
+
+	void mark_in_progress(size_t worker_id) noexcept
+	{
+		workers_status[worker_id] = WorkerStatus::IN_PROGRESS;
+	}
+
+	void mark_finished(size_t worker_id) noexcept
+	{
+		workers_status[worker_id] = WorkerStatus::FINISHED;
+	}
+
+	bool all_finished() const noexcept
+	{
+		return std::all_of(workers_status.begin(), workers_status.end(),
+		                   [](WorkerStatus status) { return status == WorkerStatus::FINISHED; });
+	}
+};
+
 void input_storage_worker(
 	std::shared_ptr<InputPlugin> inputPlugin,
 	std::shared_ptr<StoragePlugin> storagePlugin,
 	size_t queue_size,
 	uint64_t pkt_limit,
 	std::promise<WorkerResult>* out,
-	std::atomic<InputStats>* out_stats);
+	std::atomic<InputStats>* out_stats,
+	size_t worker_id,
+	FinishedWorkers* finished_workers);
 void output_worker(
 	std::shared_ptr<OutputPlugin> outputPlugin,
 	ipx_ring_t* queue,

src/plugins/storage/cache-ctt/src/cacheCtt.cpp

@@ -168,11 +168,6 @@ void NHTFlowCacheCtt::flush_ctt(const timeval now) noexcept
       m_table_flushed = !m_ctt_flow_seen; 
       m_ctt_flow_seen = false;
    }
-
-   if (m_ctt_flows_flushed >= 16) {
-      m_ctt_flows_flushed = 0;
-      usleep(400);   
-   }
 }
 
 std::optional<feta::OffloadMode> NHTFlowCacheCtt::get_offload_mode(size_t flow_index) noexcept