pq sdk deadlock has been fixed (between Close and OnReadDone) (#3115)

ydb-platform · Mar 26, 2024 · 2ef6dd4 · 2ef6dd4
1 parent ee65c88
commit 2ef6dd4
Showing 1 changed file with 10 additions and 1 deletion.
diff --git a/ydb/public/sdk/cpp/client/ydb_persqueue_core/impl/read_session.h b/ydb/public/sdk/cpp/client/ydb_persqueue_core/impl/read_session.h
@@ -739,6 +739,10 @@ class TPartitionStreamImpl : public TAPartitionStream<UseMigrationProtocol> {
         EventsQueue.clear();
     }
 
+    TRawPartitionStreamEventQueue<UseMigrationProtocol> ExtractQueue() noexcept {
+        return std::move(EventsQueue);
+    }
+
     static void GetDataEventImpl(TIntrusivePtr<TPartitionStreamImpl<UseMigrationProtocol>> partitionStream,
                                  size_t& maxEventsCount,
                                  size_t& maxByteSize,
@@ -796,14 +800,16 @@ class TReadSessionEventsQueue: public TBaseSessionEventsQueue<TAReadSessionSetti
 
     bool Close(const TASessionClosedEvent<UseMigrationProtocol>& event, TDeferredActions<UseMigrationProtocol>& deferred) {
         TWaiter waiter;
+        TVector<TRawPartitionStreamEventQueue<UseMigrationProtocol>> defferedDelete;
         with_lock (TParent::Mutex) {
             if (TParent::Closed) {
                 return false;
             }
+            defferedDelete.reserve(TParent::Events.size());
             while (!TParent::Events.empty()) {
                 auto& event = TParent::Events.front();
                 if (!event.IsEmpty()) {
-                    event.PartitionStream->ClearQueue();
+                    defferedDelete.push_back(event.PartitionStream->ExtractQueue());
                 }
                 TParent::Events.pop();
             }
@@ -812,6 +818,9 @@ class TReadSessionEventsQueue: public TBaseSessionEventsQueue<TAReadSessionSetti
             waiter = TWaiter(TParent::Waiter.ExtractPromise(), this);
         }
 
+        // Delayed deletion is necessary to avoid deadlock with PushEvent
+        defferedDelete.clear();
+
         TReadSessionEventInfo<UseMigrationProtocol> info(event);
         ApplyHandler(info, deferred);
         deferred.DeferSignalWaiter(std::move(waiter));