Make TOneOneQueueInplace usable without an extra pointer layer

Author: snaury

ydb/core/tablet_flat/logic_redo_entry.h

@@ -11,16 +11,16 @@ namespace NRedo {
 
     struct TEntry {
         template<typename ... Args>
-        static TEntry* Create(NTable::TTxStamp stamp, TArrayRef<const ui32> affects, Args&& ... args)
+        static std::unique_ptr<TEntry> Create(NTable::TTxStamp stamp, TArrayRef<const ui32> affects, Args&& ... args)
         {
-            auto *ptr = malloc(sizeof(TEntry) + affects.size() * sizeof(ui32));
+            void* ptr = ::operator new(sizeof(TEntry) + affects.size() * sizeof(ui32));
 
-            return ::new(ptr) TEntry(stamp, affects, std::forward<Args>(args)...);
+            return std::unique_ptr<TEntry>(::new(ptr) TEntry(stamp, affects, std::forward<Args>(args)...));
         }
 
-        void operator delete (void *p)
+        void operator delete(void* p)
         {
-            free(p);
+            ::operator delete(p);
         }
 
         void Describe(IOutputStream &out) const

ydb/core/tablet_flat/logic_redo_queue.h

@@ -19,10 +19,10 @@ namespace NRedo {
     struct TQueue {
         using TStamp = NTable::TTxStamp;
         using TAffects = TArrayRef<const ui32>;
+        using TLog = TOneOneQueueInplace<TEntry*, 4096, TDelete>;
 
         TQueue(THashMap<ui32, NTable::TSnapEdge> edges)
-            : Log(new TLog)
-            , Edges(std::move(edges))
+            : Edges(std::move(edges))
         {
 
         }
@@ -45,13 +45,14 @@ namespace NRedo {
                 << " (" << Memory << " mem" << ", " << LargeGlobIdsBytes << " raw)b }";
         }
 
-        void Push(TEntry *entry)
+        void Push(std::unique_ptr<TEntry>&& entryPtr)
         {
+            TEntry* entry = entryPtr.get();
             if (bool(entry->Embedded) == bool(entry->LargeGlobId)) {
                 Y_TABLET_ERROR(NFmt::Do(*entry) << " has incorrect payload");
             }
 
-            Log->Push(entry);
+            Log.Push(entryPtr.release());
 
             Items++;
             Memory += entry->BytesMem();
@@ -92,15 +93,16 @@ namespace NRedo {
             for (auto &it : Overhead)
                 it.second.Clear();
 
-            auto was = std::exchange(Log, new TLog);
+            TLog was;
+            was.Swap(Log);
 
             Items = 0;
             Memory = 0;
             LargeGlobIdsBytes = 0;
 
             auto logos = snap.MutableNonSnapLogBodies();
 
-            while (TAutoPtr<TEntry> entry = was->Pop()) {
+            while (auto entry = std::unique_ptr<TEntry>(was.Pop())) {
                 if (entry->FilterTables(Edges)) {
                     for (const auto& blobId : entry->LargeGlobId.Blobs()) {
                         LogoBlobIDFromLogoBlobID(blobId, logos->Add());
@@ -116,7 +118,7 @@ namespace NRedo {
 
                     entry->References = 0;
 
-                    Push(entry.Release());
+                    Push(std::move(entry));
                 } else {
                     Y_ENSURE(entry->References == 0);
                 }
@@ -134,9 +136,7 @@ namespace NRedo {
             return Usage;
         }
 
-        using TLog = TOneOneQueueInplace<NRedo::TEntry *, 4096>;
-
-        TAutoPtr<TLog, TLog::TPtrCleanInplaceMallocDestructor> Log;
+        TLog Log;
         THashMap<ui32, NTable::TSnapEdge> Edges;
         THashMap<ui32, TOverhead> Overhead;
         TIntrusiveList<TOverhead> Changes;

ydb/library/actors/util/queue_oneone_inplace.h

@@ -3,17 +3,15 @@
 #include "defs.h"
 #include "queue_chunk.h"
 
-template <typename T, ui32 TSize, typename TChunk = TQueueChunk<T, TSize>>
+template <typename T, ui32 TSize, class D = TNoAction, typename TChunk = TQueueChunk<T, TSize>>
 class TOneOneQueueInplace : TNonCopyable {
-    static_assert(std::is_integral<T>::value || std::is_pointer<T>::value, "expect std::is_integral<T>::value || std::is_pointer<T>::valuer");
+    static_assert(std::is_integral<T>::value || std::is_pointer<T>::value, "expect std::is_integral<T>::value || std::is_pointer<T>::value");
 
     TChunk* ReadFrom;
     ui32 ReadPosition;
     ui32 WritePosition;
     TChunk* WriteTo;
 
-    friend class TReadIterator;
-
 public:
     class TReadIterator {
         TChunk* ReadFrom;
@@ -28,14 +26,15 @@ class TOneOneQueueInplace : TNonCopyable {
 
         inline T Next() {
             TChunk* head = ReadFrom;
-            if (ReadPosition != TChunk::EntriesCount) {
-                return AtomicLoad(&head->Entries[ReadPosition++]);
-            } else if (TChunk* next = AtomicLoad(&head->Next)) {
-                ReadFrom = next;
+            if (ReadPosition == TChunk::EntriesCount) [[unlikely]] {
+                head = AtomicLoad(&head->Next);
+                if (!head) {
+                    return T{};
+                }
+                ReadFrom = head;
                 ReadPosition = 0;
-                return Next();
             }
-            return T{};
+            return AtomicLoad(&head->Entries[ReadPosition++]);
         }
     };
 
@@ -48,71 +47,83 @@ class TOneOneQueueInplace : TNonCopyable {
     }
 
     ~TOneOneQueueInplace() {
-        Y_DEBUG_ABORT_UNLESS(Head() == 0);
-        delete ReadFrom;
-    }
-
-    struct TPtrCleanDestructor {
-        static inline void Destroy(TOneOneQueueInplace<T, TSize>* x) noexcept {
-            while (T head = x->Pop())
+        if constexpr (!std::is_same_v<D, TNoAction>) {
+            while (T x = Pop()) {
+                D::Destroy(x);
+            }
+            delete ReadFrom;
+        } else {
+            TChunk* next = ReadFrom;
+            do {
+                TChunk* head = next;
+                next = AtomicLoad(&head->Next);
                 delete head;
-            delete x;
+            } while (next);
         }
-    };
+    }
 
     struct TCleanDestructor {
         static inline void Destroy(TOneOneQueueInplace<T, TSize>* x) noexcept {
-            while (x->Pop() != nullptr)
-                continue;
             delete x;
         }
     };
 
-    struct TPtrCleanInplaceMallocDestructor {
-        template <typename TPtrVal>
-        static inline void Destroy(TOneOneQueueInplace<TPtrVal*, TSize>* x) noexcept {
-            while (TPtrVal* head = x->Pop()) {
-                head->~TPtrVal();
-                free(head);
+    struct TPtrCleanDestructor {
+        static inline void Destroy(TOneOneQueueInplace<T, TSize>* x) noexcept {
+            while (T head = x->Pop()) {
+                ::CheckedDelete(head);
             }
             delete x;
         }
     };
 
-    void Push(T x) noexcept {
-        if (WritePosition != TChunk::EntriesCount) {
-            AtomicStore(&WriteTo->Entries[WritePosition], x);
-            ++WritePosition;
-        } else {
+    void Push(T x) {
+        if (WritePosition == TChunk::EntriesCount) [[unlikely]] {
             TChunk* next = new TChunk();
-            next->Entries[0] = x;
+            AtomicStore(&next->Entries[0], x);
             AtomicStore(&WriteTo->Next, next);
             WriteTo = next;
             WritePosition = 1;
+        } else {
+            AtomicStore(&WriteTo->Entries[WritePosition++], x);
         }
     }
 
     T Head() {
         TChunk* head = ReadFrom;
-        if (ReadPosition != TChunk::EntriesCount) {
-            return AtomicLoad(&head->Entries[ReadPosition]);
-        } else if (TChunk* next = AtomicLoad(&head->Next)) {
-            ReadFrom = next;
+        if (ReadPosition == TChunk::EntriesCount) [[unlikely]] {
+            TChunk* next = AtomicLoad(&head->Next);
+            if (!next) {
+                return T{};
+            }
             delete head;
+            head = next;
+            ReadFrom = next;
             ReadPosition = 0;
-            return Head();
         }
-        return T{};
+        return AtomicLoad(&head->Entries[ReadPosition]);
     }
 
     T Pop() {
         T ret = Head();
-        if (ret)
+        if (ret) {
             ++ReadPosition;
+        }
         return ret;
     }
 
     TReadIterator Iterator() {
         return TReadIterator(ReadFrom, ReadPosition);
     }
+
+    /**
+     * Swap must synchronize with both reader and writer
+     */
+    void Swap(TOneOneQueueInplace& other) {
+        using std::swap;
+        swap(ReadFrom, other.ReadFrom);
+        swap(ReadPosition, other.ReadPosition);
+        swap(WritePosition, other.WritePosition);
+        swap(WriteTo, other.WriteTo);
+    }
 };