Merge branch 'main' into it-legacy-to-fsm-switch-via-quorum1

Author: bbpetukhov

src/groups/bmq/bmqc/bmqc_orderedhashmapwithhistory.cpp

@@ -20,5 +20,13 @@
 namespace BloombergLP {
 namespace bmqc {
 
+// -------------------------------------------
+// struct OrderedHashMapWithHistory_ImpDetails
+// -------------------------------------------
+
+const int
+    OrderedHashMapWithHistory_ImpDetails::k_INSERT_GC_MESSAGES_BATCH_SIZE =
+        1000;
+
 }  // close package namespace
 }  // close enterprise namespace

src/groups/bmq/bmqc/bmqc_orderedhashmapwithhistory.h

@@ -55,6 +55,19 @@
 namespace BloombergLP {
 namespace bmqc {
 
+// ===========================================
+// struct OrderedHashMapWithHistory_ImpDetails
+// ===========================================
+
+/// PRIVATE CLASS.
+// For use only by `bmqc::OrderedHashMapWithHistory` implementation.
+struct OrderedHashMapWithHistory_ImpDetails {
+    // PRIVATE CLASS DATA
+    /// How many messages to GC when GC required in
+    /// `bmqc::OrderedHashMapWithHistory::insert`
+    static const int k_INSERT_GC_MESSAGES_BATCH_SIZE;
+};
+
 // ========================================
 // class OrderedHashMapWithHistory_Iterator
 // ========================================
@@ -219,7 +232,27 @@ class OrderedHashMapWithHistory {
 
     size_t d_historySize;  // how many historical (!d_isLive) items
 
-    gc_iterator d_gcIt;  // where to start 'gc'
+    /// Whether this container has more elements to `gc`.  This flag might be
+    /// set or unset during every `gc` call according to this container's
+    /// needs.
+    bool d_requireGC;
+
+    /// The `now` time of the last GC.  We assume that the current actual time
+    /// is no less than this timestamp.
+    TimeType d_lastGCTime;
+
+    /// The iterator pointing to the element where garbage collection should
+    /// continue once `gc` is called.  According to contract, this iterator
+    /// only goes forward.  All the elements passed by this iterator are either
+    /// removed or marked for removal, depending on what happened first:
+    /// - If the element was not erased by the user before, but its timeout
+    /// happened in this container, it is marked for deletion in `gc` and
+    /// iterator goes forward.  Next, it is the user's responsibility to call
+    /// `erase` on this element to fully remove it.
+    /// - If the user removes the element before its timeout happened, the
+    /// element becomes `not alive`, but still lives in the history.
+    /// Eventually `gc` reaches this element and fully removes it.
+    gc_iterator d_gcIt;
 
     // PRIVATE CLASS METHODS
     static const KEY& get_key(const bsl::pair<const KEY, VALUE>& value)
@@ -492,6 +525,8 @@ inline OrderedHashMapWithHistory<KEY, VALUE, HASH, VALUE_TYPE>::
 , d_first(d_impl.end())
 , d_last(d_impl.end())
 , d_historySize(0)
+, d_requireGC(false)
+, d_lastGCTime(0)
 , d_gcIt(endGc())
 {
     // NOTHING
@@ -521,6 +556,8 @@ inline void OrderedHashMapWithHistory<KEY, VALUE, HASH, VALUE_TYPE>::clear()
 
     d_first = d_last = end();
     d_gcIt           = endGc();
+    d_requireGC      = false;
+    d_lastGCTime     = 0;
     d_historySize    = 0;
 }
 
@@ -623,8 +660,15 @@ OrderedHashMapWithHistory<KEY, VALUE, HASH, VALUE_TYPE>::insert(
     const SOURCE_TYPE& value,
     TimeType           timePoint)
 {
-    TimeType                     time = d_timeout ? timePoint + d_timeout : 0;
+    TimeType time = d_timeout ? timePoint + d_timeout : 0;
+    if (d_requireGC) {
+        gc(bsl::max(timePoint, d_lastGCTime),
+           OrderedHashMapWithHistory_ImpDetails::
+               k_INSERT_GC_MESSAGES_BATCH_SIZE);
+    }
+
     bsl::pair<gc_iterator, bool> result = d_impl.insert(Value(value, time));
+
     // No need to keep track of element's timePoint if the map is not
     // maintaining any history (i.e., if d_timeout == 0).
 
@@ -668,7 +712,9 @@ OrderedHashMapWithHistory<KEY, VALUE, HASH, VALUE_TYPE>::gc(TimeType now,
     // Try to advance to either a young item, or to the end of the batch, or to
     // the end of collection.
 
-    if (d_gcIt == endGc()) {
+    d_lastGCTime = now;
+    if (BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY(d_gcIt == endGc())) {
+        BSLS_PERFORMANCEHINT_UNLIKELY_HINT;
         d_gcIt = beginGc();
     }
     const int initialHistorySize = d_historySize;
@@ -681,26 +727,30 @@ OrderedHashMapWithHistory<KEY, VALUE, HASH, VALUE_TYPE>::gc(TimeType now,
         }
         gc_iterator it = d_gcIt++;
         if (it->d_isLive) {
-            // This is an old item.  It should be removed by 'erase'.
-            // No need to return to this item.  No need to check time again.
-            // Indicate that it needs to be removed by setting its time to 0.
+            // This item was not erased by the user yet, but its timeout in
+            // this container happened.  Mark it for deletion by setting
+            // `d_time` to 0, so the next time user calls `erase` on it, it
+            // will be fully removed.
             it->d_time = 0;
         }
         else {
+            // This item was erased by the user before, and we can fully remove
+            // it right here.
             d_impl.erase(it);
             --d_historySize;
         }
         // Meaning, there is no need for 'd_gcIt' to step back.  Only forward.
 
         if (--batchSize == 0) {
             // Remember where we have stopped and resume from there next time
-
+            d_requireGC             = true;
             const int historyChange = initialHistorySize - d_historySize;
             // Note that we return either a negative value or 0 here:
             return -historyChange;  // RETURN
         }
     }
 
+    d_requireGC             = false;
     const int historyChange = initialHistorySize - d_historySize;
     // Note that we return either a positive value or 0 here:
     return historyChange;

src/groups/bmq/bmqt/bmqt_uri.t.cpp

@@ -22,17 +22,23 @@
 // BDE
 #include <ball_log.h>
 #include <bdlf_bind.h>
+#include <bdlma_localsequentialallocator.h>
 #include <bsl_string.h>
 #include <bsl_utility.h>
 #include <bsl_vector.h>
 #include <bslh_defaulthashalgorithm.h>
 #include <bslh_hash.h>
 #include <bslmt_barrier.h>
+#include <bslmt_latch.h>
 #include <bslmt_threadgroup.h>
 
 // TEST DRIVER
 #include <bmqtst_testhelper.h>
-#include <bsl_functional.h>
+
+// BENCHMARKING LIBRARY
+#ifdef BMQTST_BENCHMARK_ENABLED
+#include <benchmark/benchmark.h>
+#endif
 
 // CONVENIENCE
 using namespace BloombergLP;
@@ -647,6 +653,123 @@ static void test7_testLongUri()
     bmqt::UriParser::shutdown();
 }
 
+#ifdef BMQTST_BENCHMARK_ENABLED
+
+struct UriParserBenchmark {
+    static void bench(bslmt::Latch*   initLatch_p,
+                      bslmt::Barrier* startBarrier_p,
+                      bslmt::Latch*   finishLatch_p)
+    {
+        // PRECONDITIONS
+        BSLS_ASSERT_OPT(initLatch_p);
+        BSLS_ASSERT_OPT(startBarrier_p);
+        BSLS_ASSERT_OPT(finishLatch_p);
+
+        const size_t      k_NUM_ITERATIONS = 100000;
+        const bsl::string k_SAMPLE_URI(
+            "bmq://my.sample.domain.~dev/my-queue-name?id=consumer123",
+            bmqtst::TestHelperUtil::allocator());
+
+        bmqt::Uri   uri(bmqtst::TestHelperUtil::allocator());
+        bsl::string error(bmqtst::TestHelperUtil::allocator());
+
+        initLatch_p->arrive();
+        startBarrier_p->wait();
+
+        for (size_t i = 0; i < k_NUM_ITERATIONS; ++i) {
+            bmqt::UriParser::parse(&uri, &error, k_SAMPLE_URI);
+        }
+
+        finishLatch_p->arrive();
+    }
+};
+
+struct UriConstructorBenchmark {
+    static void bench(bslmt::Latch*   initLatch_p,
+                      bslmt::Barrier* startBarrier_p,
+                      bslmt::Latch*   finishLatch_p)
+    {
+        // PRECONDITIONS
+        BSLS_ASSERT_OPT(initLatch_p);
+        BSLS_ASSERT_OPT(startBarrier_p);
+        BSLS_ASSERT_OPT(finishLatch_p);
+
+        const size_t      k_NUM_ITERATIONS = 100000;
+        const bsl::string k_SAMPLE_URI(
+            "bmq://my.sample.domain.~dev/my-queue-name?id=consumer123",
+            bmqtst::TestHelperUtil::allocator());
+
+        // Test allocator is slow and might skew the benchmarks
+        bdlma::LocalSequentialAllocator<256> lsa(
+            bmqtst::TestHelperUtil::allocator());
+
+        initLatch_p->arrive();
+        startBarrier_p->wait();
+
+        for (size_t i = 0; i < k_NUM_ITERATIONS; ++i) {
+            bmqt::Uri uri(k_SAMPLE_URI, &lsa);
+            (void)uri;
+        }
+
+        finishLatch_p->arrive();
+    }
+};
+
+template <size_t NUM_THREADS, typename BENCHMARK>
+static void testN1_benchmark(benchmark::State& state)
+// ------------------------------------------------------------------------
+// URI PERFORMANCE TEST
+//
+// Plan: spawn NUM_THREADS and measure the time taken for BENCHMARK::bench
+//
+// Testing:
+//  Performance
+// ------------------------------------------------------------------------
+{
+    bmqtst::TestHelper::printTestName("URI PERFORMANCE TEST");
+
+    bmqt::UriParser::initialize(bmqtst::TestHelperUtil::allocator());
+
+    bslmt::Latch   initThreadLatch(NUM_THREADS);
+    bslmt::Barrier startBenchmarkBarrier(NUM_THREADS + 1);
+    bslmt::Latch   finishBenchmarkLatch(NUM_THREADS);
+
+    bslmt::ThreadGroup threadGroup(bmqtst::TestHelperUtil::allocator());
+    for (size_t i = 0; i < NUM_THREADS; ++i) {
+        const int rc = threadGroup.addThread(
+            bdlf::BindUtil::bindS(bmqtst::TestHelperUtil::allocator(),
+                                  &(BENCHMARK::bench),
+                                  &initThreadLatch,
+                                  &startBenchmarkBarrier,
+                                  &finishBenchmarkLatch));
+        BMQTST_ASSERT_EQ_D(i, rc, 0);
+    }
+
+    initThreadLatch.wait();
+
+    size_t iter = 0;
+    for (auto _ : state) {
+        // Benchmark time start
+
+        // We don't support running multi-iteration benchmarks because we
+        // prepare and start complex tasks in separate threads.
+        // Once these tasks are finished, we cannot simply re-run them without
+        // reinitialization, and it goes against benchmark library design.
+        // Make sure we run this only once.
+        BSLS_ASSERT_OPT(0 == iter++ && "Must be run only once");
+
+        startBenchmarkBarrier.wait();
+        finishBenchmarkLatch.wait();
+
+        // Benchmark time end
+    }
+
+    threadGroup.joinAll();
+    bmqt::UriParser::shutdown();
+}
+
+#endif  // BMQTST_BENCHMARK_ENABLED
+
 // ============================================================================
 //                                 MAIN PROGRAM
 // ----------------------------------------------------------------------------
@@ -664,6 +787,50 @@ int main(int argc, char* argv[])
     case 3: test3_URIBuilderMultiThreaded(); break;
     case 2: test2_URIBuilder(); break;
     case 1: test1_breathingTest(); break;
+    case -1: {
+#ifdef BMQTST_BENCHMARK_ENABLED
+        BENCHMARK(testN1_benchmark<1, UriParserBenchmark>)
+            ->Name("bmqt::UriParser::parse threads=1")
+            ->Iterations(1)
+            ->Unit(benchmark::kMillisecond);
+        BENCHMARK(testN1_benchmark<2, UriParserBenchmark>)
+            ->Name("bmqt::UriParser::parse threads=2")
+            ->Iterations(1)
+            ->Unit(benchmark::kMillisecond);
+        BENCHMARK(testN1_benchmark<4, UriParserBenchmark>)
+            ->Name("bmqt::UriParser::parse threads=4")
+            ->Iterations(1)
+            ->Unit(benchmark::kMillisecond);
+        BENCHMARK(testN1_benchmark<8, UriParserBenchmark>)
+            ->Name("bmqt::UriParser::parse threads=8")
+            ->Iterations(1)
+            ->Unit(benchmark::kMillisecond);
+
+        BENCHMARK(testN1_benchmark<1, UriConstructorBenchmark>)
+            ->Name("bmqt::Uri::Uri         threads=1")
+            ->Iterations(1)
+            ->Unit(benchmark::kMillisecond);
+        BENCHMARK(testN1_benchmark<2, UriConstructorBenchmark>)
+            ->Name("bmqt::Uri::Uri         threads=2")
+            ->Iterations(1)
+            ->Unit(benchmark::kMillisecond);
+        BENCHMARK(testN1_benchmark<4, UriConstructorBenchmark>)
+            ->Name("bmqt::Uri::Uri         threads=4")
+            ->Iterations(1)
+            ->Unit(benchmark::kMillisecond);
+        BENCHMARK(testN1_benchmark<8, UriConstructorBenchmark>)
+            ->Name("bmqt::Uri::Uri         threads=8")
+            ->Iterations(1)
+            ->Unit(benchmark::kMillisecond);
+
+        benchmark::Initialize(&argc, argv);
+        benchmark::RunSpecifiedBenchmarks();
+#else
+        cerr << "WARNING: BENCHMARK '" << _testCase
+             << "' IS NOT SUPPORTED ON THIS PLATFORM." << endl;
+        bmqtst::TestHelperUtil::testStatus() = -1;
+#endif  // BMQTST_BENCHMARK_ENABLED
+    } break;
     default: {
         cerr << "WARNING: CASE '" << _testCase << "' NOT FOUND." << endl;
         bmqtst::TestHelperUtil::testStatus() = -1;

src/groups/mqb/mqbblp/mqbblp_cluster.cpp

@@ -636,7 +636,7 @@ void Cluster::initiateShutdownDispatched(const VoidFunctor& callback)
     d_clusterData.membership().setSelfNodeStatus(
         bmqp_ctrlmsg::NodeStatus::E_STOPPING);
 
-    d_clusterOrchestrator.queueHelper().requestToStopPushing();
+    d_clusterOrchestrator.queueHelper().requestToStopQueues();
 
     bsls::TimeInterval whenToStop(
         bsls::SystemTime::now(bsls::SystemClockType::e_MONOTONIC));

src/groups/mqb/mqbblp/mqbblp_clusterproxy.cpp

@@ -170,7 +170,7 @@ void ClusterProxy::initiateShutdownDispatched(const VoidFunctor& callback)
     // Mark self as stopping.
     d_isStopping = true;
 
-    d_queueHelper.requestToStopPushing();
+    d_queueHelper.requestToStopQueues();
     // 'checkUnconfirmedV2' serves as synchronization.
     // It makes sure stopPushing() gets executed before the return.
     bsls::TimeInterval whenToStop(

src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.cpp

@@ -5296,9 +5296,7 @@ void ClusterQueueHelper::processShutdownEvent()
     }
 }
 
-/// Stop sending PUSHes but continue receiving CONFIRMs, receiving and
-/// sending PUTs and ACKs.
-void ClusterQueueHelper::requestToStopPushing()
+void ClusterQueueHelper::requestToStopQueues()
 {
     // executed by the cluster *DISPATCHER* thread
 
@@ -5309,7 +5307,7 @@ void ClusterQueueHelper::requestToStopPushing()
     // Assume Shutdown V2
     d_isShutdownLogicOn = true;
 
-    // Prevent future queue operations from sending PUSHes.
+    // Prevent future queue operations from sending PUSHes and GC.
     for (QueueContextMapIter it = d_queues.begin(); it != d_queues.end();
          ++it) {
         QueueContextSp& queueContextSp = it->second;
@@ -5321,7 +5319,7 @@ void ClusterQueueHelper::requestToStopPushing()
         }
 
         queue->dispatcher()->execute(
-            bdlf::BindUtil::bind(&mqbi::Queue::stopPushing, queue),
+            bdlf::BindUtil::bind(&mqbi::Queue::setStopping, queue),
             queue);
     }
 }