various changes in rdma logging and error handling (#4543)

- resolve #4533
- log completions with unexpected opcodes to debug #3375
- do not throttle unexpected completion messages
- rename unexpected completions -> completion errors
- replace most of the rdma crit events with error messages, because they
  aren't really all that critical

Author: tpashkin

cloud/blockstore/libs/rdma/impl/client.cpp

@@ -15,7 +15,6 @@
 #include <cloud/blockstore/libs/rdma/iface/protobuf.h>
 #include <cloud/blockstore/libs/rdma/iface/protocol.h>
 
-#include <cloud/blockstore/libs/diagnostics/critical_events.h>
 #include <cloud/blockstore/libs/service/context.h>
 
 #include <cloud/storage/core/libs/common/backoff_delay_provider.h>
@@ -54,7 +53,6 @@ constexpr TDuration POLL_TIMEOUT = TDuration::Seconds(1);
 constexpr TDuration RESOLVE_TIMEOUT = TDuration::Seconds(10);
 constexpr TDuration MIN_CONNECT_TIMEOUT = TDuration::Seconds(1);
 constexpr TDuration FLUSH_TIMEOUT = TDuration::Seconds(10);
-constexpr TDuration LOG_THROTTLER_PERIOD = TDuration::Seconds(60);
 constexpr TDuration MIN_RECONNECT_DELAY = TDuration::MilliSeconds(10);
 constexpr TDuration INSTANT_RECONNECT_DELAY = TDuration::MicroSeconds(1);
 
@@ -267,7 +265,7 @@ struct TEndpointCounters
     TDynamicCounters::TCounterPtr SendErrors;
     TDynamicCounters::TCounterPtr RecvErrors;
 
-    TDynamicCounters::TCounterPtr UnexpectedCompletions;
+    TDynamicCounters::TCounterPtr CompletionErrors;
 
     void Register(TDynamicCounters& counters)
     {
@@ -283,7 +281,7 @@ struct TEndpointCounters
         SendErrors = counters.GetCounter("SendErrors");
         RecvErrors = counters.GetCounter("RecvErrors");
 
-        UnexpectedCompletions = counters.GetCounter("UnexpectedCompletions");
+        CompletionErrors = counters.GetCounter("CompletionErrors");
     }
 
     void RequestEnqueued()
@@ -318,6 +316,11 @@ struct TEndpointCounters
         SendErrors->Inc();
     }
 
+    void SendError()
+    {
+        SendErrors->Inc();
+    }
+
     void RecvResponseCompleted()
     {
         ActiveRecv->Dec();
@@ -332,6 +335,11 @@ struct TEndpointCounters
         RecvErrors->Inc();
     }
 
+    void RecvError()
+    {
+        RecvErrors->Inc();
+    }
+
     void RequestAborted()
     {
         ActiveRequests->Dec();
@@ -343,9 +351,9 @@ struct TEndpointCounters
         UnknownRequests->Inc();
     }
 
-    void UnexpectedCompletion()
+    void CompletionError()
     {
-        UnexpectedCompletions->Inc();
+        CompletionErrors->Inc();
     }
 };
 
@@ -453,10 +461,6 @@ class TClientEndpoint final
     TLog Log;
     TReconnect Reconnect;
 
-    struct {
-        TLogThrottler Unexpected = TLogThrottler(LOG_THROTTLER_PERIOD);
-    } LogThrottler;
-
     // config might be adjusted during initial handshake
     TClientConfigPtr OriginalConfig;
     TClientConfig Config;
@@ -567,14 +571,13 @@ class TClientEndpoint final
 private:
     // called from CQ thread
     void HandleQueuedRequests();
-    bool IsWorkRequestValid(const TWorkRequestId& id) const;
-    void HandleFlush(const TWorkRequestId& id) noexcept;
     void SendRequest(TRequestPtr req, TSendWr* send);
     void SendRequestCompleted(TSendWr* send, ibv_wc_status status) noexcept;
     void RecvResponse(TRecvWr* recv);
     void RecvResponseCompleted(TRecvWr* recv, ibv_wc_status status);
     void AbortRequest(TRequestPtr req, ui32 err, const TString& msg) noexcept;
     void FreeRequest(TRequest* creq) noexcept;
