Merge pull request #111 from Flow-IPC/ipc-143_perf-dive-nov

(See description!) Pull request for repo [flow] branch [ipc-143_perf-dive-nov] to base-branch [main].

Author: ygoldfeld

.github/workflows/main.yml

@@ -183,6 +183,7 @@ jobs:
             version: 9
             c-path: /usr/bin/gcc-9
             cpp-path: /usr/bin/g++-9
+            install: True
           - id: gcc-10
             name: gcc
             version: 10
@@ -395,13 +396,27 @@ jobs:
       # (Unfortunately cannot refer to earlier-assigned `env.` entries within subsequent ones.)
       install-dir: ${{ github.workspace }}/install/${{ matrix.build-test-cfg.conan-profile-build-type }}/usr/local
       # For the remaining env entries please see comments in Flow-IPC workflow counterpart.  We use same techniques.
+      # Update: (As of this writing this is only here, not in Flow-IPC counterpart, but that'll change; might want to
+      # move the comment there then.)  ASAN is somewhat different from the others, as it's really two sanitizers with
+      # separate suppression files -- ASAN (actual safety checks) and LSAN (leak checks).  So far we do *not* need
+      # any suppressed ASAN warnings (ASAN false positives are known to be rare, so that makes sense), but there are
+      # at least a couple minor, intentional mem-leaks, so we *do* need to suppress LSAN warnings at times.  Therefore,
+      # until/unless this changes, we will essentially treat LSAN-suppression as *the* suppression for our ASAN runs,
+      # thus only requiring one suppression file type still.  If/when that changes, the below will need to be made
+      # more complicated, so that 2 suppression types per sanitizer are supported as opposed to just 1.  Until then,
+      # it's still simple.  So that's why $ASAN_OPTIONS does not mention suppression, but $LSAN_OPTIONS does... and
+      # it is the suppression file(s) (if any) under various `asan/` dirs in the source tree.
       san-suppress-cfg-file: ${{ github.workspace }}/install/${{ matrix.build-test-cfg.conan-profile-build-type }}/usr/local/bin/san_suppressions.cfg
       san-suppress-cfg-in-file1: sanitize/${{ matrix.build-test-cfg.sanitizer-name }}/suppressions_${{ matrix.compiler.name }}.cfg
       san-suppress-cfg-in-file2: sanitize/${{ matrix.build-test-cfg.sanitizer-name }}/suppressions_${{ matrix.compiler.name }}_${{ matrix.compiler.version }}.cfg
       setup-tests-env: |
         if [ '${{ matrix.build-test-cfg.sanitizer-name }}' = asan ]; then
+          export SAN_SUPP=1
+          export SAN_SUPP_CFG=${{ github.workspace }}/install/${{ matrix.build-test-cfg.conan-profile-build-type }}/usr/local/bin/san_suppressions.cfg
           export ASAN_OPTIONS='disable_coredump=0'
+          export LSAN_OPTIONS="suppressions=$SAN_SUPP_CFG"
           echo "ASAN_OPTIONS = [$ASAN_OPTIONS]."
+          echo "LSAN_OPTIONS = [$LSAN_OPTIONS]."
         elif [ '${{ matrix.build-test-cfg.sanitizer-name }}' = ubsan ]; then
           export SAN_SUPP=1
           export SAN_SUPP_CFG=${{ github.workspace }}/install/${{ matrix.build-test-cfg.conan-profile-build-type }}/usr/local/bin/san_suppressions.cfg
@@ -571,6 +586,8 @@ jobs:
     # and will grow.  The techniques will still apply.
 
     - name: Run test/demo [NetFlow echo]
+      if: |
+        !cancelled()
       run: |
         # Run test/demo [NetFlow echo].
         cd ${{ env.install-dir }}/bin
@@ -611,13 +628,15 @@ jobs:
         mkdir -p $OUT_DIR
         SUPP_DIR_A=${{ github.workspace }}/src
         # As of this writing there are TSAN suppressions for this test specifically.  TODO: Revisit them; and then this.
+        # Update: Now there are also ASAN (LSAN) suppressions.  These are likely permanent as of this writing.
+        # Reminder: the following construction handles suppression file(s) from *any* relevant sanitizer type (if any).
         SUPP_DIR_OWN=${{ github.workspace }}/test/suite/unit_test
         { cat $SUPP_DIR_A/${{ env.san-suppress-cfg-in-file1 }} $SUPP_DIR_A/${{ env.san-suppress-cfg-in-file2 }} \
               $SUPP_DIR_OWN/${{ env.san-suppress-cfg-in-file1 }} $SUPP_DIR_OWN/${{ env.san-suppress-cfg-in-file2 }} \
             > ${{ env.san-suppress-cfg-file }} 2> /dev/null; } || true
         ${{ env.setup-tests-env }}
         ${{ env.setup-run-env }}
