Merge input and storage workers thread.

Pavel Siska · Pavel Siska · commit 54f75bd90daf · 2022-05-21T13:14:45.000+02:00
diff --git a/ipfixprobe.cpp b/ipfixprobe.cpp
@@ -70,7 +70,6 @@ namespace ipxp {
 volatile sig_atomic_t stop = 0;
 
 volatile sig_atomic_t terminate_export = 0;
-volatile sig_atomic_t terminate_storage = 0;
 volatile sig_atomic_t terminate_input = 0;
 
 const uint32_t DEFAULT_IQUEUE_SIZE = 64;
@@ -350,34 +349,24 @@ bool process_plugin_args(ipxp_conf_t &conf, IpfixprobeOptParser &parser)
          storage_process_plugins.push_back(tmp);
       }
 
-      ipx_ring_t *input_queue = ipx_ring_init(conf.iqueue_size, 0);
-      if (input_queue == nullptr) {
-         throw IPXPError("unable to initialize ring buffer");
-      }
-
       std::promise<WorkerResult> *input_res = new std::promise<WorkerResult>();
-      std::promise<WorkerResult> *storage_res = new std::promise<WorkerResult>();
       conf.input_fut.push_back(input_res->get_future());
-      conf.storage_fut.push_back(storage_res->get_future());
 
       auto input_stats = new std::atomic<InputStats>();
       conf.input_stats.push_back(input_stats);
 
       WorkPipeline tmp = {
               {
                       input_plugin,
-                      new std::thread(input_worker, input_plugin, &conf.blocks[pipeline_idx * (conf.iqueue_size + 1)],
-                                      conf.iqueue_size + 1, conf.max_pkts, input_queue, input_res, input_stats),
+                      new std::thread(input_storage_worker, input_plugin, storage_plugin, &conf.blocks[pipeline_idx * (conf.iqueue_size + 1)],
+                                      conf.iqueue_size + 1, conf.max_pkts, input_res, input_stats),
                       input_res,
                       input_stats
               },
               {
                       storage_plugin,
-                      new std::thread(storage_worker, storage_plugin, input_queue, storage_res),
-                      storage_res,
                       storage_process_plugins
-              },
-              input_queue
+              }
       };
       conf.pipelines.push_back(tmp);
       pipeline_idx++;
@@ -398,9 +387,7 @@ void finish(ipxp_conf_t &conf)
    }
 
    // Terminate all storages
-   terminate_storage = 1;
    for (auto &it : conf.pipelines) {
-      it.storage.thread->join();
       for (auto &itp : it.storage.plugins) {
          itp->close();
       }
@@ -444,29 +431,6 @@ void finish(ipxp_conf_t &conf)
          std::setw(6) << status << std::endl;
    }
 
-   std::ostringstream oss;
-   oss << "Storage stats:" << std::endl <<
-      std::setw(3) << "#" <<
-      std::setw(7) << "status" << std::endl;
-
-   idx = 0;
-   bool storage_ok = true;
-   for (auto &it : conf.storage_fut) {
-      WorkerResult res = it.get();
-      std::string status = "ok";
-      if (res.error) {
-         ok = false;
-         storage_ok = false;
-         status = res.msg;
-      }
-      oss <<
-         std::setw(3) << idx++ << " " <<
-         std::setw(6) << status << std::endl;
-   }
-   if (!storage_ok) {
-      std::cout << oss.str();
-   }
-
    std::cout << "Output stats:" << std::endl <<
       std::setw(3) << "#" <<
       std::setw(10) << "biflows" <<
@@ -583,13 +547,6 @@ void main_loop(ipxp_conf_t &conf)
             break;
          }
       }