+    int ValidateCompletion(ibv_wc* wc) noexcept;
     ui64 GetNewReqId() noexcept;
 };
 
@@ -1028,48 +1031,67 @@ bool TClientEndpoint::HandleCompletionEvents()
     return false;
 }
 
-bool TClientEndpoint::IsWorkRequestValid(const TWorkRequestId& id) const
+int TClientEndpoint::ValidateCompletion(ibv_wc* wc) noexcept
 {
-    if (id.Magic == SendMagic && id.Index < SendWrs.size()) {
-        return true;
-    }
-    if (id.Magic == RecvMagic && id.Index < RecvWrs.size()) {
-        return true;
-    }
-    return false;
-}
+    auto id = TWorkRequestId(wc->wr_id);
 
-void TClientEndpoint::HandleFlush(const TWorkRequestId& id) noexcept
-{
-    // flush WRs have opcode=0
     if (id.Magic == SendMagic && id.Index < SendWrs.size()) {
-        SendQueue.Push(&SendWrs[id.Index]);
-        return;
+        if (wc->status == IBV_WC_WR_FLUSH_ERR) {
+            SendQueue.Push(&SendWrs[id.Index]);
+            return -1;
+        }
+
+        if (wc->opcode != IBV_WC_SEND) {
+            RDMA_ERROR(
+                "completion error " << NVerbs::PrintCompletion(wc)
+                                    << ": unexpected opcode");
+
+            Counters->CompletionError();
+            SendQueue.Push(&SendWrs[id.Index]);
+            return -1;
+        }
+
+        return 0;
     }
+
     if (id.Magic == RecvMagic && id.Index < RecvWrs.size()) {
-        RecvQueue.Push(&RecvWrs[id.Index]);
-        return;
+        if (wc->status == IBV_WC_WR_FLUSH_ERR) {
+            RecvQueue.Push(&RecvWrs[id.Index]);
+            return -1;
+        }
+
+        if (wc->opcode != IBV_WC_RECV) {
+            RDMA_ERROR(
+                "completion error " << NVerbs::PrintCompletion(wc)
+                                    << ": unexpected opcode");
+
+            Counters->CompletionError();
+            RecvQueue.Push(&RecvWrs[id.Index]);
+            return -1;
+        }
+
+        return 0;
     }
+
+    RDMA_ERROR(
+        "completion error " << NVerbs::PrintCompletion(wc)
+                            << ": unexpected wr_id");
+
+    Counters->CompletionError();
+    return -1;
 }
 
 // implements NVerbs::ICompletionHandler
 void TClientEndpoint::HandleCompletionEvent(ibv_wc* wc)
 {
     auto id = TWorkRequestId(wc->wr_id);
 
-    RDMA_TRACE(NVerbs::GetOpcodeName(wc->opcode) << " " << id
-        << " completed with " << NVerbs::GetStatusString(wc->status));
-
-    if (!IsWorkRequestValid(id)) {
-        RDMA_ERROR(LogThrottler.Unexpected, Log,
-            "unexpected completion " << NVerbs::PrintCompletion(wc));
+    RDMA_TRACE(
+        NVerbs::GetOpcodeName(wc->opcode)
+        << " " << id << " completed with "
+        << NVerbs::GetStatusString(wc->status));
 
-        Counters->UnexpectedCompletion();
-        return;
-    }
-
-    if (wc->status == IBV_WC_WR_FLUSH_ERR) {
-        HandleFlush(id);
+    if (ValidateCompletion(wc)) {
         return;
     }
 
@@ -1083,10 +1105,7 @@ void TClientEndpoint::HandleCompletionEvent(ibv_wc* wc)
             break;
 
         default:
-            RDMA_ERROR(LogThrottler.Unexpected, Log,
-                "unexpected completion " << NVerbs::PrintCompletion(wc));
-
-            Counters->UnexpectedCompletion();
+            break;
     }
 }
 
