Add checksum support

Author: robdrynkin

ydb/library/actors/interconnect/interconnect_channel.cpp

@@ -104,14 +104,14 @@ namespace NActors {
                     } else if (Params.UseExternalDataChannel && !SerializationInfo->Sections.empty()) {
                         State = EState::SECTIONS;
                         SectionIndex = 0;
+                        XXH3_64bits_reset(&RdmaChecksumState);
 
                         bool sendViaRdma = false;
                         // Check if any section can be send via rdma
                         for (const auto& section : SerializationInfo->Sections) {
                             sendViaRdma |= section.IsRdmaCapable;
                         }
                         if (sendViaRdma && Params.UseRdma && RdmaMemPool) {
-                            NActorsInterconnect::TRdmaCreds rdmaCreds;
                             if (SerializeEventRdma(event)) {
                                 Chunker.DiscardEvent();
                             }
@@ -337,11 +337,6 @@ namespace NActors {
     std::optional<bool> TEventOutputChannel::FeedRdmaPayload(TTcpPacketOutTask& task, TEventHolder& event, ssize_t rdmaDeviceIndex, bool checksumming) {
         Y_ABORT_UNLESS(rdmaDeviceIndex >= 0);
 
-        XXH3_state_t state;
-        if (checksumming) {
-            XXH3_64bits_reset(&state);
-        }
-
         Y_ABORT_UNLESS(event.Buffer);
         if (RdmaCredsBuffer.CredsSize() == 0) {
             auto prevIter = Iter;
@@ -357,7 +352,7 @@ namespace NActors {
                     return false;
                 }
                 if (checksumming) {
-                    XXH3_64bits_update(&state, buf.GetData(), buf.GetSize());
+                    XXH3_64bits_update(&RdmaChecksumState, buf.GetData(), buf.GetSize());
                 }
                 auto cred = RdmaCredsBuffer.AddCreds();
                 cred->SetAddress(reinterpret_cast<ui64>(memReg.GetAddr()));
@@ -398,7 +393,7 @@ namespace NActors {
 
         Y_ABORT_UNLESS(RdmaCredsBuffer.SerializePartialToArray(ptr, credsSerializedSize));
         ptr += credsSerializedSize;
-        WriteUnaligned<ui32>(ptr, checksumming ? XXH3_64bits_digest(&state) : 0);
+        WriteUnaligned<ui32>(ptr, checksumming ? XXH3_64bits_digest(&RdmaChecksumState) : 0);
         OutputQueueSize -= payloadSz;
 
         task.Write<false>(buffer, partSize);

ydb/library/actors/interconnect/interconnect_channel.h

@@ -152,6 +152,7 @@ namespace NActors {
         size_t SectionIndex = 0;
         std::vector<char> XdcData;
         std::shared_ptr<NInterconnect::NRdma::IMemPool> RdmaMemPool;
+        XXH3_state_t RdmaChecksumState;
 
         template<bool External>
         bool SerializeEvent(TTcpPacketOutTask& task, TEventHolder& event, size_t *bytesSerialized);

ydb/library/actors/interconnect/interconnect_tcp_input_session.cpp

@@ -898,21 +898,25 @@ namespace NActors {
                 for (const auto&& [data, size] : payload) {
                     checksum = Crc32cExtendMSanCompatible(checksum, data, size);
                 }
-            } else if (pendingEvent.RdmaCheckSum) {
+                if (checksum != descr.Checksum) {
+                    LOG_CRIT_IC_SESSION("ICIS05", "event checksum error Type# 0x%08" PRIx32, descr.Type);
+                    throw TExReestablishConnection{TDisconnectReason::ChecksumError()};
+                }
+            }
+            if (pendingEvent.RdmaCheckSum) {
                 XXH3_state_t state;
                 XXH3_64bits_reset(&state);
-                for (const auto&& [data, size] : payload) {
-                    XXH3_64bits_update(&state, data, size);
+                for (auto iter = payload.Begin(); iter.Valid(); ++iter) {
+                    auto memRegion = NInterconnect::NRdma::TryExtractFromRcBuf(iter.GetChunk());
+                    if (!memRegion.Empty()) {
+                        XXH3_64bits_update(&state, memRegion.GetAddr(), memRegion.GetSize());
+                    }
                 }
                 checksum = XXH3_64bits_digest(&state);
-            }
-            ui32 expectedChecksum = descr.Checksum ?: pendingEvent.RdmaCheckSum;
-            if (expectedChecksum) {
-                Y_UNUSED(checksum, expectedChecksum);
-                // if (checksum != expectedChecksum) {
-                //     LOG_CRIT_IC_SESSION("ICIS05", "event checksum error Type# 0x%08" PRIx32, descr.Type);
-                //     throw TExReestablishConnection{TDisconnectReason::ChecksumError()};
-                // }
+                if (checksum != pendingEvent.RdmaCheckSum) {
+                    LOG_CRIT_IC_SESSION("ICIS05", "event rdma checksum error Type# 0x%08" PRIx32, descr.Type);
+                    throw TExReestablishConnection{TDisconnectReason::ChecksumError()};
+                }
             }
             pendingEvent.SerializationInfo.IsExtendedFormat = descr.Flags & IEventHandle::FlagExtendedFormat;