intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

5 files changed, 266 insertions, 0 deletions

diff --git a/library/cpp/threading/equeue/equeue.cpp b/library/cpp/threading/equeue/equeue.cpp
new file mode 100644
index 0000000000..54a848e912
--- /dev/null
+++ b/library/cpp/threading/equeue/equeue.cpp
@@ -0,0 +1,80 @@
+#include "equeue.h"
+
+TElasticQueue::TElasticQueue(THolder<IThreadPool> slaveQueue)
+    : SlaveQueue_(std::move(slaveQueue))
+{
+}
+
+size_t TElasticQueue::ObjectCount() const {
+    return (size_t)AtomicGet(ObjectCount_);
+}
+
+bool TElasticQueue::TryIncCounter() {
+    if ((size_t)AtomicIncrement(GuardCount_) > MaxQueueSize_) {
+        AtomicDecrement(GuardCount_);
+        return false;
+    }
+
+    return true;
+}
+
+
+
+class TElasticQueue::TDecrementingWrapper: TNonCopyable, public IObjectInQueue {
+public:
+    TDecrementingWrapper(IObjectInQueue* realObject, TElasticQueue* queue)
+        : RealObject_(realObject)
+        , Queue_(queue)
+    {
+        AtomicIncrement(Queue_->ObjectCount_);
+    }
+
+    ~TDecrementingWrapper() override {
+        AtomicDecrement(Queue_->ObjectCount_);
+        AtomicDecrement(Queue_->GuardCount_);
+    }
+private:
+    void Process(void *tsr) override {
+        THolder<TDecrementingWrapper> self(this);
+        RealObject_->Process(tsr);
+    }
+private:
+    IObjectInQueue* const RealObject_;
+    TElasticQueue* const Queue_;
+};
+
+
+
+bool TElasticQueue::Add(IObjectInQueue* obj) {
+    if (!TryIncCounter()) {
+        return false;
+    }
+
+    THolder<TDecrementingWrapper> wrapper;
+    try {
+        wrapper.Reset(new TDecrementingWrapper(obj, this));
+    } catch (...) {
+        AtomicDecrement(GuardCount_);
+        throw;
+    }
+
+    if (SlaveQueue_->Add(wrapper.Get())) {
+        Y_UNUSED(wrapper.Release());
+        return true;
+    } else {
+        return false;
+    }
+}
+
+void TElasticQueue::Start(size_t threadCount, size_t maxQueueSize) {
+    MaxQueueSize_ = maxQueueSize;
+    SlaveQueue_->Start(threadCount, maxQueueSize);
+}
+
+void TElasticQueue::Stop() noexcept {
+    return SlaveQueue_->Stop();
+}
+
+size_t TElasticQueue::Size() const noexcept {
+    return SlaveQueue_->Size();
+}
diff --git a/library/cpp/threading/equeue/equeue.h b/library/cpp/threading/equeue/equeue.h
new file mode 100644
index 0000000000..40dd342585
--- /dev/null
+++ b/library/cpp/threading/equeue/equeue.h
@@ -0,0 +1,28 @@
+#pragma once
+
+#include <util/thread/pool.h>
+#include <util/system/atomic.h>
+#include <util/generic/ptr.h>
+
+//actual queue limit will be (maxQueueSize - numBusyThreads) or 0
+class TElasticQueue: public IThreadPool {
+public:
+    explicit TElasticQueue(THolder<IThreadPool> slaveQueue);
+
+    bool Add(IObjectInQueue* obj) override;
+    size_t Size() const noexcept override;
+
+    void Start(size_t threadCount, size_t maxQueueSize) override;
+    void Stop() noexcept override;
+
+    size_t ObjectCount() const;
+private:
+    class TDecrementingWrapper;
+
+    bool TryIncCounter();
+private:
+    THolder<IThreadPool> SlaveQueue_;
+    size_t MaxQueueSize_ = 0;
+    TAtomic ObjectCount_ = 0;
+    TAtomic GuardCount_ = 0;
+};
diff --git a/library/cpp/threading/equeue/equeue_ut.cpp b/library/cpp/threading/equeue/equeue_ut.cpp
new file mode 100644
index 0000000000..9cf2aced44
--- /dev/null
+++ b/library/cpp/threading/equeue/equeue_ut.cpp
@@ -0,0 +1,125 @@
+#include "equeue.h"
+
+#include <library/cpp/testing/unittest/registar.h>
+
+#include <util/system/event.h>
+#include <util/datetime/base.h>
+#include <util/generic/vector.h>
+
+Y_UNIT_TEST_SUITE(TElasticQueueTest) {
+    const size_t MaxQueueSize = 20;
+    const size_t ThreadCount = 10;
+    const size_t N = 100000;
+
+    static THolder<TElasticQueue> Queue;
+
+    struct TQueueSetup {
+        TQueueSetup() {
+            Queue.Reset(new TElasticQueue(MakeHolder<TSimpleThreadPool>()));
+            Queue->Start(ThreadCount, MaxQueueSize);
+        }
+        ~TQueueSetup() {
+            Queue->Stop();
+        }
+    };
+
+    struct TCounters {
+        void Reset() {
+            Processed = Scheduled = Discarded = Total = 0;
+        }
+
+        TAtomic Processed;
+        TAtomic Scheduled;
+        TAtomic Discarded;
+        TAtomic Total;
+    };
+    static TCounters Counters;
+
+//fill test -- fill queue with "endless" jobs
+    TSystemEvent WaitEvent;
+    Y_UNIT_TEST(FillTest) {
+        Counters.Reset();
+
+        struct TWaitJob: public IObjectInQueue {
+            void Process(void*) override {
+                WaitEvent.Wait();
+                AtomicIncrement(Counters.Processed);
+            }
+        } job;
+
+        struct TLocalSetup: TQueueSetup {
+            ~TLocalSetup() {
+                WaitEvent.Signal();
+            }
+        };
+
+        size_t enqueued = 0;
+        {
+            TLocalSetup setup;
+            while (Queue->Add(&job) && enqueued < MaxQueueSize + 100) {
+                ++enqueued;
+            }
+
+            UNIT_ASSERT_VALUES_EQUAL(enqueued, MaxQueueSize);
+            UNIT_ASSERT_VALUES_EQUAL(enqueued, Queue->ObjectCount());
+        }
+
+        UNIT_ASSERT_VALUES_EQUAL(0u, Queue->ObjectCount());
+        UNIT_ASSERT_VALUES_EQUAL(0u, Queue->Size());
+        UNIT_ASSERT_VALUES_EQUAL((size_t)Counters.Processed, enqueued);
+    }
+
+
+//concurrent test -- send many jobs from different threads
+    struct TJob: public IObjectInQueue {
+        void Process(void*) override {
+            AtomicIncrement(Counters.Processed);
+        };
+    };
+    static TJob Job;
+
+    static bool TryAdd() {
+        AtomicIncrement(Counters.Total);
+        if (Queue->Add(&Job)) {
+            AtomicIncrement(Counters.Scheduled);
+            return true;
+        } else {
+            AtomicIncrement(Counters.Discarded);
+            return false;
+        }
+    }
+
+    static size_t TryCounter;
+
+    Y_UNIT_TEST(ConcurrentTest) {
+        Counters.Reset();
+        TryCounter = 0;
+
+        struct TSender: public IThreadFactory::IThreadAble {
+            void DoExecute() override {
+                while ((size_t)AtomicIncrement(TryCounter) <= N) {
+                    if (!TryAdd()) {
+                        Sleep(TDuration::MicroSeconds(50));
+                    }
+                }
+            }
+        } sender;
+
+        {
+            TQueueSetup setup;
+
+            TVector< TAutoPtr<IThreadFactory::IThread> > senders;
+            for (size_t i = 0; i < ThreadCount; ++i) {
+                senders.push_back(::SystemThreadFactory()->Run(&sender));
+            }
+
+            for (size_t i = 0; i < senders.size(); ++i) {
+                senders[i]->Join();
+            }
+        }
+
+        UNIT_ASSERT_VALUES_EQUAL((size_t)Counters.Total, N);
+        UNIT_ASSERT_VALUES_EQUAL(Counters.Processed, Counters.Scheduled);
+        UNIT_ASSERT_VALUES_EQUAL(Counters.Total, Counters.Scheduled + Counters.Discarded);
+    }
+}
diff --git a/library/cpp/threading/equeue/ut/ya.make b/library/cpp/threading/equeue/ut/ya.make
new file mode 100644
index 0000000000..2f6293d47d
--- /dev/null
+++ b/library/cpp/threading/equeue/ut/ya.make
@@ -0,0 +1,18 @@
+UNITTEST()
+
+OWNER(
+    g:base
+    g:middle
+)
+
+PEERDIR(
+    ADDINCL library/cpp/threading/equeue
+)
+
+SRCDIR(library/cpp/threading/equeue)
+
+SRCS(
+    equeue_ut.cpp
+)
+
+END()
diff --git a/library/cpp/threading/equeue/ya.make b/library/cpp/threading/equeue/ya.make
new file mode 100644
index 0000000000..314f4d3c86
--- /dev/null
+++ b/library/cpp/threading/equeue/ya.make
@@ -0,0 +1,15 @@
+LIBRARY()
+
+OWNER(
+    g:base
+    g:middle
+    ironpeter
+    mvel
+)
+
+SRCS(
+    equeue.h
+    equeue.cpp
+)
+
+END()