@@ -1108,17 +1127,14 @@ void TClientEndpoint::SendRequest(TRequestPtr req, TSendWr* send)
     try {
         Verbs->PostSend(Connection->qp, &send->wr);
     } catch (const TServiceError& e) {
-        SendQueue.Push(send);
-
-        ReportRdmaError(
-            TStringBuilder()
-            << "SEND " << TWorkRequestId(send->wr.wr_id) << ": " << e.what());
-
-        Disconnect();
+        RDMA_ERROR(
+            "SEND " << TWorkRequestId(send->wr.wr_id) << ": " << e.what());
 
+        SendQueue.Push(send);
+        Counters->SendError();
         Counters->RequestEnqueued();
         QueuedRequests.Enqueue(std::move(req));
-
+        Disconnect();
         return;
     }
 
@@ -1143,13 +1159,11 @@ void TClientEndpoint::SendRequestCompleted(
     };
 
     if (status != IBV_WC_SUCCESS) {
-        Counters->SendRequestError();
-
-        ReportRdmaError(
-            TStringBuilder()
-            << "SEND request completed " << TWorkRequestId(send->wr.wr_id)
-            << " failed: " << NVerbs::GetStatusString(status));
+        RDMA_ERROR(
+            "SEND " << TWorkRequestId(send->wr.wr_id) << ": "
+                    << NVerbs::GetStatusString(status));
 
+        Counters->SendRequestError();
         Disconnect();
         return;
     }
@@ -1163,21 +1177,26 @@ void TClientEndpoint::SendRequestCompleted(
             req->CallContext->RequestId);
 
     } else if (ActiveRequests.TimedOut(reqId)) {
-        RDMA_INFO("SEND " << TWorkRequestId(send->wr.wr_id)
-            << ": request has timed out before receiving send wc");
+        RDMA_INFO(
+            "SEND "
+            << TWorkRequestId(send->wr.wr_id)
+            << ": request has timed out before receiving send completion");
 
     } else if (ActiveRequests.Completed(reqId)) {
-        RDMA_INFO("SEND " << TWorkRequestId(send->wr.wr_id)
-            << ": request has been completed before receiving send wc");
+        RDMA_INFO(
+            "SEND "
+            << TWorkRequestId(send->wr.wr_id)
+            << ": request has been completed before receiving send completion");
 
     } else if (ActiveRequests.Cancelled(reqId)) {
         RDMA_INFO(
-            "SEND " << TWorkRequestId(send->wr.wr_id)
-                    << ": request was cancelled before receiving send wc");
+            "SEND "
+            << TWorkRequestId(send->wr.wr_id)
+            << ": request was cancelled before receiving send completion");
 
     } else {
-        RDMA_ERROR("SEND " << TWorkRequestId(send->wr.wr_id)
-            << ": request not found")
+        RDMA_ERROR(
+            "SEND " << TWorkRequestId(send->wr.wr_id) << ": request not found")
         Counters->UnknownRequest();
     }
 }
@@ -1192,13 +1211,11 @@ void TClientEndpoint::RecvResponse(TRecvWr* recv)
     try {
         Verbs->PostRecv(Connection->qp, &recv->wr);
     } catch (const TServiceError& e) {
-        RecvQueue.Push(recv);
-
-        ReportRdmaError(
-            TStringBuilder()
-            << "RECV " << TWorkRequestId(recv->wr.wr_id)
-            << " failed to post receive request: " << e.what());
+        RDMA_ERROR(
+            "RECV " << TWorkRequestId(recv->wr.wr_id) << ": " << e.what());
 
+        Counters->RecvError();
+        RecvQueue.Push(recv);
         Disconnect();
         return;
     }
