dqrun pq simulation: add queue limit (ydb-platform#9962)

Author: yumkam

ydb/library/yql/providers/pq/gateway/dummy/yql_pq_file_topic_client.cpp

@@ -12,9 +12,13 @@ namespace NYql {
 
 class TBlockingEQueue {
 public:
-    void Push(NYdb::NTopic::TReadSessionEvent::TEvent&& e) {
+    TBlockingEQueue(size_t maxSize):MaxSize_(maxSize) {
+    }
+    void Push(NYdb::NTopic::TReadSessionEvent::TEvent&& e, size_t size) {
         with_lock(Mutex_) {
-            Events_.push_back(std::move(e));
+            CanPush_.WaitI(Mutex_, [this] () {return Stopped_ || Size_ < MaxSize_;});
+            Events_.emplace_back(std::move(e), size );
+            Size_ += size;
         }
         CanPop_.BroadCast();
     }
@@ -29,26 +33,26 @@ class TBlockingEQueue {
         with_lock(Mutex_) {
             if (block) {
                 CanPop_.WaitI(Mutex_, [this] () {return CanPopPredicate();});
-
-                auto front = Events_.front();
-                Events_.pop_front();
-                return front;
             } else {
                 if (!CanPopPredicate()) {
                     return {};
                 }
-
-                auto front = Events_.front();
-                Events_.pop_front();
-                return front;
             }
+            auto [front, size] = std::move(Events_.front());
+            Events_.pop_front();
+            Size_ -= size;
+            if (Size_ < MaxSize_) {
+                CanPush_.BroadCast();
+            }
+            return front;
         }
     }
 
     void Stop() {
         with_lock(Mutex_) {
             Stopped_ = true;
             CanPop_.BroadCast();
+            CanPush_.BroadCast();
         }
     }
 
@@ -63,10 +67,13 @@ class TBlockingEQueue {
         return !Events_.empty() && !Stopped_;
     }
 
-    TDeque<NYdb::NTopic::TReadSessionEvent::TEvent> Events_;
+    size_t MaxSize_;
+    size_t Size_ = 0;
+    TDeque<std::pair<NYdb::NTopic::TReadSessionEvent::TEvent, size_t>> Events_;
     bool Stopped_ = false;
     TMutex Mutex_;
     TCondVar CanPop_;
+    TCondVar CanPush_;
 };
 
 class TFileTopicReadSession : public NYdb::NTopic::IReadSession {
@@ -174,21 +181,23 @@ constexpr static auto FILE_POLL_PERIOD = TDuration::MilliSeconds(100);
         while (!EventsQ_.IsStopped()) {
             TString rawMsg;
             TVector<TMessage> msgs;
+            size_t size = 0;
 
             while (size_t read = fi.ReadLine(rawMsg)) {
                 msgs.emplace_back(MakeNextMessage(rawMsg));
                 MsgOffset_++;
+                size += rawMsg.size();
             }
             if (!msgs.empty()) {
-                EventsQ_.Push(NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent(msgs, {}, Session_));
+                EventsQ_.Push(NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent(msgs, {}, Session_), size);
             }
 
             Sleep(FILE_POLL_PERIOD);
         }
     }
 
     TFile File_;
-    TBlockingEQueue EventsQ_;
+    TBlockingEQueue EventsQ_ {4_MB};
     NYdb::NTopic::TPartitionSession::TPtr Session_;
     TString ProducerId_;
     std::thread FilePoller_;
@@ -298,4 +307,4 @@ NYdb::TAsyncStatus TFileTopicClient::CommitOffset(const TString& path, ui64 part
     return NThreading::MakeFuture(NYdb::TStatus(NYdb::EStatus::SUCCESS, {}));
 }
 
-}
+}