YT-27754: Avoid fiber context switch in TSyncBufferedOutputStreamAdapter::dtor

commit_hash:459060fb05ec6c10d4d065f96dee05dc28a91beb

5 files changed, 13 insertions, 19 deletions

diff --git a/yt/yt/client/driver/driver.cpp b/yt/yt/client/driver/driver.cpp
index 133c324ce0a..899d51c4b1a 100644
--- a/yt/yt/client/driver/driver.cpp
+++ b/yt/yt/client/driver/driver.cpp
@@ -733,7 +733,7 @@ private:
             Serialize(yson, consumer.get());
 
             consumer->Flush();
-            syncOutputStream->Flush();
+            syncOutputStream->Finish();
         }
 
         void Finish()
diff --git a/yt/yt/client/driver/table_commands.cpp b/yt/yt/client/driver/table_commands.cpp
index 59026a4214e..3f6f661a9e1 100644
--- a/yt/yt/client/driver/table_commands.cpp
+++ b/yt/yt/client/driver/table_commands.cpp
@@ -316,7 +316,9 @@ void TLocateSkynetShareCommand::DoExecute(ICommandContextPtr context)
         syncOutputStream.get());
 
     Serialize(*skynetPartsLocations.ValueOrThrow(), consumer.get());
+
     consumer->Flush();
+    syncOutputStream->Finish();
 }
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/core/concurrency/async_stream_helpers.cpp b/yt/yt/core/concurrency/async_stream_helpers.cpp
index ddeed1af017..5c90d8f79b2 100644
--- a/yt/yt/core/concurrency/async_stream_helpers.cpp
+++ b/yt/yt/core/concurrency/async_stream_helpers.cpp
@@ -178,14 +178,6 @@ public:
         Reset();
     }
 
-    virtual ~TSyncBufferedOutputStreamAdapter()
-    {
-        try {
-            Finish();
-        } catch (...) {
-        }
-    }
-
 private:
     const IAsyncOutputStreamPtr UnderlyingStream_;
     const EWaitForStrategy Strategy_;
diff --git a/yt/yt/library/formats/schemaful_dsv_writer.cpp b/yt/yt/library/formats/schemaful_dsv_writer.cpp
index 9842e083cfd..56281bada13 100644
--- a/yt/yt/library/formats/schemaful_dsv_writer.cpp
+++ b/yt/yt/library/formats/schemaful_dsv_writer.cpp
@@ -184,6 +184,8 @@ public:
         : TSchemafulDsvWriterBase(
             config,
             IdToIndexInRow)
+        // XXX(babenko): this leads to unexpected context switches and must be
+        // completely reworked.
         , Output_(CreateBufferedSyncAdapter(stream))
     {
         WriteColumnNamesHeader([this] (TStringBuf buf, char c) {
@@ -194,7 +196,7 @@ public:
 
     TFuture<void> Close() override
     {
-        DoFlushBuffer();
+        Output_->Finish();
         return OKFuture;
     }
 
@@ -238,7 +240,7 @@ public:
             }
             Output_->Write(Config_->RecordSeparator);
         }
-        DoFlushBuffer();
+        Output_->Flush();
 
         return true;
     }
@@ -254,13 +256,9 @@ public:
     }
 
 private:
-    std::unique_ptr<IOutputStream> Output_;
-
-    void DoFlushBuffer()
-    {
-        Output_->Flush();
-    }
+    const std::unique_ptr<IOutputStream> Output_;
 
+    // XXX(babenko): is not actually used
     TFuture<void> Result_;
 };
 
diff --git a/yt/yt/library/formats/web_json_writer.cpp b/yt/yt/library/formats/web_json_writer.cpp
index 5fe921d3717..8b77a3aa5ed 100644
--- a/yt/yt/library/formats/web_json_writer.cpp
+++ b/yt/yt/library/formats/web_json_writer.cpp
@@ -523,10 +523,10 @@ private:
     const TNameTablePtr NameTable_;
     const TNameTableReader NameTableReader_;
 
-    std::unique_ptr<IOutputStream> UnderlyingOutput_;
+    const std::unique_ptr<IOutputStream> UnderlyingOutput_;
     TCountingOutput Output_;
 
-    std::unique_ptr<IJsonWriter> ResponseBuilder_;
+    const std::unique_ptr<IJsonWriter> ResponseBuilder_;
 
     TWebJsonColumnFilter ColumnFilter_;
     THashMap<ui16, TString> AllColumnIdToName_;
@@ -560,6 +560,8 @@ TWriterForWebJson<TValueWriter>::TWriterForWebJson(
     : Config_(std::move(config))
     , NameTable_(std::move(nameTable))
     , NameTableReader_(NameTable_)
+    // XXX(babenko): this leads to unexpected context switches and must be
+    // completely reworked.
     , UnderlyingOutput_(CreateBufferedSyncAdapter(
         std::move(output),
         EWaitForStrategy::WaitFor,