@@ -1211,24 +1228,23 @@ void TClientEndpoint::RecvResponseCompleted(
     ibv_wc_status wc_status)
 {
     if (wc_status != IBV_WC_SUCCESS) {
+        RDMA_ERROR(
+            "RECV " << TWorkRequestId(recv->wr.wr_id) << ": "
+                    << NVerbs::GetStatusString(wc_status));
+
         Counters->RecvResponseError();
         RecvQueue.Push(recv);
-
-        ReportRdmaError(
-            TStringBuilder()
-            << "RECV " << TWorkRequestId(recv->wr.wr_id)
-            << " failed: " << NVerbs::GetStatusString(wc_status));
-
         Disconnect();
         return;
     }
 
     auto* msg = recv->Message<TResponseMessage>();
     int version = ParseMessageHeader(msg);
     if (version != RDMA_PROTO_VERSION) {
-        RDMA_ERROR("RECV " << TWorkRequestId(recv->wr.wr_id)
-            << ": incompatible protocol version "
-            << version << ", expected " << int(RDMA_PROTO_VERSION));
+        RDMA_ERROR(
+            "RECV " << TWorkRequestId(recv->wr.wr_id)
+                    << ": incompatible protocol version " << version
+                    << ", expected " << int(RDMA_PROTO_VERSION));
 
         Counters->RecvResponseError();
         RecvResponse(recv);
@@ -1243,8 +1259,8 @@ void TClientEndpoint::RecvResponseCompleted(
 
     auto req = ActiveRequests.Pop(reqId);
     if (!req) {
-        RDMA_ERROR("RECV " << TWorkRequestId(recv->wr.wr_id)
-            << ": request not found");
+        RDMA_ERROR(
+            "RECV " << TWorkRequestId(recv->wr.wr_id) << ": request not found");
 
         Counters->UnknownRequest();
         return;
@@ -2285,7 +2301,7 @@ void TClient::DumpHtml(IOutputStream& out) const
                     TABLEH() { out << "ActiveRecv"; }
                     TABLEH() { out << "SendErrors"; }
                     TABLEH() { out << "RecvErrors"; }
-                    TABLEH() { out << "UnexpectedCompletions"; }
+                    TABLEH() { out << "CompletionErrors"; }
                 }
                 TABLER() {
                     TABLED() { out << Counters->QueuedRequests->Val(); }
@@ -2297,7 +2313,7 @@ void TClient::DumpHtml(IOutputStream& out) const
                     TABLED() { out << Counters->ActiveRecv->Val(); }
                     TABLED() { out << Counters->SendErrors->Val(); }
                     TABLED() { out << Counters->RecvErrors->Val(); }
-                    TABLED() { out << Counters->UnexpectedCompletions->Val(); }
+                    TABLED() { out << Counters->CompletionErrors->Val(); }
                 }
             }
         }

cloud/blockstore/libs/rdma/impl/client_ut.cpp

@@ -656,10 +656,10 @@ Y_UNIT_TEST_SUITE(TRdmaClientTest)
             ->GetSubgroup("counters", "blockstore")
             ->GetSubgroup("component", "rdma_client");
 
-        auto unexpected = counters->GetCounter("UnexpectedCompletions");
+        auto completionErrors = counters->GetCounter("CompletionErrors");
         auto active = counters->GetCounter("ActiveRecv");
         auto errors = counters->GetCounter("RecvErrors");
-        auto unexpectedOld = unexpected->Val();
+        auto completionErrorsOld = completionErrors->Val();
         auto errorsOld = errors->Val();
 
         ibv_recv_wr* wr = context->RecvEvents.back();
@@ -716,8 +716,8 @@ Y_UNIT_TEST_SUITE(TRdmaClientTest)
             }
         };
 
-        wait(unexpected, unexpectedOld + 2);
-        wait(errors, errorsOld + 2);
+        wait(completionErrors, completionErrorsOld + 2);
+        wait(errors, errorsOld + 3);
         wait(active, 8);
     }
 };