intermediate changes

ref:fd45b675f102830e5f7311e0b3c6aaa6ee7d4eaa

3 files changed, 167 insertions, 0 deletions

diff --git a/library/cpp/lwtrace/control.h b/library/cpp/lwtrace/control.h
index 16b24eafd2d..b7c47aaec6f 100644
--- a/library/cpp/lwtrace/control.h
+++ b/library/cpp/lwtrace/control.h
@@ -200,6 +200,12 @@ namespace NLWTrace {
             CyclicDepot.ReadItems(r);
             DurationDepot.ReadItems(now, duration, r);
         }
+
+        template <class TReader>
+        void ExtractItemsFromCyclicDepot(TReader& r) const {
+            CyclicDepot.ExtractItems(r);
+        }
+
         void ToProtobuf(TLogPb& pb) const;
     };
 
@@ -310,6 +316,18 @@ namespace NLWTrace {
             }
         }
 
+        template <class TReader>
+        void ExtractItemsFromCyclicDepot(const TString& id, TReader& reader) {
+            TGuard<TMutex> g(Mtx);
+            TTraces::iterator it = Traces.find(id);
+            if (it == Traces.end()) {
+                ythrow yexception() << "trace id '" << id << "' is not used";
+            } else {
+                TSession* trace = it->second;
+                trace->ExtractItemsFromCyclicDepot(reader);
+            }
+        }
+
         bool GetDestructiveActionsAllowed() {
             return DestructiveActionsAllowed;
         }
diff --git a/library/cpp/lwtrace/log.h b/library/cpp/lwtrace/log.h
index ddf85d9916b..0e05ec6306c 100644
--- a/library/cpp/lwtrace/log.h
+++ b/library/cpp/lwtrace/log.h
@@ -444,6 +444,12 @@ namespace NLWTrace {
                     }
                 }
             }
+
+            template <class TReader>
+            void ExtractItems(TReader& r) {
+                ReadItems(r);
+                OldBuffer->Clear();
+            }
         };
 
         size_t Capacity;
@@ -547,6 +553,17 @@ namespace NLWTrace {
             }
         }
 
+        template <class TReader>
+        void ExtractItems(TReader& r) const {
+            TGuard<TSpinLock> g(Lock);
+            for (auto i: StoragesVec) {
+                i->ExtractItems(r);
+            }
+            for (const auto& orphanStorage: OrphanStorages) {
+                orphanStorage->ExtractItems(r);
+            }
+        }
+
         class TAccessor {
         private:
             TStorage& Storage;
diff --git a/library/cpp/lwtrace/log_ut.cpp b/library/cpp/lwtrace/log_ut.cpp
new file mode 100644
index 00000000000..d9ee5f10ae3
--- /dev/null
+++ b/library/cpp/lwtrace/log_ut.cpp
@@ -0,0 +1,132 @@
+#include "log.h"
+
+#include <library/cpp/testing/unittest/registar.h>
+
+using namespace NLWTrace;
+
+namespace
+{
+    struct TData
+    {
+        ui32 Val = 0;
+
+        void Clear()
+        {
+        }
+    };
+
+    struct TReader
+    {
+        ui32 NumSeen = 0;
+
+        void Reset()
+        {
+            NumSeen = 0;
+        }
+
+        void Push(TThread::TId tid, const TData& val)
+        {
+            Y_UNUSED(tid);
+            Y_UNUSED(val);
+
+            ++NumSeen;
+        }
+    };
+};
+
+Y_UNIT_TEST_SUITE(LWTraceLog) {
+    Y_UNIT_TEST(ShouldAccumulateTracesViaReadItems) {
+        TCyclicLogImpl<TData> log(100);
+
+        {
+            TCyclicLogImpl<TData>::TAccessor acc1(log);
+            TCyclicLogImpl<TData>::TAccessor acc2(log);
+            TCyclicLogImpl<TData>::TAccessor acc3(log);
+
+            acc1.Add()->Val = 1;
+            acc2.Add()->Val = 2;
+            acc3.Add()->Val = 3;
+        }
+
+        TReader reader;
+
+        log.ReadItems(reader);
+        UNIT_ASSERT_VALUES_EQUAL(3, reader.NumSeen);
+
+        {
+            TCyclicLogImpl<TData>::TAccessor acc1(log);
+            TCyclicLogImpl<TData>::TAccessor acc2(log);
+            TCyclicLogImpl<TData>::TAccessor acc3(log);
+
+            acc1.Add()->Val = 4;
+            acc2.Add()->Val = 5;
+            acc3.Add()->Val = 6;
+        }
+
+        reader.Reset();
+        log.ReadItems(reader);
+
+        UNIT_ASSERT_VALUES_EQUAL(6, reader.NumSeen);
+   }
+
+    Y_UNIT_TEST(ShouldNotReturnProcessedItemsViaMoveItems) {
+        struct TData
+        {
+            ui32 Val = 0;
+
+            void Clear()
+            {
+            }
+        };
+
+        struct TReader
+        {
+            ui32 NumSeen = 0;
+
+            void Reset()
+            {
+                NumSeen = 0;
+            }
+
+            void Push(TThread::TId tid, const TData& val)
+            {
+                Y_UNUSED(tid);
+                Y_UNUSED(val);
+
+                ++NumSeen;
+            }
+        };
+
+        TCyclicLogImpl<TData> log(100);
+
+        {
+            TCyclicLogImpl<TData>::TAccessor acc1(log);
+            TCyclicLogImpl<TData>::TAccessor acc2(log);
+            TCyclicLogImpl<TData>::TAccessor acc3(log);
+
+            acc1.Add()->Val = 1;
+            acc2.Add()->Val = 2;
+            acc3.Add()->Val = 3;
+        }
+
+        TReader reader;
+
+        log.ExtractItems(reader);
+        UNIT_ASSERT_VALUES_EQUAL(3, reader.NumSeen);
+
+        {
+            TCyclicLogImpl<TData>::TAccessor acc1(log);
+            TCyclicLogImpl<TData>::TAccessor acc2(log);
+            TCyclicLogImpl<TData>::TAccessor acc3(log);
+
+            acc1.Add()->Val = 4;
+            acc2.Add()->Val = 5;
+            acc3.Add()->Val = 6;
+        }
+
+        reader.Reset();
+        log.ExtractItems(reader);
+
+        UNIT_ASSERT_VALUES_EQUAL(3, reader.NumSeen);
+   }
+}