High-performance hub component for dealing with many sockets and high throughput

include/nmranet_config.h

@@ -128,6 +128,14 @@ DECLARE_CONST(gridconnect_buffer_delay_usec);
  * two threads per client (multi-threaded) execution model. */
 DECLARE_CONST(gridconnect_tcp_use_select);
 
+/// Maximum number of packets to parse from a single DirectHubPort before we
+/// wait for data to drain from the system.
+DECLARE_CONST(directhub_port_max_incoming_packets);
+
+/// Number of bytes that we will be reading in one go from an incoming port. We
+/// will allocate at least this many bytes dedicated for each input port.
+DECLARE_CONST(directhub_port_incoming_buffer_size);
+
 /** Number of entries in the remote alias cache */
 DECLARE_CONST(remote_alias_cache_size);
 

src/executor/AsyncNotifiableBlock.cxx

@@ -0,0 +1,68 @@
+/** \copyright
+ * Copyright (c) 2020, Balazs Racz
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are  permitted provided that the following conditions are met:
+ *
+ *  - Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer.
+ *
+ *  - Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ *
+ * \file AsyncNotifiableBlock.cxx
+ *
+ * An advanced notifiable construct that acts as a fixed pool of
+ * BarrierNotifiables. A stateflow can pend on acquiring one of them, use that
+ * barrier, with it automatically returning to the next caller when the Barrier
+ * goes out of counts.
+ *
+ * @author Balazs Racz
+ * @date 18 Feb 2020
+ */
+
+#ifndef _DEFAULT_SOURCE
+#define _DEFAULT_SOURCE
+#endif
+
+#include "AsyncNotifiableBlock.hxx"
+
+#include "os/sleep.h"
+
+AsyncNotifiableBlock::~AsyncNotifiableBlock()
+{
+    // Recollects all notifiable instances, including waiting as long as needed
+    // if there are some that have not finished yet.
+    for (unsigned i = 0; i < count_; ++i)
+    {
+        while (true)
+        {
+            QMember *m = next().item;
+            if (!m)
+            {
+                LOG(VERBOSE,
+                    "shutdown async notifiable block: waiting for returns");
+                microsleep(100);
+            }
+            else
+            {
+                HASSERT(initialize(m)->abort_if_almost_done());
+                break;
+            }
+        }
+    }
+}

src/executor/AsyncNotifiableBlock.cxxtest

@@ -0,0 +1,110 @@
+#include "executor/AsyncNotifiableBlock.hxx"
+
+#include "utils/test_main.hxx"
+
+class AsyncNotifiableBlockTest : public ::testing::Test
+{
+protected:
+    AsyncNotifiableBlock b_ {2};
+};
+
+TEST_F(AsyncNotifiableBlockTest, create)
+{
+}
+
+TEST_F(AsyncNotifiableBlockTest, count_request_release)
+{
+    EXPECT_EQ(2u, b_.pending());
+    QMember *e = b_.next(0);
+    EXPECT_NE(nullptr, e);
+    EXPECT_EQ(1u, b_.pending());
+
+    QMember *f = b_.next(0);
+    EXPECT_NE(nullptr, f);
+    EXPECT_EQ(0u, b_.pending());
+
+    QMember *g = b_.next(0);
+    EXPECT_EQ(nullptr, g);
+    EXPECT_EQ(0u, b_.pending());
+
+    b_.initialize(e)->notify();
+    EXPECT_EQ(1u, b_.pending());
+
+    QMember *h = b_.next(0);
+    EXPECT_EQ(e, h);
+
+    EXPECT_EQ(0u, b_.pending());
+
+    b_.initialize(f)->notify();
+    b_.initialize(h)->notify();
+}
+
+TEST_F(AsyncNotifiableBlockTest, barrier_semantics)
+{
+    EXPECT_EQ(2u, b_.pending());
+    QMember *e = b_.next(0);
+    BarrierNotifiable *bn = b_.initialize(e);
+    EXPECT_EQ(1u, b_.pending());
+
+    bn->new_child();
+    bn->notify();
+    EXPECT_EQ(1u, b_.pending());
+    bn->notify();
+    EXPECT_EQ(2u, b_.pending());
+}
+
+class FakeExecutable : public Executable
+{
+public:
+    void run() override
+    {
+        DIE("unexpected.");
+    }
+
+    void alloc_result(QMember *m) override
+    {
+        ASSERT_TRUE(m);
+        m_ = m;
+    }
+
+    QMember *m_ {nullptr};
+};
+
+TEST_F(AsyncNotifiableBlockTest, async_allocation)
+{
+    EXPECT_EQ(2u, b_.pending());
+    QMember *e = b_.next(0);
+    EXPECT_NE(nullptr, e);
+    EXPECT_EQ(1u, b_.pending());
+
+    FakeExecutable cli1, cli2, cli3;
+    EXPECT_EQ(nullptr, cli1.m_);
+    EXPECT_EQ(nullptr, cli2.m_);
+    EXPECT_EQ(nullptr, cli3.m_);
+
+    b_.next_async(&cli1);
+    EXPECT_EQ(0u, b_.pending());
+    EXPECT_NE(nullptr, cli1.m_);
+    EXPECT_NE(e, cli1.m_);
+
+    b_.next_async(&cli2);
+    b_.next_async(&cli3);
+    EXPECT_EQ(nullptr, cli2.m_);
+    EXPECT_EQ(nullptr, cli3.m_);
+    EXPECT_EQ(0u, b_.pending());
+
+    b_.initialize(e)->notify(); // will be handed out to cli2
+
+    EXPECT_EQ(0u, b_.pending());
+    EXPECT_EQ(e, cli2.m_);
+
+    b_.initialize(cli1.m_)->notify(); // will be handed out to cli3
+    EXPECT_EQ(cli1.m_, cli3.m_);
+    EXPECT_EQ(0u, b_.pending());
+
+    b_.initialize(cli3.m_)->notify(); // will be handed back
+    EXPECT_EQ(1u, b_.pending());
+
+    b_.initialize(cli2.m_)->notify(); // will be handed back
+    EXPECT_EQ(2u, b_.pending());
+}

