From e451b5905880d7aba0a4ede5e145ec59bf74e87e Mon Sep 17 00:00:00 2001 From: babenko Date: Fri, 27 Mar 2026 16:17:31 +0300 Subject: YT-27754: Avoid fiber context switch in TSyncBufferedOutputStreamAdapter::dtor commit_hash:459060fb05ec6c10d4d065f96dee05dc28a91beb --- yt/yt/client/driver/driver.cpp | 2 +- yt/yt/client/driver/table_commands.cpp | 2 ++ yt/yt/core/concurrency/async_stream_helpers.cpp | 8 -------- yt/yt/library/formats/schemaful_dsv_writer.cpp | 14 ++++++-------- yt/yt/library/formats/web_json_writer.cpp | 6 ++++-- 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 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 Output_; - - void DoFlushBuffer() - { - Output_->Flush(); - } + const std::unique_ptr Output_; + // XXX(babenko): is not actually used TFuture 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 UnderlyingOutput_; + const std::unique_ptr UnderlyingOutput_; TCountingOutput Output_; - std::unique_ptr ResponseBuilder_; + const std::unique_ptr ResponseBuilder_; TWebJsonColumnFilter ColumnFilter_; THashMap AllColumnIdToName_; @@ -560,6 +560,8 @@ TWriterForWebJson::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, -- cgit v1.3