Rework interconnect sending stream machinery

14 files changed, 612 insertions, 306 deletions

diff --git a/library/cpp/actors/interconnect/event_holder_pool.h b/library/cpp/actors/interconnect/event_holder_pool.h
index b6090a3bc8..59df9bd4c8 100644
--- a/library/cpp/actors/interconnect/event_holder_pool.h
+++ b/library/cpp/actors/interconnect/event_holder_pool.h
@@ -8,7 +8,6 @@ namespace NActors {
     struct TEvFreeItems : TEventLocal<TEvFreeItems, EventSpaceBegin(TEvents::ES_PRIVATE)> {
         static constexpr size_t MaxEvents = 256;
 
-        TList<TTcpPacketOutTask> Items;
         std::list<TEventHolder> FreeQueue;
         TStackVec<THolder<IEventBase>, MaxEvents> Events;
         TStackVec<THolder<TEventSerializedData>, MaxEvents> Buffers;
diff --git a/library/cpp/actors/interconnect/interconnect_channel.cpp b/library/cpp/actors/interconnect/interconnect_channel.cpp
index 316d57551d..64a6bbbd09 100644
--- a/library/cpp/actors/interconnect/interconnect_channel.cpp
+++ b/library/cpp/actors/interconnect/interconnect_channel.cpp
@@ -12,8 +12,7 @@ LWTRACE_USING(ACTORLIB_PROVIDER);
 
 namespace NActors {
     bool TEventOutputChannel::FeedDescriptor(TTcpPacketOutTask& task, TEventHolder& event, ui64 *weightConsumed) {
-        const size_t descrSize = sizeof(TEventDescr2);
-        const size_t amount = sizeof(TChannelPart) + descrSize;
+        const size_t amount = sizeof(TChannelPart) + sizeof(TEventDescr2);
         if (task.GetVirtualFreeAmount() < amount) {
             return false;
         }
@@ -25,28 +24,33 @@ namespace NActors {
         task.Orbit.Take(event.Orbit);
 
         Y_VERIFY(SerializationInfo);
-        event.Descr.Flags = (event.Descr.Flags & ~IEventHandle::FlagForwardOnNondelivery) |
+        const ui32 flags = (event.Descr.Flags & ~IEventHandle::FlagForwardOnNondelivery) |
             (SerializationInfo->IsExtendedFormat ? IEventHandle::FlagExtendedFormat : 0);
 
-        TChannelPart *part = static_cast<TChannelPart*>(task.GetFreeArea());
-        part->Channel = ChannelId | TChannelPart::LastPartFlag;
-        part->Size = descrSize;
-
-        auto *p = reinterpret_cast<TEventDescr2*>(part + 1);
-        *p = {
+        // prepare descriptor record
+        TEventDescr2 descr{
             event.Descr.Type,
-            event.Descr.Flags,
+            flags,
             event.Descr.Recipient,
             event.Descr.Sender,
             event.Descr.Cookie,
             {},
             event.Descr.Checksum
         };
-        traceId.Serialize(&p->TraceId);
+        traceId.Serialize(&descr.TraceId);
+
+        // and channel header before the descriptor
+        TChannelPart part{
+            .Channel = static_cast<ui16>(ChannelId | TChannelPart::LastPartFlag),
+            .Size = sizeof(descr),
+        };
+
+        // append them to the packet
+        task.Write(&part, sizeof(part));
+        task.Write(&descr, sizeof(descr));
 
-        task.AppendBuf(part, amount);
         *weightConsumed += amount;
-        OutputQueueSize -= part->Size;
+        OutputQueueSize -= part.Size;
         Metrics->UpdateOutputChannelEvents(ChannelId);
 
         return true;
@@ -75,7 +79,9 @@ namespace NActors {
                     } else if (event.Event) {
                         State = EState::CHUNKER;
                         IEventBase *base = event.Event.Get();
-                        Chunker.SetSerializingEvent(base);
+                        if (event.EventSerializedSize) {
+                            Chunker.SetSerializingEvent(base);
+                        }
                         SerializationInfoContainer = base->CreateSerializationInfo();
                         SerializationInfo = &SerializationInfoContainer;
                     } else { // event without buffer and IEventBase instance
@@ -83,28 +89,28 @@ namespace NActors {
                         SerializationInfoContainer = {};
                         SerializationInfo = &SerializationInfoContainer;
                     }
+                    if (!event.EventSerializedSize) {
+                        State = EState::DESCRIPTOR;
+                    }
                     break;
 
                 case EState::CHUNKER:
                 case EState::BUFFER: {
-                    size_t maxBytes = task.GetVirtualFreeAmount();
-                    if (maxBytes <= sizeof(TChannelPart)) {
+                    if (task.GetVirtualFreeAmount() <= sizeof(TChannelPart)) {
                         return false;
                     }
 
-                    TChannelPart *part = static_cast<TChannelPart*>(task.GetFreeArea());
-                    part->Channel = ChannelId;
-                    part->Size = 0;
-                    task.AppendBuf(part, sizeof(TChannelPart));
-                    maxBytes -= sizeof(TChannelPart);
-                    Y_VERIFY(maxBytes);
+                    TChannelPart part{
+                        .Channel = ChannelId,
+                        .Size = 0,
+                    };
+
+                    auto partBookmark = task.Bookmark(sizeof(part));
 
                     auto addChunk = [&](const void *data, size_t len) {
                         event.UpdateChecksum(data, len);
-                        task.AppendBuf(data, len);
-                        part->Size += len;
-                        Y_VERIFY_DEBUG(maxBytes >= len);
-                        maxBytes -= len;
+                        task.Append(data, len);
+                        part.Size += len;
 
                         event.EventActuallySerialized += len;
                         if (event.EventActuallySerialized > MaxSerializedEventSize) {
@@ -114,18 +120,18 @@ namespace NActors {
 
                     bool complete = false;
                     if (State == EState::CHUNKER) {
-                        Y_VERIFY_DEBUG(task.GetFreeArea() == part + 1);
-                        while (!complete && maxBytes) {
-                            const auto [first, last] = Chunker.FeedBuf(task.GetFreeArea(), maxBytes);
+                        while (!complete && !task.IsFull()) {
+                            TMutableContiguousSpan out = task.AcquireSpanForWriting();
+                            const auto [first, last] = Chunker.FeedBuf(out.data(), out.size());
                             for (auto p = first; p != last; ++p) {
                                 addChunk(p->first, p->second);
                             }
                             complete = Chunker.IsComplete();
                         }
                         Y_VERIFY(!complete || Chunker.IsSuccessfull());
-                        Y_VERIFY_DEBUG(complete || !maxBytes);
+                        Y_VERIFY_DEBUG(complete || task.IsFull());
                     } else { // BUFFER
-                        while (const size_t numb = Min(maxBytes, Iter.ContiguousSize())) {
+                        while (const size_t numb = Min(task.GetVirtualFreeAmount(), Iter.ContiguousSize())) {
                             const char *obuf = Iter.ContiguousData();
                             addChunk(obuf, numb);
                             Iter += numb;
@@ -138,12 +144,10 @@ namespace NActors {
                             event.EventActuallySerialized, event.EventSerializedSize, event.Descr.Type);
                     }
 
-                    if (!part->Size) {
-                        task.Undo(sizeof(TChannelPart));
-                    } else {
-                        *weightConsumed += sizeof(TChannelPart) + part->Size;
-                        OutputQueueSize -= part->Size;
-                    }
+                    Y_VERIFY_DEBUG(part.Size);
+                    task.WriteBookmark(std::exchange(partBookmark, {}), &part, sizeof(part));
+                    *weightConsumed += sizeof(TChannelPart) + part.Size;
+                    OutputQueueSize -= part.Size;
                     if (complete) {
                         State = EState::DESCRIPTOR;
                     }
diff --git a/library/cpp/actors/interconnect/interconnect_tcp_input_session.cpp b/library/cpp/actors/interconnect/interconnect_tcp_input_session.cpp
index ad2da30ec3..34770133ae 100644
--- a/library/cpp/actors/interconnect/interconnect_tcp_input_session.cpp
+++ b/library/cpp/actors/interconnect/interconnect_tcp_input_session.cpp
@@ -93,7 +93,7 @@ namespace NActors {
 
             switch (State) {
                 case EState::HEADER:
-                    if (IncomingData.GetSize() < sizeof(Header)) {
+                    if (IncomingData.GetSize() < sizeof(TTcpPacketHeader_v2)) {
                         break;
                     } else {
                         ProcessHeader();
@@ -172,14 +172,15 @@ namespace NActors {
     }
 
     void TInputSessionTCP::ProcessHeader() {
-        const bool success = IncomingData.ExtractFrontPlain(&Header, sizeof(Header));
+        TTcpPacketHeader_v2 header;
+        const bool success = IncomingData.ExtractFrontPlain(&header, sizeof(header));
         Y_VERIFY(success);
-        PayloadSize = Header.PayloadLength;
-        HeaderSerial = Header.Serial;
-        HeaderConfirm = Header.Confirm;
+        PayloadSize = header.PayloadLength;
+        const ui64 serial = header.Serial;
+        const ui64 confirm = header.Confirm;
         if (!Params.Encryption) {
-            ChecksumExpected = std::exchange(Header.Checksum, 0);
-            Checksum = Crc32cExtendMSanCompatible(0, &Header, sizeof(Header)); // start calculating checksum now
+            ChecksumExpected = std::exchange(header.Checksum, 0);
+            Checksum = Crc32cExtendMSanCompatible(0, &header, sizeof(header)); // start calculating checksum now
             if (!PayloadSize && Checksum != ChecksumExpected) {
                 LOG_ERROR_IC_SESSION("ICIS10", "payload checksum error");
                 return ReestablishConnection(TDisconnectReason::ChecksumError());
@@ -189,9 +190,9 @@ namespace NActors {
             LOG_CRIT_IC_SESSION("ICIS07", "payload is way too big");
             return DestroySession(TDisconnectReason::FormatError());
         }
-        if (ConfirmedByInput < HeaderConfirm) {
-            ConfirmedByInput = HeaderConfirm;
-            if (AtomicGet(Context->ControlPacketId) <= HeaderConfirm && !NewPingProtocol) {
+        if (ConfirmedByInput < confirm) {
+            ConfirmedByInput = confirm;
+            if (AtomicGet(Context->ControlPacketId) <= confirm && !NewPingProtocol) {
                 ui64 sendTime = AtomicGet(Context->ControlPacketSendTimer);
                 TDuration duration = CyclesToDuration(GetCycleCountFast() - sendTime);
                 const auto durationUs = duration.MicroSeconds();
@@ -205,20 +206,20 @@ namespace NActors {
         }
         if (PayloadSize) {
             const ui64 expected = Context->GetLastProcessedPacketSerial() + 1;
-            if (HeaderSerial == 0 || HeaderSerial > expected) {
-                LOG_CRIT_IC_SESSION("ICIS06", "packet serial %" PRIu64 ", but %" PRIu64 " expected", HeaderSerial, expected);
+            if (serial == 0 || serial > expected) {
+                LOG_CRIT_IC_SESSION("ICIS06", "packet serial %" PRIu64 ", but %" PRIu64 " expected", serial, expected);
                 return DestroySession(TDisconnectReason::FormatError());
             }
-            IgnorePayload = HeaderSerial != expected;
+            IgnorePayload = serial != expected;
             State = EState::PAYLOAD;
-        } else if (HeaderSerial & TTcpPacketBuf::PingRequestMask) {
-            Send(SessionId, new TEvProcessPingRequest(HeaderSerial & ~TTcpPacketBuf::PingRequestMask));
-        } else if (HeaderSerial & TTcpPacketBuf::PingResponseMask) {
-            const ui64 sent = HeaderSerial & ~TTcpPacketBuf::PingResponseMask;
+        } else if (serial & TTcpPacketBuf::PingRequestMask) {
+            Send(SessionId, new TEvProcessPingRequest(serial & ~TTcpPacketBuf::PingRequestMask));
+        } else if (serial & TTcpPacketBuf::PingResponseMask) {
+            const ui64 sent = serial & ~TTcpPacketBuf::PingResponseMask;
             const ui64 received = GetCycleCountFast();
             HandlePingResponse(CyclesToDuration(received - sent));
-        } else if (HeaderSerial & TTcpPacketBuf::ClockMask) {
-            HandleClock(TInstant::MicroSeconds(HeaderSerial & ~TTcpPacketBuf::ClockMask));
+        } else if (serial & TTcpPacketBuf::ClockMask) {
+            HandleClock(TInstant::MicroSeconds(serial & ~TTcpPacketBuf::ClockMask));
         }
     }
 
diff --git a/library/cpp/actors/interconnect/interconnect_tcp_session.cpp b/library/cpp/actors/interconnect/interconnect_tcp_session.cpp
index 72982c1458..16cadc9e92 100644
--- a/library/cpp/actors/interconnect/interconnect_tcp_session.cpp
+++ b/library/cpp/actors/interconnect/interconnect_tcp_session.cpp
@@ -271,37 +271,27 @@ namespace NActors {
         // also reset SendQueuePos
 
         // drop confirmed packets first as we do not need unwanted retransmissions
-        SendQueuePos = SendQueue.end();
-        DropConfirmed(nextPacket);
+        DropConfirmed(nextPacket, true);
+        OutgoingStream.Rewind();
+        SendQueuePos = 0;
+        SendOffset = 0;
 
-        for (TSendQueue::iterator it = SendQueue.begin(); it != SendQueue.end(); ) {
-            const TSendQueue::iterator next = std::next(it);
-            if (it->IsEmpty()) {
-                SendQueueCache.splice(SendQueueCache.begin(), SendQueue, it);
-            } else {
-                it->ResetBufs();
-            }
-            it = next;
-        }
-        TrimSendQueueCache();
-        SendQueuePos = SendQueue.begin();
-
-        TMaybe<ui64> s;
-        for (auto it = SendQueuePos; it != SendQueue.end(); ++it) {
-            if (!it->IsEmpty()) {
-                s = it->GetSerial();
+        ui64 serial = Max<ui64>();
+        BytesUnwritten = 0;
+        for (const auto& packet : SendQueue) {
+            if (packet.Data && packet.Serial < serial) {
+                serial = packet.Serial;
             }
+            BytesUnwritten += packet.PacketSize;
         }
-        const ui64 serial = s.GetOrElse(Max<ui64>());
 
         Y_VERIFY(serial > LastConfirmed, "%s serial# %" PRIu64 " LastConfirmed# %" PRIu64, LogPrefix.data(), serial, LastConfirmed);
         LOG_DEBUG_IC_SESSION("ICS06", "rewind SendQueue size# %zu LastConfirmed# %" PRIu64 " SendQueuePos.Serial# %" PRIu64 "\n",
             SendQueue.size(), LastConfirmed, serial);
 
-        BytesUnwritten = 0;
-        for (const auto& packet : SendQueue) {
-            BytesUnwritten += sizeof(TTcpPacketHeader_v2) + packet.GetDataSize();
-        }
+        Y_VERIFY_DEBUG(BytesUnwritten == OutgoingStream.CalculateUnsentSize(), "%s", TString(TStringBuilder()
+            << "BytesUnwritten# " << BytesUnwritten
+            << " CalculateUnsentSize# " << OutgoingStream.CalculateUnsentSize()).data());
 
         SwitchStuckPeriod();
 
@@ -581,9 +571,21 @@ namespace NActors {
             LOG_DEBUG_IC_SESSION("ICS30", "WriteData WriteBlockedByFullSendBuffer# %s SendQueue.size# %zu",
                 ReceiveContext->WriteBlockedByFullSendBuffer ? "true" : "false", SendQueue.size());
 
-            while (SendQueuePos != SendQueue.end() && !ReceiveContext->WriteBlockedByFullSendBuffer) {
-                for (auto it = SendQueuePos; it != SendQueue.end() && wbuffers.size() < maxElementsInIOV; ++it) {
-                    it->AppendToIoVector(wbuffers, maxElementsInIOV);
+            auto calculateUnsentQueueSize = [&] {
+                size_t res = -SendOffset;
+                for (auto it = SendQueue.begin() + SendQueuePos; it != SendQueue.end(); ++it) {
+                    res += it->PacketSize;
+                }
+                return res;
+            };
+
+            while (!ReceiveContext->WriteBlockedByFullSendBuffer) {
+                Y_VERIFY_DEBUG(BytesUnwritten == OutgoingStream.CalculateUnsentSize());
+                Y_VERIFY_DEBUG(BytesUnwritten == calculateUnsentQueueSize());
+
+                OutgoingStream.ProduceIoVec(wbuffers, maxElementsInIOV);
+                if (!wbuffers) { // done sending
+                    break;
                 }
 
                 const struct iovec* iovec = reinterpret_cast<const struct iovec*>(wbuffers.data());
@@ -610,21 +612,31 @@ namespace NActors {
                 wbuffers.clear();
 
                 if (r > 0) {
+                    written += r;
+
                     Y_VERIFY(static_cast<size_t>(r) <= BytesUnwritten);
                     BytesUnwritten -= r;
-                    written += r;
-                    ui64 packets = 0;
 
-                    // advance SendQueuePos to eat all processed items
-                    for (size_t amount = r; amount && SendQueuePos->DropBufs(amount); ++SendQueuePos) {
-                        if (!SendQueuePos->IsEmpty()) {
-                            LastSentSerial = Max(LastSentSerial, SendQueuePos->GetSerial());
+                    OutgoingStream.Advance(r);
+
+                    ui64 packets = 0;
+                    Y_VERIFY_DEBUG(SendQueuePos != SendQueue.size());
+                    SendOffset += r;
+                    for (auto it = SendQueue.begin() + SendQueuePos; SendOffset && it->PacketSize <= SendOffset; ++SendQueuePos, ++it) {
+                        SendOffset -= it->PacketSize;
+                        Y_VERIFY_DEBUG(!it->Data || it->Serial <= OutputCounter);
+                        if (it->Data && LastSentSerial < it->Serial) {
+                            LastSentSerial = it->Serial;
                         }
                         ++PacketsWrittenToSocket;
                         ++packets;
-                        LWTRACK(PacketWrittenToSocket, SendQueuePos->Orbit, Proxy->PeerNodeId, PacketsWrittenToSocket, SendQueuePos->TriedWriting, SendQueuePos->GetDataSize(), BytesUnwritten, GetWriteBlockedTotal(), (SOCKET)*Socket);
+                        Y_VERIFY_DEBUG(SendOffset == 0 || SendQueuePos != SendQueue.size() - 1);
+//                        LWTRACK(PacketWrittenToSocket, SendQueuePos->Orbit, Proxy->PeerNodeId, PacketsWrittenToSocket, false, SendQueuePos->PacketSize, BytesUnwritten, GetWriteBlockedTotal(), (SOCKET)*Socket);
                     }
 
+                    Y_VERIFY_DEBUG(BytesUnwritten == OutgoingStream.CalculateUnsentSize());
+                    Y_VERIFY_DEBUG(BytesUnwritten == calculateUnsentQueueSize());
+
                     LWPROBE(WriteToSocket, Proxy->PeerNodeId, r, packets, PacketsWrittenToSocket, BytesUnwritten, GetWriteBlockedTotal(), (SOCKET)*Socket);
                 } else if (-r != EAGAIN && -r != EWOULDBLOCK) {
                     const TString message = r == 0 ? "connection closed by peer"
@@ -636,12 +648,6 @@ namespace NActors {
                     }
                     return ReestablishConnectionWithHandshake(r == 0 ? TDisconnectReason::EndOfStream() : TDisconnectReason::FromErrno(-r));
                 } else {
-                    // we have to do some hack for secure socket -- mark the packet as 'tried writing'
-                    if (Params.Encryption) {
-                        Y_VERIFY(SendQueuePos != SendQueue.end());
-                        SendQueuePos->MarkTriedWriting(); // do not try to replace buffer under SSL
-                    }
-
                     // we have received EAGAIN error code, this means that we can't issue more data until we have received
                     // TEvPollerReadyWrite event from poller; set up flag meaning this and wait for that event
                     Y_VERIFY(!ReceiveContext->WriteBlockedByFullSendBuffer);
@@ -711,45 +717,10 @@ namespace NActors {
         }
     }
 
-    void TInterconnectSessionTCP::TrimSendQueueCache() {
-        static constexpr size_t maxItems = 32;
-        static constexpr size_t trimThreshold = maxItems * 2;
-        if (SendQueueCache.size() >= trimThreshold) {
-            auto it = SendQueueCache.end();
-            for (size_t n = SendQueueCache.size() - maxItems; n; --n) {
-                --it;
-            }
-
-            auto ev = std::make_unique<TEvFreeItems>();
-            ev->Items.splice(ev->Items.end(), SendQueueCache, it, SendQueueCache.end());
-            ev->NumBytes = ev->Items.size() * sizeof(TTcpPacketOutTask);
-            if (ev->GetInLineForDestruction(Proxy->Common)) {
-                Send(Proxy->Common->DestructorId, ev.release());
-            }
-        }
-    }
-
     ui64 TInterconnectSessionTCP::MakePacket(bool data, TMaybe<ui64> pingMask) {
         Y_VERIFY(Socket);
 
-        TSendQueue::iterator packet;
-        if (SendQueueCache) {
-            // we have entries in cache, take one and move it to the end of SendQueue
-            packet = SendQueueCache.begin();
-            SendQueue.splice(SendQueue.end(), SendQueueCache, packet);
-            packet->Reuse(); // reset packet to initial state
-        } else {
-            // we have to allocate new packet, so just do it
-            LWPROBE_IF_TOO_LONG(SlowICAllocPacketBuffer, Proxy->PeerNodeId, ms) {
-                packet = SendQueue.emplace(SendQueue.end(), Params);
-            }
-        }
-
-        // update send queue position
-        if (SendQueuePos == SendQueue.end()) {
-            SendQueuePos = packet; // start sending this packet if we are not sending anything for now
-        }
-
+        TTcpPacketOutTask packet(Params, OutgoingStream);
         ui64 serial = 0;
 
         if (data) {
@@ -759,12 +730,12 @@ namespace NActors {
             // fill the data packet
             Y_VERIFY(NumEventsInReadyChannels);
             LWPROBE_IF_TOO_LONG(SlowICFillSendingBuffer, Proxy->PeerNodeId, ms) {
-                FillSendingBuffer(*packet, serial);
+                FillSendingBuffer(packet, serial);
             }
-            Y_VERIFY(!packet->IsEmpty());
+            Y_VERIFY(!packet.IsEmpty());
 
-            InflightDataAmount += packet->GetDataSize();
-            Proxy->Metrics->AddInflightDataAmount(packet->GetDataSize());
+            InflightDataAmount += packet.GetDataSize();
+            Proxy->Metrics->AddInflightDataAmount(packet.GetDataSize());
             if (InflightDataAmount > GetTotalInflightAmountOfData()) {
                 Proxy->Metrics->IncInflyLimitReach();
             }
@@ -778,39 +749,31 @@ namespace NActors {
             LastPayloadActivityTimestamp = TActivationContext::Monotonic();
         } else if (pingMask) {
             serial = *pingMask;
-
-            // make this packet a priority one
-            if (SendQueuePos != packet) {
-                Y_VERIFY(SendQueuePos != SendQueue.end());
-                if (SendQueuePos->IsAtBegin()) {
-                    // insert this packet just before the next being sent and step back
-                    SendQueue.splice(SendQueuePos, SendQueue, packet);
-                    --SendQueuePos;
-                    Y_VERIFY(SendQueuePos == packet);
-                } else {
-                    // current packet is already being sent, so move new packet just after it
-                    SendQueue.splice(std::next(SendQueuePos), SendQueue, packet);
-                }
-            }
         }
 
         const ui64 lastInputSerial = ReceiveContext->GetLastProcessedPacketSerial();
-        packet->SetMetadata(serial, lastInputSerial);
-        packet->Sign();
+
+        packet.Finish(serial, lastInputSerial);
 
         // count number of bytes pending for write
-        ui64 packetSize = sizeof(TTcpPacketHeader_v2) + packet->GetDataSize();
+        const size_t packetSize = packet.GetFullSize();
         BytesUnwritten += packetSize;
+        Y_VERIFY_DEBUG(BytesUnwritten == OutgoingStream.CalculateUnsentSize(), "%s", TString(TStringBuilder()
+                << "BytesUnwritten# " << BytesUnwritten << " packetSize# " << packetSize
+                << " CalculateUnsentSize# " << OutgoingStream.CalculateUnsentSize()).data());
+
+        // put outgoing packet metadata here
+        SendQueue.push_back(TOutgoingPacket{packetSize, serial, data});
 
         LOG_DEBUG_IC_SESSION("ICS22", "outgoing packet Serial# %" PRIu64 " Confirm# %" PRIu64 " DataSize# %zu"
-            " InflightDataAmount# %" PRIu64 " BytesUnwritten# %" PRIu64, serial, lastInputSerial, packet->GetDataSize(),
+            " InflightDataAmount# %" PRIu64 " BytesUnwritten# %" PRIu64, serial, lastInputSerial, packet.GetDataSize(),
             InflightDataAmount, BytesUnwritten);
 
         // reset forced packet sending timestamp as we have confirmed all received data
         ResetFlushLogic();
 
         ++PacketsGenerated;
-        LWTRACK(PacketGenerated, packet->Orbit, Proxy->PeerNodeId, BytesUnwritten, InflightDataAmount, PacketsGenerated, packetSize);
+        LWTRACK(PacketGenerated, packet.Orbit, Proxy->PeerNodeId, BytesUnwritten, InflightDataAmount, PacketsGenerated, packetSize);
 
         if (!data) {
             WriteData();
@@ -819,7 +782,7 @@ namespace NActors {
         return packetSize;
     }
 
-    bool TInterconnectSessionTCP::DropConfirmed(ui64 confirm) {
+    bool TInterconnectSessionTCP::DropConfirmed(ui64 confirm, bool ignoreSendQueuePos) {
         LOG_DEBUG_IC_SESSION("ICS23", "confirm count: %" PRIu64, confirm);
 
         Y_VERIFY(LastConfirmed <= confirm && confirm <= LastSentSerial && LastSentSerial <= OutputCounter,
@@ -828,24 +791,48 @@ namespace NActors {
         LastConfirmed = confirm;
 
         ui64 droppedDataAmount = 0;
+        std::optional<ui64> lastDroppedSerial = 0;
         ui32 numDropped = 0;
 
+#ifndef NDEBUG
+        {
+            size_t totalBytesInSendQueue = 0;
+            size_t unsentBytesInSendQueue = 0;
+            for (size_t i = 0; i < SendQueue.size(); ++i) {
+                totalBytesInSendQueue += SendQueue[i].PacketSize;
+                if (i > SendQueuePos) {
+                    unsentBytesInSendQueue += SendQueue[i].PacketSize;
+                } else if (i == SendQueuePos) {
+                    unsentBytesInSendQueue += SendQueue[i].PacketSize - SendOffset;
+                }
+            }
+            Y_VERIFY(totalBytesInSendQueue == OutgoingStream.CalculateOutgoingSize());
+            Y_VERIFY(ignoreSendQueuePos || unsentBytesInSendQueue == OutgoingStream.CalculateUnsentSize());
+        }
+#endif
+
         // drop confirmed packets; this also includes any auxiliary packets as their serial is set to zero, effectively
         // making Serial <= confirm true
-        TSendQueue::iterator it;
-        ui64 lastDroppedSerial = 0;
-        for (it = SendQueue.begin(); it != SendQueuePos && it->Confirmed(confirm); ++it) {
-            if (!it->IsEmpty()) {
-                lastDroppedSerial = it->GetSerial();
+        size_t bytesDropped = 0;
+        for (; !SendQueue.empty(); SendQueue.pop_front(), --SendQueuePos) {
+            auto& front = SendQueue.front();
+            if (front.Data && confirm < front.Serial) {
+                break;
+            }
+            if (front.Data) {
+                lastDroppedSerial.emplace(front.Serial);
             }
-            droppedDataAmount += it->GetDataSize();
+            droppedDataAmount += front.PacketSize - sizeof(TTcpPacketHeader_v2);
+            bytesDropped += front.PacketSize;
             ++numDropped;
+            Y_VERIFY_DEBUG(ignoreSendQueuePos || SendQueuePos != 0);
+        }
+        OutgoingStream.DropFront(bytesDropped);
+        if (lastDroppedSerial) {
+            ChannelScheduler->ForEach([&](TEventOutputChannel& channel) {
+                channel.DropConfirmed(*lastDroppedSerial);
+            });
         }
-        SendQueueCache.splice(SendQueueCache.begin(), SendQueue, SendQueue.begin(), it);
-        TrimSendQueueCache();
-        ChannelScheduler->ForEach([&](TEventOutputChannel& channel) {
-            channel.DropConfirmed(lastDroppedSerial);
-        });
 
         const ui64 current = InflightDataAmount;
         const ui64 limit = GetTotalInflightAmountOfData();
@@ -1202,7 +1189,6 @@ namespace NActors {
 
                             MON_VAR(OutputStuckFlag)
                             MON_VAR(SendQueue.size())
-                            MON_VAR(SendQueueCache.size())
                             MON_VAR(NumEventsInReadyChannels)
                             MON_VAR(TotalOutputQueueSize)
                             MON_VAR(BytesUnwritten)
diff --git a/library/cpp/actors/interconnect/interconnect_tcp_session.h b/library/cpp/actors/interconnect/interconnect_tcp_session.h
index 7d0e003a34..48fa39b273 100644
--- a/library/cpp/actors/interconnect/interconnect_tcp_session.h
+++ b/library/cpp/actors/interconnect/interconnect_tcp_session.h
@@ -24,6 +24,7 @@
 #include "watchdog_timer.h"
 #include "event_holder_pool.h"
 #include "channel_scheduler.h"
+#include "outgoing_stream.h"
 
 #include <unordered_set>
 #include <unordered_map>
@@ -217,10 +218,6 @@ namespace NActors {
         const ui32 NodeId;
         const TSessionParams Params;
 
-        // header we are currently processing (parsed from the stream)
-        TTcpPacketHeader_v2 Header;
-        ui64 HeaderConfirm, HeaderSerial;
-
         size_t PayloadSize;
         ui32 ChecksumExpected, Checksum;
         bool IgnorePayload;
@@ -233,6 +230,7 @@ namespace NActors {
 
         THolder<TEvUpdateFromInputSession> UpdateFromInputSession;
 
+        std::optional<ui64> LastReceivedSerial;
         ui64 ConfirmedByInput;
 
         std::shared_ptr<IInterconnectMetrics> Metrics;
@@ -390,7 +388,7 @@ namespace NActors {
 
         ui64 MakePacket(bool data, TMaybe<ui64> pingMask = {});
         void FillSendingBuffer(TTcpPacketOutTask& packet, ui64 serial);
-        bool DropConfirmed(ui64 confirm);
+        bool DropConfirmed(ui64 confirm, bool ignoreSendQueuePos = false);
         void ShutdownSocket(TDisconnectReason reason);
 
         void StartHandshake();
@@ -447,16 +445,21 @@ namespace NActors {
         void SetOutputStuckFlag(bool state);
         void SwitchStuckPeriod();
 
-        using TSendQueue = TList<TTcpPacketOutTask>;
-        TSendQueue SendQueue;
-        TSendQueue SendQueueCache;
-        TSendQueue::iterator SendQueuePos;
+        NInterconnect::TOutgoingStream OutgoingStream;
+
+        struct TOutgoingPacket {
+            size_t PacketSize;
+            ui64 Serial;
+            bool Data;
+        };
+        std::deque<TOutgoingPacket> SendQueue; // packet boundaries
+        size_t SendQueuePos = 0; // packet being sent now
+        size_t SendOffset = 0;
+
         ui64 WriteBlockedCycles = 0; // start of current block period
         TDuration WriteBlockedTotal; // total incremental duration that session has been blocked
         ui64 BytesUnwritten = 0;
 
-        void TrimSendQueueCache();
-
         TDuration GetWriteBlockedTotal() const {
             if (ReceiveContext->WriteBlockedByFullSendBuffer) {
                 double blockedUs = NHPTimer::GetSeconds(GetCycleCountFast() - WriteBlockedCycles) * 1000000.0;
diff --git a/library/cpp/actors/interconnect/outgoing_stream.h b/library/cpp/actors/interconnect/outgoing_stream.h
new file mode 100644
index 0000000000..0306295b52
--- /dev/null
+++ b/library/cpp/actors/interconnect/outgoing_stream.h
@@ -0,0 +1,224 @@
+#pragma once
+
+#include <library/cpp/actors/core/event_load.h>
+#include <library/cpp/actors/util/rc_buf.h>
+#include <library/cpp/containers/stack_vector/stack_vec.h>
+#include <deque>
+
+namespace NInterconnect {
+
+    template<size_t TotalSize>
+    class TOutgoingStreamT {
+        static constexpr size_t BufferSize = TotalSize - sizeof(ui32) * 2;
+
+        struct TBuffer {
+            char Data[BufferSize];
+            ui32 RefCount;
+            ui32 Index;
+
+            struct TDeleter {
+                void operator ()(TBuffer *buffer) const {
+                    free(buffer);
+                }
+            };
+        };
+
+        static_assert(sizeof(TBuffer) == TotalSize);
+
+        struct TSendChunk {
+            TContiguousSpan Span;
+            TBuffer *Buffer;
+        };
+
+        std::vector<std::unique_ptr<TBuffer, typename TBuffer::TDeleter>> Buffers;
+        TBuffer *AppendBuffer = nullptr;
+        size_t AppendOffset = BufferSize; // into the last buffer
+        std::deque<TSendChunk> SendQueue;
+        size_t SendQueuePos = 0;
+        size_t SendOffset = 0;
+
+    public:
+        size_t CalculateOutgoingSize() const {
+            size_t res = 0;
+            for (const TSendChunk& chunk : SendQueue) {
+                res += chunk.Span.size();
+            }
+            return res;
+        }
+
+        size_t CalculateUnsentSize() const {
+            size_t res = 0;
+            for (auto it = SendQueue.begin() + SendQueuePos; it != SendQueue.end(); ++it) {
+                res += it->Span.size();
+            }
+            return res - SendOffset;
+        }
+
+        TMutableContiguousSpan AcquireSpanForWriting(size_t maxLen) {
+            if (AppendOffset == BufferSize) { // we have no free buffer, allocate one
+                Buffers.emplace_back(static_cast<TBuffer*>(malloc(sizeof(TBuffer))));
+                AppendBuffer = Buffers.back().get();
+                Y_VERIFY(AppendBuffer);
+                AppendBuffer->RefCount = 1; // through AppendBuffer pointer
+                AppendBuffer->Index = Buffers.size() - 1;
+                AppendOffset = 0;
+            }
+            return {AppendBuffer->Data + AppendOffset, Min(maxLen, BufferSize - AppendOffset)};
+        }
+
+        void Append(TContiguousSpan span) {
+            TBuffer *buffer = nullptr;
+            if (AppendBuffer && span.data() == AppendBuffer->Data + AppendOffset) { // the only valid case to use previously acquired span
+                buffer = AppendBuffer;
+                AppendOffset += span.size();
+                Y_VERIFY_DEBUG(AppendOffset <= BufferSize);
+                if (AppendOffset != BufferSize) {
+                    ++buffer->RefCount;
+                }
+            } else {
+#ifndef NDEBUG
+                // ensure this span does not point into any existing buffer part
+                const char *begin = span.data();
+                const char *end = span.data() + span.size();
+                for (const auto& buffer : Buffers) {
+                    const char *bufferBegin = buffer->Data;
+                    const char *bufferEnd = bufferBegin + BufferSize;
+                    if (bufferBegin < end && begin < bufferEnd) {
+                        Y_FAIL();
+                    }
+                }
+#endif
+            }
+
+            if (!SendQueue.empty()) {
+                auto& back = SendQueue.back();
+                if (back.Span.data() + back.Span.size() == span.data()) { // check if it is possible just to extend the last span
+                    if (SendQueuePos == SendQueue.size()) {
+                        --SendQueuePos;
+                        SendOffset = back.Span.size();
+                    }
+                    back.Span = {back.Span.data(), back.Span.size() + span.size()};
+                    DropBufferReference(buffer);
+                    return;
+                }
+            }
+
+            if (buffer) {
+                ++buffer->RefCount;
+            }
+            SendQueue.push_back(TSendChunk{span, buffer});
+            DropBufferReference(buffer);
+        }
+
+        void Write(TContiguousSpan in) {
+            while (in.size()) {
+                auto outChunk = AcquireSpanForWriting(in.size());
+                Append(outChunk);
+                memcpy(outChunk.data(), in.data(), outChunk.size());
+                in = in.SubSpan(outChunk.size(), Max<size_t>());
+            }
+        }
+
+        using TBookmark = TStackVec<TMutableContiguousSpan, 2>;
+
+        TBookmark Bookmark(size_t len) {
+            TBookmark bookmark;
+
+            while (len) {
+                const auto span = AcquireSpanForWriting(len);
+                Append(span);
+                bookmark.push_back(span);
+                len -= span.size();
+            }
+
+            return bookmark;
+        }
+
+        void WriteBookmark(TBookmark&& bookmark, TContiguousSpan in) {
+            for (auto& outChunk : bookmark) {
+                Y_VERIFY(outChunk.size() <= in.size());
+                memcpy(outChunk.data(), in.data(), outChunk.size());
+                in = in.SubSpan(outChunk.size(), Max<size_t>());
+            }
+        }
+
+        void Rewind() {
+            SendQueuePos = 0;
+            SendOffset = 0;
+        }
+
+        template<typename T>
+        void ProduceIoVec(T& container, size_t maxItems) {
+            size_t offset = SendOffset;
+            for (auto it = SendQueue.begin() + SendQueuePos; it != SendQueue.end() && std::size(container) < maxItems; ++it) {
+                const TContiguousSpan span = it->Span.SubSpan(offset, Max<size_t>());
+                container.push_back(NActors::TConstIoVec{span.data(), span.size()});
+                offset = 0;
+            }
+        }
+
+        void Advance(size_t numBytes) { // called when numBytes portion of data has been sent
+            Y_VERIFY_DEBUG(numBytes == 0 || SendQueuePos != SendQueue.size());
+            SendOffset += numBytes;
+            for (auto it = SendQueue.begin() + SendQueuePos; SendOffset && it->Span.size() <= SendOffset; ++SendQueuePos, ++it) {
+                SendOffset -= it->Span.size();
+                Y_VERIFY_DEBUG(SendOffset == 0 || SendQueuePos != SendQueue.size() - 1);
+            }
+        }
+
+        void DropFront(size_t numBytes) { // drops first numBytes from the queue, freeing buffers when necessary
+            while (numBytes) {
+                Y_VERIFY_DEBUG(!SendQueue.empty());
+                auto& front = SendQueue.front();
+                if (numBytes < front.Span.size()) {
+                    front.Span = front.Span.SubSpan(numBytes, Max<size_t>());
+                    if (SendQueuePos == 0) {
+                        Y_VERIFY_DEBUG(numBytes <= SendOffset);
+                        SendOffset -= numBytes;
+                    }
+                    break;
+                } else {
+                    numBytes -= front.Span.size();
+                }
+                Y_VERIFY_DEBUG(!front.Buffer || (front.Span.data() >= front.Buffer->Data &&
+                    front.Span.data() + front.Span.size() <= front.Buffer->Data + BufferSize));
+                DropBufferReference(front.Buffer);
+                SendQueue.pop_front();
+                if (SendQueuePos) {
+                    --SendQueuePos;
+                } else {
+                    SendOffset = 0;
+                }
+            }
+        }
+
+        template<typename T>
+        void ScanLastBytes(size_t numBytes, T&& callback) const {
+            auto it = SendQueue.end();
+            ssize_t offset = -numBytes;
+            while (offset < 0) {
+                Y_VERIFY_DEBUG(it != SendQueue.begin());
+                const TSendChunk& chunk = *--it;
+                offset += chunk.Span.size();
+            }
+            for (; it != SendQueue.end(); ++it, offset = 0) {
+                callback(it->Span.SubSpan(offset, Max<size_t>()));
+            }
+        }
+
+    private:
+        void DropBufferReference(TBuffer *buffer) {
+            if (buffer && !--buffer->RefCount) {
+                const size_t index = buffer->Index;
+                auto& cell = Buffers[index];
+                Y_VERIFY_DEBUG(cell.get() == buffer);
+                std::swap(cell, Buffers.back());
+                cell->Index = index;
+                Buffers.pop_back();
+            }
+        }
+    };
+
+    using TOutgoingStream = TOutgoingStreamT<262144>;
+
+} // NInterconnect
diff --git a/library/cpp/actors/interconnect/packet.h b/library/cpp/actors/interconnect/packet.h
index 501fb08ab1..c1d63fa3b8 100644
--- a/library/cpp/actors/interconnect/packet.h
+++ b/library/cpp/actors/interconnect/packet.h
@@ -14,6 +14,7 @@
 #include <util/generic/list.h>
 
 #include "types.h"
+#include "outgoing_stream.h"
 
 #ifndef FORCE_EVENT_CHECKSUM
 #define FORCE_EVENT_CHECKSUM 0
@@ -45,9 +46,6 @@ struct TTcpPacketBuf {
     static constexpr ui64 ClockMask = 0x2000000000000000ULL;
 
     static constexpr size_t PacketDataLen = 4096 * 2 - 96 - sizeof(TTcpPacketHeader_v2);
-
-    TTcpPacketHeader_v2 Header;
-    char Data[PacketDataLen];
 };
 
 struct TEventData {
@@ -121,144 +119,83 @@ namespace NActors {
 
 struct TTcpPacketOutTask : TNonCopyable {
     const TSessionParams& Params;
-    TTcpPacketBuf Packet;
-    size_t DataSize;
-    TStackVec<TConstIoVec, 32> Bufs;
-    size_t BufferIndex;
-    size_t FirstBufferOffset;
-    bool TriedWriting;
-    char *FreeArea;
-    char *End;
+    NInterconnect::TOutgoingStream& OutgoingStream;
+    NInterconnect::TOutgoingStream::TBookmark HeaderBookmark;
+    size_t DataSize = 0;
     mutable NLWTrace::TOrbit Orbit;
 
-public:
-    TTcpPacketOutTask(const TSessionParams& params)
+    TTcpPacketOutTask(const TSessionParams& params, NInterconnect::TOutgoingStream& outgoingStream)
         : Params(params)
-    {
-        Reuse();
-    }
-
-    bool IsAtBegin() const {
-        return !BufferIndex && !FirstBufferOffset && !TriedWriting;
-    }
-
-    void MarkTriedWriting() {
-        TriedWriting = true;
-    }
+        , OutgoingStream(outgoingStream)
+        , HeaderBookmark(OutgoingStream.Bookmark(sizeof(TTcpPacketHeader_v2)))
+    {}
 
-    void Reuse() {
-        DataSize = 0;
-        Bufs.assign(1, {&Packet.Header, sizeof(Packet.Header)});
-        BufferIndex = 0;
-        FirstBufferOffset = 0;
-        TriedWriting = false;
-        FreeArea = Packet.Data;
-        End = FreeArea + TTcpPacketBuf::PacketDataLen;
-        Orbit.Reset();
-    }
-
-    bool IsEmpty() const {
-        return !DataSize;
+    // Preallocate some space to fill it later.
+    NInterconnect::TOutgoingStream::TBookmark Bookmark(size_t len) {
+        Y_VERIFY_DEBUG(len <= GetVirtualFreeAmount());
+        DataSize += len;
+        return OutgoingStream.Bookmark(len);
     }
 
-    void SetMetadata(ui64 serial, ui64 confirm) {
-        Packet.Header.Serial = serial;
-        Packet.Header.Confirm = confirm;
-    }
-
-    size_t GetDataSize() const { return DataSize; }
-
-    ui64 GetSerial() const {
-        return Packet.Header.Serial;
+    // Write previously bookmarked space.
+    void WriteBookmark(NInterconnect::TOutgoingStream::TBookmark&& bookmark, const void *buffer, size_t len) {
+        OutgoingStream.WriteBookmark(std::move(bookmark), {static_cast<const char*>(buffer), len});
     }
 
-    bool Confirmed(ui64 confirm) const {
-        return IsEmpty() || Packet.Header.Serial <= confirm;
+    // Acquire raw pointer to write some data.
+    TMutableContiguousSpan AcquireSpanForWriting() {
+        return OutgoingStream.AcquireSpanForWriting(GetVirtualFreeAmount());
     }
 
-    void *GetFreeArea() {
-        return FreeArea;
+    // Append reference to some data (acquired previously or external pointer).
+    void Append(const void *buffer, size_t len) {
+        Y_VERIFY_DEBUG(len <= GetVirtualFreeAmount());
+        DataSize += len;
+        OutgoingStream.Append({static_cast<const char*>(buffer), len});
     }
 
-    size_t GetVirtualFreeAmount() const {
-        return TTcpPacketBuf::PacketDataLen - DataSize;
+    // Write some data with copying.
+    void Write(const void *buffer, size_t len) {
+        Y_VERIFY_DEBUG(len <= GetVirtualFreeAmount());
+        DataSize += len;
+        OutgoingStream.Write({static_cast<const char*>(buffer), len});
     }
 
-    void AppendBuf(const void *buf, size_t size) {
-        DataSize += size;
-        Y_VERIFY_DEBUG(DataSize <= TTcpPacketBuf::PacketDataLen, "DataSize# %zu AppendBuf buf# %p size# %zu"
-            " FreeArea# %p End# %p", DataSize, buf, size, FreeArea, End);
-
-        if (Bufs && static_cast<const char*>(Bufs.back().Data) + Bufs.back().Size == buf) {
-            Bufs.back().Size += size;
-        } else {
-            Bufs.push_back({buf, size});
-        }
+    void Finish(ui64 serial, ui64 confirm) {
+        Y_VERIFY(DataSize <= Max<ui16>());
 
-        if (buf >= FreeArea && buf < End) {
-            Y_VERIFY_DEBUG(buf == FreeArea);
-            FreeArea = const_cast<char*>(static_cast<const char*>(buf)) + size;
-            Y_VERIFY_DEBUG(FreeArea <= End);
-        }
-    }
+        TTcpPacketHeader_v2 header{
+            confirm,
+            serial,
+            0,
+            static_cast<ui16>(DataSize)
+        };
 
-    void Undo(size_t size) {
-        Y_VERIFY(Bufs);
-        auto& buf = Bufs.back();
-        Y_VERIFY(buf.Data == FreeArea - buf.Size);
-        buf.Size -= size;
-        if (!buf.Size) {
-            Bufs.pop_back();
-        }
-        FreeArea -= size;
-        DataSize -= size;
-    }
+        if (Checksumming()) {
+            // pre-write header without checksum for correct checksum calculation
+            WriteBookmark(NInterconnect::TOutgoingStream::TBookmark(HeaderBookmark), &header, sizeof(header));
 
-    bool DropBufs(size_t& amount) {
-        while (BufferIndex != Bufs.size()) {
-            TConstIoVec& item = Bufs[BufferIndex];
-            // calculate number of bytes to the end in current buffer
-            const size_t remain = item.Size - FirstBufferOffset;
-            if (amount >= remain) {
-                // vector item completely fits into the received amount, drop it out and switch to next buffer
-                amount -= remain;
-                ++BufferIndex;
-                FirstBufferOffset = 0;
-            } else {
-                // adjust first buffer by "amount" bytes forward and reset amount to zero
-                FirstBufferOffset += amount;
-                amount = 0;
-                // return false meaning that we have some more data to send
-                return false;
-            }
+            size_t total = 0;
+            ui32 checksum = 0;
+            OutgoingStream.ScanLastBytes(GetFullSize(), [&](TContiguousSpan span) {
+                checksum = Crc32cExtendMSanCompatible(checksum, span.data(), span.size());
+                total += span.size();
+            });
+            header.Checksum = checksum;
+            Y_VERIFY(total == sizeof(header) + DataSize, "total# %zu DataSize# %zu GetFullSize# %zu", total, DataSize,
+                GetFullSize());
         }
-        return true;
-    }
 
-    void ResetBufs() {
-        BufferIndex = FirstBufferOffset = 0;
-        TriedWriting = false;
+        WriteBookmark(std::exchange(HeaderBookmark, {}), &header, sizeof(header));
     }
 
-    template <typename TVectorType>
-    void AppendToIoVector(TVectorType& vector, size_t max) {
-        for (size_t k = BufferIndex, offset = FirstBufferOffset; k != Bufs.size() && vector.size() < max; ++k, offset = 0) {
-            TConstIoVec v = Bufs[k];
-            v.Data = static_cast<const char*>(v.Data) + offset;
-            v.Size -= offset;
-            vector.push_back(v);
-        }
+    bool Checksumming() const {
+        return !Params.Encryption;
     }
 
-    void Sign() {
-        Packet.Header.Checksum = 0;
-        Packet.Header.PayloadLength = DataSize;
-        if (!Params.Encryption) {
-            ui32 sum = 0;
-            for (const auto& item : Bufs) {
-                sum = Crc32cExtendMSanCompatible(sum, item.Data, item.Size);
-            }
-            Packet.Header.Checksum = sum;
-        }
-    }
+    bool IsFull() const { return GetVirtualFreeAmount() == 0; }
+    bool IsEmpty() const { return GetDataSize() == 0; }
+    size_t GetDataSize() const { return DataSize; }
+    size_t GetFullSize() const { return sizeof(TTcpPacketHeader_v2) + GetDataSize(); }
+    size_t GetVirtualFreeAmount() const { return TTcpPacketBuf::PacketDataLen - DataSize; }
 };
diff --git a/library/cpp/actors/interconnect/ut/CMakeLists.darwin-x86_64.txt b/library/cpp/actors/interconnect/ut/CMakeLists.darwin-x86_64.txt
index e1e00ed6ad..1b744e906c 100644
--- a/library/cpp/actors/interconnect/ut/CMakeLists.darwin-x86_64.txt
+++ b/library/cpp/actors/interconnect/ut/CMakeLists.darwin-x86_64.txt
@@ -35,6 +35,7 @@ target_sources(library-cpp-actors-interconnect-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/event_holder_pool_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/interconnect_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/large.cpp
+  ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/outgoing_stream_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/poller_actor_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/dynamic_proxy_ut.cpp
 )
diff --git a/library/cpp/actors/interconnect/ut/CMakeLists.linux-aarch64.txt b/library/cpp/actors/interconnect/ut/CMakeLists.linux-aarch64.txt
index a869318b87..b77f4f4e6c 100644
--- a/library/cpp/actors/interconnect/ut/CMakeLists.linux-aarch64.txt
+++ b/library/cpp/actors/interconnect/ut/CMakeLists.linux-aarch64.txt
@@ -38,6 +38,7 @@ target_sources(library-cpp-actors-interconnect-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/event_holder_pool_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/interconnect_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/large.cpp
+  ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/outgoing_stream_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/poller_actor_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/dynamic_proxy_ut.cpp
 )
diff --git a/library/cpp/actors/interconnect/ut/CMakeLists.linux-x86_64.txt b/library/cpp/actors/interconnect/ut/CMakeLists.linux-x86_64.txt
index 479e70158a..96489d5e6b 100644
--- a/library/cpp/actors/interconnect/ut/CMakeLists.linux-x86_64.txt
+++ b/library/cpp/actors/interconnect/ut/CMakeLists.linux-x86_64.txt
@@ -39,6 +39,7 @@ target_sources(library-cpp-actors-interconnect-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/event_holder_pool_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/interconnect_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/large.cpp
+  ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/outgoing_stream_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/poller_actor_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/dynamic_proxy_ut.cpp
 )
diff --git a/library/cpp/actors/interconnect/ut/CMakeLists.windows-x86_64.txt b/library/cpp/actors/interconnect/ut/CMakeLists.windows-x86_64.txt
index 61244b1f9d..ddae9008cf 100644
--- a/library/cpp/actors/interconnect/ut/CMakeLists.windows-x86_64.txt
+++ b/library/cpp/actors/interconnect/ut/CMakeLists.windows-x86_64.txt
@@ -28,6 +28,7 @@ target_sources(library-cpp-actors-interconnect-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/event_holder_pool_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/interconnect_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/large.cpp
+  ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/outgoing_stream_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/poller_actor_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/dynamic_proxy_ut.cpp
 )
diff --git a/library/cpp/actors/interconnect/ut/channel_scheduler_ut.cpp b/library/cpp/actors/interconnect/ut/channel_scheduler_ut.cpp
index 32c8237b59..2de8dce457 100644
--- a/library/cpp/actors/interconnect/ut/channel_scheduler_ut.cpp
+++ b/library/cpp/actors/interconnect/ut/channel_scheduler_ut.cpp
@@ -43,8 +43,10 @@ Y_UNIT_TEST_SUITE(ChannelScheduler) {
 
         std::deque<std::map<ui16, ui32>> window;
 
+        NInterconnect::TOutgoingStream stream;
+
         for (; numEvents; ++step) {
-            TTcpPacketOutTask task(p);
+            TTcpPacketOutTask task(p, stream);
 
             if (step == 100) {
                 for (ui32 i = 0; i < 200; ++i) {
diff --git a/library/cpp/actors/interconnect/ut/lib/node.h b/library/cpp/actors/interconnect/ut/lib/node.h
index 0b538cdb1c..e15baf639e 100644
--- a/library/cpp/actors/interconnect/ut/lib/node.h
+++ b/library/cpp/actors/interconnect/ut/lib/node.h
@@ -2,6 +2,7 @@
 
 #include <library/cpp/actors/core/actorsystem.h>
 #include <library/cpp/actors/core/executor_pool_basic.h>
+#include <library/cpp/actors/core/executor_pool_io.h>
 #include <library/cpp/actors/core/scheduler_basic.h>
 #include <library/cpp/actors/core/mailbox.h>
 #include <library/cpp/actors/dnsresolver/dnsresolver.h>
@@ -24,11 +25,10 @@ public:
           TIntrusivePtr<NLog::TSettings> loggerSettings = nullptr) {
         TActorSystemSetup setup;
         setup.NodeId = nodeId;
-        setup.ExecutorsCount = 1;
+        setup.ExecutorsCount = 2;
         setup.Executors.Reset(new TAutoPtr<IExecutorPool>[setup.ExecutorsCount]);
-        for (ui32 i = 0; i < setup.ExecutorsCount; ++i) {
-            setup.Executors[i].Reset(new TBasicExecutorPool(i, numThreads, 20 /* magic number */));
-        }
+        setup.Executors[0].Reset(new TBasicExecutorPool(0, numThreads, 20 /* magic number */));
+        setup.Executors[1].Reset(new TIOExecutorPool(1, 1));
         setup.Scheduler.Reset(new TBasicSchedulerThread());
         const ui32 interconnectPoolId = 0;
 
@@ -108,8 +108,7 @@ public:
         // register logger
         setup.LocalServices.emplace_back(loggerActorId, TActorSetupCmd(new TLoggerActor(loggerSettings,
             CreateStderrBackend(), counters->GetSubgroup("subsystem", "logger")),
-            TMailboxType::ReadAsFilled, interconnectPoolId));
-
+            TMailboxType::ReadAsFilled, 1));
 
         if (common->OutgoingHandshakeInflightLimit) {
             // create handshake broker actor
diff --git a/library/cpp/actors/interconnect/ut/outgoing_stream_ut.cpp b/library/cpp/actors/interconnect/ut/outgoing_stream_ut.cpp
new file mode 100644
index 0000000000..14fcf25c2c
--- /dev/null
+++ b/library/cpp/actors/interconnect/ut/outgoing_stream_ut.cpp
@@ -0,0 +1,147 @@
+#include <library/cpp/actors/interconnect/outgoing_stream.h>
+#include <library/cpp/testing/unittest/registar.h>
+#include <util/random/entropy.h>
+#include <util/stream/null.h>
+
+#define Ctest Cnull
+
+Y_UNIT_TEST_SUITE(OutgoingStream) {
+    Y_UNIT_TEST(Basic) {
+        std::vector<char> buffer;
+        buffer.resize(4 << 20);
+
+        TReallyFastRng32 rng(EntropyPool());
+        for (char *p = buffer.data(); p != buffer.data() + buffer.size(); p += sizeof(ui32)) {
+            *reinterpret_cast<ui32*>(p) = rng();
+        }
+
+        for (ui32 nIter = 0; nIter < 10; ++nIter) {
+            Cerr << "nIter# " << nIter << Endl;
+
+            size_t base = 0; // number of dropped bytes
+            size_t sendOffset = 0; // offset to base
+            size_t pending = 0; // number of bytes in queue
+
+            NInterconnect::TOutgoingStreamT<4096> stream;
+
+            size_t numRewindsRemain = 10;
+
+            while (base != buffer.size()) {
+                const size_t bytesToEnd = buffer.size() - (base + sendOffset);
+
+                Ctest << "base# " << base << " sendOffset# " << sendOffset << " pending# " << pending
+                    << " bytesToEnd# " << bytesToEnd;
+
+                UNIT_ASSERT_VALUES_EQUAL(stream.CalculateOutgoingSize(), pending + sendOffset);
+                UNIT_ASSERT_VALUES_EQUAL(stream.CalculateUnsentSize(), pending);
+
+                const size_t maxBuffers = 128;
+                std::vector<NActors::TConstIoVec> iov;
+                stream.ProduceIoVec(iov, maxBuffers);
+                size_t offset = base + sendOffset;
+                for (const auto& [ptr, len] : iov) {
+                    UNIT_ASSERT(memcmp(buffer.data() + offset, ptr, len) == 0);
+                    offset += len;
+                }
+                UNIT_ASSERT(iov.size() == maxBuffers || offset == base + sendOffset + pending);
+
+                const char *nextData = buffer.data() + base + sendOffset + pending;
+                const size_t nextDataMaxLen = bytesToEnd - pending;
+                const size_t nextDataLen = nextDataMaxLen ? rng() % Min<size_t>(16384, nextDataMaxLen) + 1 : 0;
+
+                if (size_t bytesToScan = sendOffset + pending) {
+                    bytesToScan = rng() % bytesToScan + 1;
+                    size_t offset = base + sendOffset + pending - bytesToScan;
+                    stream.ScanLastBytes(bytesToScan, [&](TContiguousSpan span) {
+                        UNIT_ASSERT(offset + span.size() <= base + sendOffset + pending);
+                        UNIT_ASSERT(memcmp(buffer.data() + offset, span.data(), span.size()) == 0);
+                        offset += span.size();
+                    });
+                    UNIT_ASSERT_VALUES_EQUAL(offset, base + sendOffset + pending);
+                }
+
+                enum class EAction {
+                    COPY_APPEND,
+                    WRITE,
+                    REF_APPEND,
+                    ADVANCE,
+                    REWIND,
+                    DROP,
+                    BOOKMARK
+                };
+
+                std::vector<EAction> actions;
+                if (nextDataLen) {
+                    actions.push_back(EAction::COPY_APPEND);
+                    actions.push_back(EAction::WRITE);
+                    actions.push_back(EAction::REF_APPEND);
+                    actions.push_back(EAction::BOOKMARK);
+                }
+                if (numRewindsRemain && sendOffset > 65536) {
+                    actions.push_back(EAction::REWIND);
+                }
+                actions.push_back(EAction::ADVANCE);
+                actions.push_back(EAction::DROP);
+
+                switch (actions[rng() % actions.size()]) {
+                    case EAction::COPY_APPEND: {
+                        Ctest << " COPY_APPEND nextDataLen# " << nextDataLen;
+                        auto span = stream.AcquireSpanForWriting(nextDataLen);
+                        UNIT_ASSERT(span.size() != 0);
+                        memcpy(span.data(), nextData, span.size());
+                        stream.Append(span);
+                        pending += span.size();
+                        break;
+                    }
+
+                    case EAction::WRITE:
+                        Ctest << " WRITE nextDataLen# " << nextDataLen;
+                        stream.Write({nextData, nextDataLen});
+                        pending += nextDataLen;
+                        break;
+
+                    case EAction::REF_APPEND:
+                        Ctest << " REF_APPEND nextDataLen# " << nextDataLen;
+                        stream.Append({nextData, nextDataLen});
+                        pending += nextDataLen;
+                        break;
+
+                    case EAction::ADVANCE: {
+                        const size_t advance = rng() % Min<size_t>(4096, pending + 1);
+                        Ctest << " ADVANCE advance# " << advance;
+                        stream.Advance(advance);
+                        sendOffset += advance;
+                        pending -= advance;
+                        break;
+                    }
+
+                    case EAction::REWIND:
+                        Ctest << " REWIND";
+                        stream.Rewind();
+                        pending += sendOffset;
+                        sendOffset = 0;
+                        --numRewindsRemain;
+                        break;
+
+                    case EAction::DROP: {
+                        const size_t drop = rng() % Min<size_t>(65536, sendOffset + 1);
+                        Ctest << " DROP drop# " << drop;
+                        stream.DropFront(drop);
+                        base += drop;
+                        sendOffset -= drop;
+                        break;
+                    }
+
+                    case EAction::BOOKMARK:
+                        Ctest << " BOOKMARK nextDataLen# " << nextDataLen;
+                        auto bookmark = stream.Bookmark(nextDataLen);
+                        stream.WriteBookmark(std::move(bookmark), {nextData, nextDataLen});
+                        pending += nextDataLen;
+                        break;
+                }
+
+                Ctest << Endl;
+            }
+        }
+    }
+}