src/executor/AsyncNotifiableBlock.hxx

@@ -0,0 +1,139 @@
+/** \copyright
+ * Copyright (c) 2013, Balazs Racz
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are  permitted provided that the following conditions are met:
+ *
+ *  - Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer.
+ *
+ *  - Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ *
+ * \file AsyncNotifiableBlock.hxx
+ *
+ * An advanced notifiable construct that acts as a fixed pool of
+ * BarrierNotifiables. A stateflow can pend on acquiring one of them, use that
+ * barrier, with it automatically returning to the next caller when the Barrier
+ * goes out of counts.
+ *
+ * @author Balazs Racz
+ * @date 18 Feb 2020
+ */
+
+#ifndef _EXECUTOR_ASYNCNOTIFIABLEBLOCK_HXX_
+#define _EXECUTOR_ASYNCNOTIFIABLEBLOCK_HXX_
+
+#include <memory>
+
+#include "executor/Notifiable.hxx"
+#include "utils/Queue.hxx"
+#include "utils/logging.h"
+
+#include "utils/Buffer.hxx"
+
+/// A block of BarrierNotifiable objects, with an asynchronous allocation
+/// call. Caller StateFlows can block on allocating a new entry, and then get
+/// back a fresh BarrierNotifiable, which, upon being released will
+/// automatically be reallocated to a waiting flow, if any.
+class AsyncNotifiableBlock : private Notifiable, public QAsync
+{
+private:
+    /// Notifiable class that can act as a BarrierNotifiable but also be
+    /// enlisted in a queue.
+    class QueuedBarrier : public BarrierNotifiable, public QMember
+    {
+    public:
+        /// Notification implementation.
+        ///
+        /// Theory of operation: If this was the last notification (count goes
+        /// from 1 to 0), we take the done_ pointer, cast it to the owning
+        /// AsyncNotifiableBlock, and release outselves into the queue
+        /// there. We keep the count at 1 at all times, which ensures that the
+        /// done_ pointer remains pointing to the owner AsyncNotifiableBlock.
+        void notify() override
+        {
+            AtomicHolder h(this);
+            if (count_ == 1)
+            {
+                LOG(VERBOSE, "block notifiable %p returned pool size %u",
+                    (BarrierNotifiable *)this,
+                    (unsigned)mainBufferPool->total_size());
+                auto *tgt = static_cast<AsyncNotifiableBlock *>(done_);
+                tgt->insert(this);
+            }
+            else
+            {
+                --count_;
+            }
+        }
+
+        /// Checks that there is exactly one count in here.
+        void check_one_count()
+        {
+            HASSERT(count_ == 1);
+        }
+    };
+
+public:
+    /// Constructor. @param num_parallelism tells how many BarrierNotifiables
+    /// we should have and hand out to callers requesting them.
+    AsyncNotifiableBlock(unsigned num_parallelism)
+        : count_(num_parallelism)
+        , barriers_(new QueuedBarrier[num_parallelism])
+    {
+        for (unsigned i = 0; i < num_parallelism; ++i)
+        {
+            barriers_[i].reset(this);
+            this->insert(&barriers_[i]);
+        }
+    }
+
+    /// Destructor.
+    ~AsyncNotifiableBlock();
+
+    /// Turns an allocated entry from the QAsync into a usable
+    /// BarrierNotifiable.
+    /// @param entry a QMember that was allocated from *this.
+    /// @return an initialized BarrierNotifiable with exactly one count, and
+    /// done_ set up to be returned for further use.
+    BarrierNotifiable *initialize(QMember *entry)
+    {
+        QueuedBarrier *b = static_cast<QueuedBarrier *>(entry);
+        // We must be owning this entry.
+        HASSERT(barriers_.get() <= b);
+        HASSERT(b <= (barriers_.get() + count_));
+        b->check_one_count();
+        return b;
+    }
+
+    /// Notification implementation -- should never be called.
+    void notify() override
+    {
+        DIE("Should not receive this notification");
+    }
+
+private:
+    /// How many barriers do we have.
+    unsigned count_;
+    /// The pointer to the block of barriernotifiables.
+    std::unique_ptr<QueuedBarrier[]> barriers_;
+
+    DISALLOW_COPY_AND_ASSIGN(AsyncNotifiableBlock);
+};
+
+#endif // _EXECUTOR_ASYNCNOTIFIABLEBLOCK_HXX_

src/executor/Notifiable.hxx

@@ -171,7 +171,7 @@ private:
 /// A BarrierNotifiable allows to create a number of child Notifiable and wait
 /// for all of them to finish. When the last one is finished, the parent done
 /// callback is called.
-class BarrierNotifiable : public Notifiable, private Atomic
+class BarrierNotifiable : public Notifiable, protected Atomic
 {
 public:
     /** Constructs a barrier notifiable that is done. Users should call reset()
@@ -240,7 +240,7 @@ public:
         }
     }
 
-private:
+protected:
     /// How many outstanding notifications we are still waiting for. When 0,
     /// the barrier is not live; when reaches zero, done_ will be called.
     unsigned count_;

src/executor/sources

@@ -3,6 +3,7 @@ VPATH := $(SRCDIR)
 CSRCS += 
 
 CXXSRCS += \
+        AsyncNotifiableBlock.cxx \
         Executor.cxx \
         Notifiable.cxx \
         Service.cxx \