-        # Sensitive benchmarks in this setting should run and be warned about it they "fail," but they should not
+        # Sensitive benchmarks in this setting should run and be warned about, if they "fail," but they should not
         # fail the test.
         $RUN_IT --do-not-fail-benchmarks > $OUT_DIR/console.log 2>&1
 

src/CMakeLists.txt

@@ -79,6 +79,7 @@ set(SRCS
     flow/net_flow/server_socket.cpp
     flow/perf/checkpt_timer.cpp
     flow/perf/clock_type.cpp
+    flow/util/blob.cpp
     flow/util/detail/sched_task_handle_state.cpp
     flow/util/detail/util.cpp
     flow/util/sched_task.cpp
@@ -159,6 +160,7 @@ set(HDRS
     flow/util/basic_blob.hpp
     flow/util/blob.hpp
     flow/util/blob_fwd.hpp
+    flow/util/detail/linked_hash.hpp
     flow/util/detail/sched_task_handle_state.hpp
     flow/util/detail/util.hpp
     flow/util/detail/util_fwd.hpp
@@ -171,6 +173,7 @@ set(HDRS
     flow/util/shared_ptr_alias_holder.hpp
     flow/util/string_ostream.hpp
     flow/util/string_view.hpp
+    flow/util/thread_lcl.hpp
     flow/util/traits.hpp
     flow/util/uniq_id_holder.hpp
     flow/util/util.hpp

src/flow/async/async_fwd.hpp

@@ -104,7 +104,7 @@ using Task = Function<void ()>;
  * In addition, it is guaranteed that copying (via constructor or assignment) of async::Op is
  * has performance characteristics no worse than those of `shared_ptr`.  I.e., it is to be thought of as light-weight.
  *
- * The value `Op()` is designated as a null/sentinel value and must not be passed to Concurrent_task_loop::post()
+ * The value `Op{}` is designated as a null/sentinel value and must not be passed to Concurrent_task_loop::post()
  * or anything built on it.
  *
  * That's the formal definition.  We reiterate that copying these is cheap; and moreover two `Op`s such that

src/flow/async/concurrent_task_loop.cpp

@@ -210,9 +210,9 @@ void optimize_pinning_in_thread_pool(log::Logger* logger_ptr,
     const auto native_mach_thread_id = pthread_mach_thread_np(native_pthread_thread_id);
     if (native_pthread_thread_id == 0)
     {
-      const Error_code sys_err_code(errno, system_category()); // As above....
+      const Error_code sys_err_code{errno, system_category()}; // As above....
       FLOW_ERROR_SYS_ERROR_LOG_WARNING();
-      throw error::Runtime_error(sys_err_code, "pthread_mach_thread_np() call in optimize_pinning_in_thread_pool()");
+      throw error::Runtime_error{sys_err_code, "pthread_mach_thread_np() call in optimize_pinning_in_thread_pool()"};
     }
     // else
     FLOW_LOG_TRACE("pthread ID [" << native_pthread_thread_id << "] "
@@ -249,8 +249,8 @@ void optimize_pinning_in_thread_pool(log::Logger* logger_ptr,
        * @todo For sure though should use error::Runtime_error here, the ctor that takes no Error_code.
        * That ctor did not exist when the present code was written; as of this writing Flow is Linux-only.
        * Would do it right now but lack the time to verify any changes for Mac at the moment. */
-      throw runtime_error(ostream_op_string("[MACH_KERN_RETURN_T:", code,
-                                            "] [thread_policy_set(THREAD_AFFINITY_POLICY) failed]"));
+      throw runtime_error{ostream_op_string("[MACH_KERN_RETURN_T:", code,
+                                            "] [thread_policy_set(THREAD_AFFINITY_POLICY) failed]")};
     }
     // else OK!
 #  endif // if 0

src/flow/async/concurrent_task_loop.hpp

@@ -208,11 +208,11 @@ namespace flow::async
  *   flow::async::Concurrent_task_loop L;
  *   auto op J = L.create_op(); // ATTN! The syntax is different from Strands but the idea is identical.
  *   ...
- *   X_type X(L.task_engine());
+ *   X_type X{L.task_engine()};
  *   // ATTN! The syntax is again somewhat different from bind_executor(S, F), but the idea is equivalent.
  *   X.async_A(&A_target, A_settings, flow::async::asio_handler_via_op(&L, J, F));
  *   ...
- *   Y_type Y(L.task_engine());
+ *   Y_type Y{L.task_engine()};
  *   Y.async_B(&B_target, B_settings, flow::async::asio_handler_via_op(&L, J, G));
  *   // X.sync_A() and Y.sync_B() are executing in background; F and G will run on respective completion;
  *   // but F() and G() shall run non-concurrently by virtue of being wrapped by the same Op: J.
@@ -408,8 +408,8 @@ class Concurrent_task_loop :
    *        in each thread, for all `thread_idx` in [0, n_threads()).  start() will return no sooner than
    *        when each such callback has finished.
    */
-  virtual void start(Task&& init_task_or_empty = Task(),
-                     const Thread_init_func& thread_init_func_or_empty = Thread_init_func()) = 0;
+  virtual void start(Task&& init_task_or_empty = Task{},
+                     const Thread_init_func& thread_init_func_or_empty = Thread_init_func{}) = 0;
 
   /**
    * Waits for any ongoing task(s)/completion handler(s) to return; then prevents any further-queued such tasks
@@ -535,7 +535,7 @@ class Concurrent_task_loop :
    *
    * @param op
    *        The (presumably) multi-async-step operation to which `task` belongs, such that no `Task`s associated with
-   *        `op` may execute concurrently with `task`.  If `op.empty()` (a/k/a `op == Op()`, recalling that `Op()`
+   *        `op` may execute concurrently with `task`.  If `op.empty()` (a/k/a `op == Op{}`, recalling that `Op{}`
    *        is null/sentinel), then `assert()` trips.
    * @param task
    *        See other post().

src/flow/async/detail/task_qing_thread.cpp

@@ -58,7 +58,7 @@ Task_qing_thread::Task_qing_thread(flow::log::Logger* logger_ptr, util::String_v
   using Log_config = log::Config;
 
   assert(m_task_engine);
-  string nickname(nickname_view); // We need an std::string below anyway, so copy this now.
+  string nickname{nickname_view}; // We need an std::string below anyway, so copy this now.
 
   // Some programs start tons of threads.  Let's be stingy with INFO messages.
 
@@ -93,7 +93,7 @@ Task_qing_thread::Task_qing_thread(flow::log::Logger* logger_ptr, util::String_v
    * `sev_override == Sev::S_END_SENTINEL`; we need not even track it as a special case.) */
   const auto sev_override = *(Log_config::this_thread_verbosity_override());
 
-  m_worker_thread.reset(new Thread([this, // Valid throughout thread { body }.
+  m_worker_thread.reset(new Thread{[this, // Valid throughout thread { body }.
                                     sev_override,
                                     nickname = std::move(nickname), // Valid throughout thread { body }.
                                     init_func_or_empty = std::move(init_func_or_empty),
@@ -148,7 +148,7 @@ Task_qing_thread::Task_qing_thread(flow::log::Logger* logger_ptr, util::String_v
     } // const auto sev_override_auto = // Restore logging to normal (how it normally is at thread start).
 
     // Avoid loop, thread exiting when no pending tasks remain.
-    Task_engine_work avoid_task_engine_stop(make_work_guard(*m_task_engine));
+    Task_engine_work avoid_task_engine_stop{make_work_guard(*m_task_engine)};
 
     // Block -- wait for tasks to be posted on this thread's (possibly shared with other threads) Task_engine.
     m_task_engine->run();
@@ -230,7 +230,7 @@ Task_qing_thread::Task_qing_thread(flow::log::Logger* logger_ptr, util::String_v
      * trace to the logs as well.)  The answer is yes, though it's not on us to do it.  One should do such work either
      * in std::terminate() (by using std::set_terminate()) or, arguably even better, in a global SIGABRT handler.
      * I am only mentioning it here as opportunistic advice -- again, it's not in our purview, as shown above. */
-  })); // Thread body.
+  }}); // Thread body.
   // `nickname`, `init_task_or_empty` may now be hosed.
 
   if (done_promise_else_block)

src/flow/async/detail/task_qing_thread.hpp

@@ -176,8 +176,8 @@ class Task_qing_thread :
    */
   explicit Task_qing_thread(flow::log::Logger* logger_ptr, util::String_view nickname,
                             const Task_engine_ptr& task_engine, bool own_task_engine,
-                            boost::promise<void>* done_promise_else_block = 0,
-                            Task&& init_func_or_empty = Task());
+                            boost::promise<void>* done_promise_else_block = nullptr,
+                            Task&& init_func_or_empty = Task{});
 
   /**
    * stop(), followed by forgetting the `Task_engine` returned by task_engine(); the latter action may

src/flow/async/op.hpp

@@ -114,7 +114,7 @@ class Op_list :
    * @return See above.  Note the address stored in the returned *reference* is valid until destructor runs;
    *         hence it's not necessary (though cheap) to copy the `Op`.
    */
-  const Op& random_op(size_t* chosen_idx = 0) const;
+  const Op& random_op(size_t* chosen_idx = nullptr) const;
 
   /**
    * Returns a randomly selected index from range [O, size()).

src/flow/async/segregated_thread_task_loop.cpp

@@ -58,7 +58,7 @@ Segregated_thread_task_loop::Segregated_thread_task_loop(log::Logger* logger_ptr
   for (Task_engine_ptr& task_engine_ptr_in_container : m_task_engines)
   {
     // Attn: The concurrency-hint=1 may avoid or all most locking in boost.asio.  Exactly 1 thread in the Task_engine.
-    task_engine_ptr_in_container.reset(new Task_engine(1));
+    task_engine_ptr_in_container.reset(new Task_engine{1});
 
     /* Task_engine starts in !stopped() mode ready to run().  start() pre-condition is stopped() so for simplicity
      * start in the same state that our stop() would put the Task_engine into: */
@@ -68,9 +68,9 @@ Segregated_thread_task_loop::Segregated_thread_task_loop(log::Logger* logger_ptr
 
   // Initialize our Ops_list of pre-created Ops which in our case simply store all `n` `Task_engine_ptr`s.
   const size_t n = n_threads();
-  m_per_thread_ops.reset(new Op_list(get_logger(), n,
+  m_per_thread_ops.reset(new Op_list{get_logger(), n,
                                      [this](size_t idx) -> Op
-                                       { return Op(static_cast<Task_engine_ptr>(m_task_engines[idx])); }));
+                                       { return Op{static_cast<Task_engine_ptr>(m_task_engines[idx])}; }});
   /* (The static_cast<> is probably unnecessary but makes the compiler check our type logic for us.  That's quite
    * helpful in this rare situation where we're essentially using a dynamically typed variable in C++ [boost::any].
    * There is 0 perf cost to it by the way.) */
@@ -124,7 +124,7 @@ void Segregated_thread_task_loop::start(Task&& init_task_or_empty,
    * though, so let's keep to the letter of our contract.  Also, this way we can do it in parallel instead of
    * serially. */
 
-  vector<promise<void>> thread_init_done_promises(n);
+  vector<promise<void>> thread_init_done_promises{n};
   for (size_t idx = 0; idx != n; ++idx)
   {
     Task task_qing_thread_init_func;
@@ -147,11 +147,11 @@ void Segregated_thread_task_loop::start(Task&& init_task_or_empty,
     // Now its Task_qing_thread can do ->run() as most of its thread body (and it won't just return).
 
     // Create/start the thread.
-    m_qing_threads[idx].reset(new Task_qing_thread(get_logger(),
+    m_qing_threads[idx].reset(new Task_qing_thread{get_logger(),
                                                    (n == 1) ? m_nickname : util::ostream_op_string(m_nickname, idx),
                                                    task_engine, true, // Its *own* 1-1 Task_engine.
                                                    &(thread_init_done_promises[idx]),
-                                                   std::move(task_qing_thread_init_func)));
+                                                   std::move(task_qing_thread_init_func)});
   } // for (idx in [0, n))
   FLOW_LOG_INFO("All threads are asynchronously starting.  Awaiting their readiness barrier-style, in sequence.");
   for (size_t idx = 0; idx != n; ++idx)
@@ -167,7 +167,7 @@ void Segregated_thread_task_loop::start(Task&& init_task_or_empty,
   {
     FLOW_LOG_INFO("Thread count was auto-determined.  Further attempting thread-to-core scheduling optimization.");
 
-    vector<Thread*> worker_threads(n); // Initialized to nulls.  Now set them to the raw `Thread*`s.
+    vector<Thread*> worker_threads{n}; // Initialized to nulls.  Now set them to the raw `Thread*`s.
     transform(m_qing_threads.begin(), m_qing_threads.end(), worker_threads.begin(),
               [](const Task_qing_thread_ptr& qing_thread_ptr) -> Thread*
                 { return qing_thread_ptr->raw_worker_thread(); });

src/flow/async/segregated_thread_task_loop.hpp

@@ -114,8 +114,8 @@ class Segregated_thread_task_loop :
    * @param thread_init_func_or_empty
    *        See superclass API.
    */
-  void start(Task&& init_task_or_empty = Task(),
-             const Thread_init_func& thread_init_func_or_empty = Thread_init_func()) override;
+  void start(Task&& init_task_or_empty = Task{},
+             const Thread_init_func& thread_init_func_or_empty = Thread_init_func{}) override;
 
   /**
    * Implements superclass API.  In this implementation this essentially boils down to N `Task_engine::stop()`s,