Merge 714ec12 into 8f51162

Author: serbel324

ydb/core/blobstorage/dsproxy/dsproxy.h

@@ -37,7 +37,8 @@ const ui64 UnconfiguredBufferSizeLimit = 32 << 20;
 
 const TDuration ProxyEstablishSessionsTimeout = TDuration::Seconds(5);
 
-const ui64 DsPutWakeupMs = 60000;
+const TDuration DsMinimumDelayBetweenPutWakeups = TDuration::Seconds(1);
+const TDuration DsMaximumPutTimeout = TDuration::Seconds(60);
 
 const ui64 BufferSizeThreshold = 1 << 20;
 

ydb/core/blobstorage/dsproxy/dsproxy_put.cpp

@@ -42,6 +42,10 @@ class TBlobStorageGroupPutRequest : public TBlobStorageGroupRequestActor {
     TErasureSplitContext ErasureSplitContext = TErasureSplitContext::Init(MaxBytesToSplitAtOnce);
     TBatchedVec<TStackVec<TRope, TypicalPartsInBlob>> PartSets;
 
+    using TDeadlineMask = std::bitset<MaxBatchedPutRequests>;
+    std::map<TInstant, TDeadlineMask> PutDeadlineMasks;
+    TDeadlineMask DeadlineMask; 
+
     TStackVec<ui64, TypicalDisksInGroup> WaitingVDiskResponseCount;
     ui64 WaitingVDiskCount = 0;
 
@@ -645,7 +649,11 @@ class TBlobStorageGroupPutRequest : public TBlobStorageGroupRequestActor {
             << " Tactic# " << TEvBlobStorage::TEvPut::TacticName(Tactic)
             << " RestartCounter# " << RestartCounter);
 
+        TInstant now = TActivationContext::Now();
+
         for (size_t blobIdx = 0; blobIdx < PutImpl.Blobs.size(); ++blobIdx) {
+            TInstant deadline = std::min(now + DsMaximumPutTimeout, PutImpl.Blobs[blobIdx].Deadline);
+            PutDeadlineMasks[deadline].set(blobIdx);
             LWTRACK(DSProxyPutBootstrapStart, PutImpl.Blobs[blobIdx].Orbit);
         }
 
@@ -666,7 +674,8 @@ class TBlobStorageGroupPutRequest : public TBlobStorageGroupRequestActor {
             getTotalSize()
         );
 
-        Become(&TBlobStorageGroupPutRequest::StateWait, TDuration::MilliSeconds(DsPutWakeupMs), new TKikimrEvents::TEvWakeup);
+        Become(&TBlobStorageGroupPutRequest::StateWait);
+        ScheduleNextWakeup(TInstant::Zero());
 
         PartSets.resize(PutImpl.Blobs.size());
         for (auto& partSet : PartSets) {
@@ -717,15 +726,27 @@ class TBlobStorageGroupPutRequest : public TBlobStorageGroupRequestActor {
             << " BlobIDs# " << BlobIdSequenceToString()
             << " Not answered in "
             << (TActivationContext::Monotonic() - RequestStartTime) << " seconds");
+
         const TInstant now = TActivationContext::Now();
+        while (!PutDeadlineMasks.empty()) {
+            auto [deadline, mask] = *PutDeadlineMasks.begin();
+            if (deadline <= now) {
+                DeadlineMask |= mask;
+                PutDeadlineMasks.erase(PutDeadlineMasks.begin());
+            } else {
+                break;
+            }
+        }
+
         TPutImpl::TPutResultVec putResults;
         for (size_t blobIdx = 0; blobIdx < PutImpl.Blobs.size(); ++blobIdx) {
-            if (!PutImpl.Blobs[blobIdx].Replied && now > PutImpl.Blobs[blobIdx].Deadline) {
+            if (!PutImpl.Blobs[blobIdx].Replied && DeadlineMask[blobIdx]) {
                 PutImpl.PrepareOneReply(NKikimrProto::DEADLINE, blobIdx, LogCtx, "Deadline timer hit", putResults);
             }
         }
-        ReplyAndDieWithLastResponse(putResults);
-        Schedule(TDuration::MilliSeconds(DsPutWakeupMs), new TKikimrEvents::TEvWakeup);
+        if (!ReplyAndDieWithLastResponse(putResults)) {
+            ScheduleNextWakeup(now);
+        }
     }
 
     void UpdatePengingVDiskResponseCount(const TDeque<TPutImpl::TPutEvent>& putEvents) {
@@ -792,6 +813,23 @@ class TBlobStorageGroupPutRequest : public TBlobStorageGroupRequestActor {
             << " State# " << PutImpl.DumpFullState());
     }
 
+    void ScheduleNextWakeup(TInstant lastWakeup) {
+        TInstant now = TActivationContext::Now();
+        TInstant deadline = lastWakeup + DsMinimumDelayBetweenPutWakeups;
+
+        // find first deadline after now
+        for (auto it = PutDeadlineMasks.begin(); it != PutDeadlineMasks.end(); ++it) {
+            deadline = std::max(deadline, it->first);
+            if (it->first > now) {
+                break;
+            }
+        }
+
+        if (deadline != TInstant::Max()) {
+            Schedule(deadline, new TKikimrEvents::TEvWakeup);
+        }
+    }
+
     STATEFN(StateWait) {
         if (ProcessEvent(ev, true)) {
             return;

ydb/core/blobstorage/dsproxy/dsproxy_request.cpp

@@ -502,7 +502,7 @@ namespace NKikimr {
                     .ForceGroupGeneration = ev->Get()->ForceGroupGeneration,
                 }
             }),
-            TInstant::Max()
+            ev->Get()->Deadline
         );
     }