Add test

Author: robdrynkin

ydb/library/actors/core/event_pb.cpp

@@ -470,14 +470,14 @@ namespace NActors {
                     for (const TRope& rope : payload) {
                         headerLen += SerializeNumber(rope.size(), temp);
                     }
-                    info.Sections.push_back(TEventSectionInfo{0, headerLen, 0, 0, true, true});
+                    info.Sections.push_back(TEventSectionInfo{0, headerLen, 0, 0, true, false});
                     for (const TRope& rope : payload) {
                         info.Sections.push_back(TEventSectionInfo{0, rope.size(), 0, 0, false, IsRdma(rope)});
                     }
                 }
 
                 const size_t byteSize = Max<ssize_t>(0, recordSize) + preserializedSize;
-                info.Sections.push_back(TEventSectionInfo{0, byteSize, 0, 0, true, true}); // protobuf itself
+                info.Sections.push_back(TEventSectionInfo{0, byteSize, 0, 0, true, false}); // protobuf itself
 
 #ifndef NDEBUG
                 size_t total = 0;

ydb/library/actors/interconnect/interconnect_channel.cpp

@@ -105,17 +105,12 @@ namespace NActors {
                         State = EState::SECTIONS;
                         SectionIndex = 0;
 
-                        size_t totalSize = 0;
-                        // It is possible to have event without payload. Such events has only one section.
-                        // We do not send such events via rdma.
-                        bool sendViaRdma = Params.UseRdma && RdmaMemPool && SerializationInfo->Sections.size() > 2;
-                        // Check each section can be send via rdma
+                        bool sendViaRdma = false;
+                        // Check if any section can be send via rdma
                         for (const auto& section : SerializationInfo->Sections) {
-                            sendViaRdma &= section.IsRdmaCapable;
-                            totalSize += section.Size;
+                            sendViaRdma |= section.IsRdmaCapable;
                         }
-                        if (sendViaRdma) {
-                            Y_ABORT_UNLESS(totalSize, "got empty sz, sections: %d type: %d ", SerializationInfo->Sections.size(),  event.Event->Type());
+                        if (sendViaRdma && Params.UseRdma && RdmaMemPool) {
                             NActorsInterconnect::TRdmaCreds rdmaCreds;
                             if (SerializeEventRdma(event)) {
                                 Chunker.DiscardEvent();

ydb/library/actors/interconnect/ut_rdma/rdma_xdc_ut.cpp

@@ -317,6 +317,47 @@ TEST_F(XdcRdmaTest, SendRdma) {
     UNIT_ASSERT(recieverPtr->WhaitForRecieve(1, 20));
 }
 
+TEST_F(XdcRdmaTest, SendRdmaWithShuffledPayload) {
+    TTestICCluster cluster(2);
+    auto memPool = NInterconnect::NRdma::CreateDummyMemPool();
+    auto ev = new TEvTestSerialization();
+    ev->Record.SetBlobID(123);
+    ev->Record.SetBuffer("hello world");
+    for (ui32 i = 0; i < 10; ++i) {
+        if (i % 2 == 0) {
+            TRope tmp(TString(5000, 'X'));
+            ev->AddPayload(std::move(tmp));
+        } else {
+            auto buf = memPool->AllocRcBuf(5000, 0).value();
+            std::fill(buf.GetDataMut(), buf.GetDataMut() + 5000, 'Y');
+            ev->AddPayload(TRope(std::move(buf)));
+        }
+    }
+
+    auto recieverPtr = new TReceiveActor([](TEvTestSerialization::TPtr ev) {
+        Cerr << "Blob ID: " << ev->Get()->Record.GetBlobID() << Endl;
+        UNIT_ASSERT_VALUES_EQUAL(ev->Get()->Record.GetBlobID(), 123u);
+        UNIT_ASSERT_VALUES_EQUAL(ev->Get()->Record.GetBuffer(), "hello world");
+        UNIT_ASSERT_VALUES_EQUAL(ev->Get()->GetPayload().size(), 10u);
+        for (ui32 i = 0; i < 10; ++i) {
+            if (i % 2 == 0) {
+                UNIT_ASSERT_VALUES_EQUAL(ev->Get()->GetPayload()[i].GetSize(), 5000u);
+                UNIT_ASSERT_VALUES_EQUAL(ev->Get()->GetPayload()[i].ConvertToString(), TString(5000, 'X'));
+            } else {
+                UNIT_ASSERT_VALUES_EQUAL(ev->Get()->GetPayload()[i].GetSize(), 5000u);
+                UNIT_ASSERT_VALUES_EQUAL(ev->Get()->GetPayload()[i].ConvertToString(), TString(5000, 'Y'));
+            }
+        }
+    });
+    const TActorId receiver = cluster.RegisterActor(recieverPtr, 1);
+
+    Sleep(TDuration::MilliSeconds(1000));
+
+    auto senderPtr = new TSendActor(receiver, ev);
+    cluster.RegisterActor(senderPtr, 2);
+    UNIT_ASSERT(recieverPtr->WhaitForRecieve(1, 20));
+}
+
 TEST_F(XdcRdmaTest, SendRdmaWithRegionOffset) {
     TTestICCluster cluster(2);
     auto memPool = NInterconnect::NRdma::CreateDummyMemPool();