-      for (auto &it : conf.storage_fut) {
-         std::future_status status = it.wait_for(std::chrono::seconds(0));
-         if (status == std::future_status::ready) {
-            stop = 1;
-            break;
-         }
-      }
       for (auto &it : conf.output_fut) {
          std::future_status status = it.wait_for(std::chrono::seconds(0));
          if (status == std::future_status::ready) {
diff --git a/ipfixprobe.hpp b/ipfixprobe.hpp
@@ -72,7 +72,6 @@ extern const uint32_t DEFAULT_FPS;
 
 // global termination variable
 extern volatile sig_atomic_t terminate_export;
-extern volatile sig_atomic_t terminate_storage;
 extern volatile sig_atomic_t terminate_input;
 
 class IpfixprobeOptParser;
@@ -212,7 +211,6 @@ struct ipxp_conf_t {
    std::vector<std::atomic<OutputStats> *> output_stats;
 
    std::vector<std::shared_future<WorkerResult>> input_fut;
-   std::vector<std::future<WorkerResult>> storage_fut;
    std::vector<std::future<WorkerResult>> output_fut;  
 
    size_t pkt_bufsize;
@@ -243,15 +241,8 @@ struct ipxp_conf_t {
          delete it.input.promise;
       }
 
-      terminate_storage = 1;
       for (auto &it : pipelines) {
-         if (it.storage.thread->joinable()) {
-            it.storage.thread->join();
-         }
          delete it.storage.plugin;
-         delete it.storage.thread;
-         delete it.storage.promise;
-         ipx_ring_destroy(it.queue);
       }
 
       for (auto &it : pipelines) {
diff --git a/workers.cpp b/workers.cpp
@@ -51,15 +51,26 @@ namespace ipxp {
 
 #define MICRO_SEC 1000000L
 
-void input_worker(InputPlugin *plugin, PacketBlock *pkts, size_t block_cnt, uint64_t pkt_limit, ipx_ring_t *queue,
+void input_storage_worker(InputPlugin *plugin, StoragePlugin *cache, PacketBlock *pkts, size_t block_cnt, uint64_t pkt_limit,
                   std::promise<WorkerResult> *out, std::atomic<InputStats> *out_stats)
 {
-   struct timespec start;
-   struct timespec end;
+   struct timespec start_cache;
+   struct timespec end_cache;
+   struct timespec begin = {0, 0};
+   struct timespec end = {0, 0};
+   struct timeval ts = {0, 0};
+   bool timeout = false;
    size_t i = 0;
    InputPlugin::Result ret;
    InputStats stats = {0, 0, 0, 0, 0};
    WorkerResult res = {false, ""};
+
+#ifdef __linux__
+   const clockid_t clk_id = CLOCK_MONOTONIC_COARSE;
+#else
+   const clockid_t clk_id = CLOCK_MONOTONIC;
+#endif
+
    while (!terminate_input) {
       PacketBlock *block = &pkts[i];
       block->cnt = 0;
@@ -79,24 +90,40 @@ void input_worker(InputPlugin *plugin, PacketBlock *pkts, size_t block_cnt, uint
          break;
       }
       if (ret == InputPlugin::Result::TIMEOUT) {
+         clock_gettime(clk_id, &end);
+         if (!timeout) {
+            timeout = true;
+            begin = end;
+         }
+         struct timespec diff = {end.tv_sec - begin.tv_sec, end.tv_nsec - begin.tv_nsec};
+         if (diff.tv_nsec < 0) {
+            diff.tv_nsec += 1000000000;
+            diff.tv_sec--;
+         }
+         cache->export_expired(ts.tv_sec + diff.tv_sec);
          usleep(1);
          continue;
       } else if (ret == InputPlugin::Result::PARSED) {
          stats.packets = plugin->m_seen;
          stats.parsed = plugin->m_parsed;
          stats.dropped = plugin->m_dropped;
          stats.bytes += block->bytes;
-#ifdef __linux__
-         const clockid_t clk_id = CLOCK_MONOTONIC_COARSE;
-#else
-         const clockid_t clk_id = CLOCK_MONOTONIC;
-#endif
-         clock_gettime(clk_id, &start);
-         ipx_ring_push(queue, static_cast<void *>(block));
-         clock_gettime(clk_id, &end);
+         clock_gettime(clk_id, &start_cache);
+         try {
+            for (unsigned i = 0; i < block->cnt; i++) {
+               cache->put_pkt(block->pkts[i]);
+            }
+            ts = block->pkts[block->cnt - 1].ts;
+         } catch (PluginError &e) {
+            res.error = true;
+            res.msg = e.what();
+            break;
+         }
+         timeout = false;
+         clock_gettime(clk_id, &end_cache);
 
-         int64_t time = end.tv_nsec - start.tv_nsec;
-         if (start.tv_sec != end.tv_sec) {
+         int64_t time = end_cache.tv_nsec - start_cache.tv_nsec;
+         if (start_cache.tv_sec != end_cache.tv_sec) {
             time += 1000000000;
          }
          stats.qtime += time;
@@ -116,53 +143,6 @@ void input_worker(InputPlugin *plugin, PacketBlock *pkts, size_t block_cnt, uint
    stats.parsed = plugin->m_parsed;
    stats.dropped = plugin->m_dropped;
    out_stats->store(stats);
-   out->set_value(res);
-}
-
-void storage_worker(StoragePlugin *cache, ipx_ring_t *queue, std::promise<WorkerResult> *out)
-{
-   WorkerResult res = {false, ""};
-   bool timeout = false;
-   struct timeval ts = {0, 0};
-   struct timespec begin = {0, 0};
-   struct timespec end = {0, 0};
-#ifdef __linux__
-   const clockid_t clk_id = CLOCK_MONOTONIC_COARSE;
-#else
-   const clockid_t clk_id = CLOCK_MONOTONIC;
-#endif
-   while (1) {
-      PacketBlock *block = static_cast<PacketBlock *>(ipx_ring_pop(queue));
-      if (block) {
-         try {
-            for (unsigned i = 0; i < block->cnt; i++) {
-               cache->put_pkt(block->pkts[i]);
-            }
-            ts = block->pkts[block->cnt - 1].ts;
-         } catch (PluginError &e) {
-            res.error = true;
-            res.msg = e.what();
-            break;
-         }
-         timeout = false;
-      } else if (terminate_storage && !ipx_ring_cnt(queue)) {
-         break;
-      } else {
-         clock_gettime(clk_id, &end);
-         if (!timeout) {
-            timeout = true;
-            begin = end;
-         }
-         struct timespec diff = {end.tv_sec - begin.tv_sec, end.tv_nsec - begin.tv_nsec};
-         if (diff.tv_nsec < 0) {
-            diff.tv_nsec += 1000000000;
-            diff.tv_sec--;
-         }
-         cache->export_expired(ts.tv_sec + diff.tv_sec);
-         usleep(1);
-      }
-   }
-
    cache->finish();
    auto outq = cache->get_queue();
    while (ipx_ring_cnt(outq)) {
diff --git a/workers.hpp b/workers.hpp
@@ -74,11 +74,8 @@ struct WorkPipeline {
    } input;
    struct {
       StoragePlugin *plugin;
-      std::thread *thread;
-      std::promise<WorkerResult> *promise;
       std::vector<ProcessPlugin *> plugins;
    } storage;
-   ipx_ring_t *queue;
 };
 
 struct OutputWorker {
@@ -89,9 +86,8 @@ struct OutputWorker {
    ipx_ring_t *queue;
 };
 
-void input_worker(InputPlugin *plugin, PacketBlock *pkts, size_t block_cnt, uint64_t pkt_limit, ipx_ring_t *queue,
+void input_storage_worker(InputPlugin *plugin, StoragePlugin *cache, PacketBlock *pkts, size_t block_cnt, uint64_t pkt_limit, 
       std::promise<WorkerResult> *out, std::atomic<InputStats> *out_stats);
-void storage_worker(StoragePlugin *cache, ipx_ring_t *queue, std::promise<WorkerResult> *out);
 void output_worker(OutputPlugin *exp, ipx_ring_t *queue, std::promise<WorkerResult> *out, std::atomic<OutputStats> *out_stats,
       uint32_t fps);
 
