AsyncDecoding and HttpGateway have been fixed (#9118)

Author: dorooleg

ydb/library/yql/providers/common/http_gateway/yql_http_gateway.cpp

@@ -746,7 +746,7 @@ friend class IHTTPGateway;
     }
 
     size_t FillHandlers() {
-        const std::unique_lock lock(Sync);
+        const std::unique_lock lock(SyncRef());
         for (auto it = Streams.cbegin(); Streams.cend() != it;) {
             if (const auto& stream = it->lock()) {
                 const auto streamHandle = stream->GetHandle();
@@ -795,7 +795,7 @@ friend class IHTTPGateway;
         TEasyCurl::TPtr easy;
         long httpResponseCode = 0L;
         {
-            const std::unique_lock lock(Sync);
+            const std::unique_lock lock(SyncRef());
             if (const auto it = Allocated.find(handle); Allocated.cend() != it) {
                 easy = std::move(it->second);
                 TString codeLabel;
@@ -847,7 +847,7 @@ friend class IHTTPGateway;
     void Fail(CURLMcode result) {
         std::stack<TEasyCurl::TPtr> works;
         {
-            const std::unique_lock lock(Sync);
+            const std::unique_lock lock(SyncRef());
 
             for (auto& item : Allocated) {
                 works.emplace(std::move(item.second));
@@ -868,7 +868,7 @@ friend class IHTTPGateway;
     void Upload(TString url, THeaders headers, TString body, TOnResult callback, bool put, TRetryPolicy::TPtr retryPolicy) final {
         Rps->Inc();
 
-        const std::unique_lock lock(Sync);
+        const std::unique_lock lock(SyncRef());
         auto easy = TEasyCurlBuffer::Make(InFlight, DownloadedBytes, UploadedBytes, std::move(url), put ? TEasyCurl::EMethod::PUT : TEasyCurl::EMethod::POST, std::move(body), std::move(headers), 0U, 0U, std::move(callback), retryPolicy ? retryPolicy->CreateRetryState() : nullptr, InitConfig, DnsGateway.GetDNSCurlList());
         Await.emplace(std::move(easy));
         Wakeup(0U);
@@ -877,7 +877,7 @@ friend class IHTTPGateway;
     void Delete(TString url, THeaders headers, TOnResult callback, TRetryPolicy::TPtr retryPolicy) final {
         Rps->Inc();
 
-        const std::unique_lock lock(Sync);
+        const std::unique_lock lock(SyncRef());
         auto easy = TEasyCurlBuffer::Make(InFlight, DownloadedBytes, UploadedBytes, std::move(url), TEasyCurl::EMethod::DELETE, 0, std::move(headers), 0U, 0U, std::move(callback), retryPolicy ? retryPolicy->CreateRetryState() : nullptr, InitConfig, DnsGateway.GetDNSCurlList());
         Await.emplace(std::move(easy));
         Wakeup(0U);
@@ -898,7 +898,7 @@ friend class IHTTPGateway;
             callback(TResult(CURLE_OK, TIssues{error}));
             return;
         }
-        const std::unique_lock lock(Sync);
+        const std::unique_lock lock(SyncRef());
         auto easy = TEasyCurlBuffer::Make(InFlight, DownloadedBytes, UploadedBytes, std::move(url), TEasyCurl::EMethod::GET, std::move(data), std::move(headers), offset, sizeLimit, std::move(callback), retryPolicy ? retryPolicy->CreateRetryState() : nullptr, InitConfig, DnsGateway.GetDNSCurlList());
         Await.emplace(std::move(easy));
         Wakeup(sizeLimit);
@@ -915,13 +915,14 @@ friend class IHTTPGateway;
         const ::NMonitoring::TDynamicCounters::TCounterPtr& inflightCounter) final
     {
         auto stream = TEasyCurlStream::Make(InFlightStreams, DownloadedBytes, UploadedBytes, std::move(url), std::move(headers), offset, sizeLimit, std::move(onStart), std::move(onNewData), std::move(onFinish), inflightCounter, InitConfig, DnsGateway.GetDNSCurlList());
-        const std::unique_lock lock(Sync);
+        const std::unique_lock lock(SyncRef());
         const auto handle = stream->GetHandle();
         TEasyCurlStream::TWeakPtr weak = stream;
         Streams.emplace_back(stream);
         Allocated.emplace(handle, std::move(stream));
         Wakeup(0ULL);
-        return [weak](TIssue issue) {
+        return [weak, sync=Sync](TIssue issue) {
+            const std::unique_lock lock(*sync);
             if (const auto& stream = weak.lock())
                 stream->Cancel(issue);
         };
@@ -932,7 +933,7 @@ friend class IHTTPGateway;
     }
 
     void OnRetry(TEasyCurlBuffer::TPtr easy) {
-        const std::unique_lock lock(Sync);
+        const std::unique_lock lock(SyncRef());
         const size_t sizeLimit = easy->GetSizeLimit();
         Await.emplace(std::move(easy));
         Wakeup(sizeLimit);
@@ -950,6 +951,10 @@ friend class IHTTPGateway;
     }
 
 private:
+    std::mutex& SyncRef() {
+        return *Sync;
+    }
+
     CURLM* Handle = nullptr;
 
     std::queue<TEasyCurlBuffer::TPtr> Await;
@@ -959,7 +964,7 @@ friend class IHTTPGateway;
     std::unordered_map<CURL*, TEasyCurl::TPtr> Allocated;
     std::priority_queue<std::pair<TInstant, TEasyCurlBuffer::TPtr>> Delayed;
 
-    std::mutex Sync;
+    std::shared_ptr<std::mutex> Sync = std::make_shared<std::mutex>();
     std::thread Thread;
     std::atomic<bool> IsStopped = false;
 

ydb/library/yql/providers/s3/actors/yql_s3_read_actor.cpp

@@ -167,7 +167,11 @@ struct TReadSpec {
     NDB::ColumnsWithTypeAndName CHColumns;
     std::shared_ptr<arrow::Schema> ArrowSchema;
     NDB::FormatSettings Settings;
-    TString Format, Compression;
+    // It's very important to keep here std::string instead of TString 
+    // because of the cast from TString to std::string is using the MutRef (it isn't thread-safe).
+    // This behaviour can be found in the getInputFormat call
+    std::string Format;  
+    TString Compression;
     ui64 SizeLimit = 0;
     ui32 BlockLengthPosition = 0;
     std::vector<ui32> ColumnReorder;
@@ -1375,12 +1379,7 @@ class TS3StreamReadActor : public TActorBootstrapped<TS3StreamReadActor>, public
             DecodedChunkSizeHist,
             HttpInflightSize,
             HttpDataRps,
-            DeferredQueueSize,
-            ReadSpec->Format,
-            ReadSpec->Compression,
-            ReadSpec->ArrowSchema,
-            ReadSpec->RowSpec,
-            ReadSpec->Settings
+            DeferredQueueSize
         );
 
         if (!UseRuntimeListing) {

ydb/library/yql/providers/s3/common/source_context.h

@@ -34,12 +34,7 @@ struct TSourceContext {
         , NMonitoring::THistogramPtr decodedChunkSizeHist
         , NMonitoring::TDynamicCounters::TCounterPtr httpInflightSize
         , NMonitoring::TDynamicCounters::TCounterPtr httpDataRps
-        , NMonitoring::TDynamicCounters::TCounterPtr deferredQueueSize
-        , const TString format
-        , const TString compression
-        , std::shared_ptr<arrow::Schema> schema
-        , std::unordered_map<TStringBuf, NKikimr::NMiniKQL::TType*, THash<TStringBuf>> rowTypes
-        , NDB::FormatSettings settings)
+        , NMonitoring::TDynamicCounters::TCounterPtr deferredQueueSize)
         : SourceId(sourceId)
         , Limit(limit)
         , ActorSystem(actorSystem)
@@ -54,11 +49,6 @@ struct TSourceContext {
         , HttpInflightSize(httpInflightSize)
         , HttpDataRps(httpDataRps)
         , DeferredQueueSize(deferredQueueSize)
-        , Format(format)
-        , Compression(compression)
-        , Schema(schema)
-        , RowTypes(rowTypes)
-        , Settings(settings)
     {
     }
 
@@ -105,11 +95,6 @@ struct TSourceContext {
     NMonitoring::TDynamicCounters::TCounterPtr HttpInflightSize;
     NMonitoring::TDynamicCounters::TCounterPtr HttpDataRps;
     NMonitoring::TDynamicCounters::TCounterPtr DeferredQueueSize;
-    const TString Format;
-    const TString Compression;
-    std::shared_ptr<arrow::Schema> Schema;
-    std::unordered_map<TStringBuf, NKikimr::NMiniKQL::TType*, THash<TStringBuf>> RowTypes;
-    NDB::FormatSettings Settings;
 private:
     std::atomic_uint64_t Value;
     std::mutex Mutex;