Restoring authorship annotation for <kulikov@yandex-team.ru>. Commit 2 of 2.

3 files changed, 214 insertions, 214 deletions

diff --git a/library/cpp/threading/equeue/equeue.cpp b/library/cpp/threading/equeue/equeue.cpp
index aaec19daa6..54a848e912 100644
--- a/library/cpp/threading/equeue/equeue.cpp
+++ b/library/cpp/threading/equeue/equeue.cpp
@@ -1,80 +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_); 
-    } 
- 
+#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())) { 
+        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); 
-} 
- 
+        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(); 
-} 
- 
+    return SlaveQueue_->Stop();
+}
+
 size_t TElasticQueue::Size() const noexcept {
-    return SlaveQueue_->Size(); 
-} 
+    return SlaveQueue_->Size();
+}
diff --git a/library/cpp/threading/equeue/equeue.h b/library/cpp/threading/equeue/equeue.h
index 403e993713..40dd342585 100644
--- a/library/cpp/threading/equeue/equeue.h
+++ b/library/cpp/threading/equeue/equeue.h
@@ -1,28 +1,28 @@
-#pragma once 
- 
+#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 
+#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: 
+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; 
-}; 
+    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
index defa1a0e82..9cf2aced44 100644
--- a/library/cpp/threading/equeue/equeue_ut.cpp
+++ b/library/cpp/threading/equeue/equeue_ut.cpp
@@ -1,125 +1,125 @@
-#include "equeue.h" 
- 
+#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> 
- 
+
+#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() { 
+    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 
+            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 { 
+        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 { 
+                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; 
- 
+            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; 
- 
+        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; 
- 
+                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) { 
+            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); 
-    } 
-} 
+            }
+
+            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);
+    }
+}