diff --git a/labview source/Client Server Support New/Server Template/RPC Service/ServiceName/RPC Methods/Set package_service_method Response.vi b/labview source/Client Server Support New/Server Template/RPC Service/ServiceName/RPC Methods/Set package_service_method Response.vi index 7e2fe143..047812c3 100644 Binary files a/labview source/Client Server Support New/Server Template/RPC Service/ServiceName/RPC Methods/Set package_service_method Response.vi and b/labview source/Client Server Support New/Server Template/RPC Service/ServiceName/RPC Methods/Set package_service_method Response.vi differ diff --git a/labview source/gRPC lv Support/Client API/Client Complete Client Streaming Call.vim b/labview source/gRPC lv Support/Client API/Client Complete Client Streaming Call.vim index 595a2a7c..5058e806 100644 Binary files a/labview source/gRPC lv Support/Client API/Client Complete Client Streaming Call.vim and b/labview source/gRPC lv Support/Client API/Client Complete Client Streaming Call.vim differ diff --git a/labview source/gRPC lv Support/Client API/Client Unary Call.vim b/labview source/gRPC lv Support/Client API/Client Unary Call.vim index 21bb9d16..5675e962 100644 Binary files a/labview source/gRPC lv Support/Client API/Client Unary Call.vim and b/labview source/gRPC lv Support/Client API/Client Unary Call.vim differ diff --git a/src/event_data.cc b/src/event_data.cc index c0dd77d0..09285027 100644 --- a/src/event_data.cc +++ b/src/event_data.cc @@ -16,9 +16,12 @@ namespace grpc_labview { auto callData = std::shared_ptr(new CallData(server, service, cq)); - auto finishedTag = new CallFinishedTag(callData); - callData->_callFinishedTag = finishedTag; - callData->_ctx.AsyncNotifyWhenDone(finishedTag); + // Do not call AsyncNotifyWhenDone. Holding CallData on that CQ tag + // leaked one object per RPC because the tag is not reliably delivered + // for AsyncGenericService. Instead, rely on the existing CompletionQueueTag + // to keep CallData alive only until stream.Finish() is delivered on the CQ + // (after LabVIEW has unregistered the CallData pointer via CloseServerEvent). + // Read/Write. // Start the state machine which waits for a new call to arrive. auto tag = new CompletionQueueTag(callData); @@ -68,14 +71,16 @@ namespace grpc_labview //--------------------------------------------------------------------- bool CallData::IsCancelled() { - return _ctx.IsCancelled(); + // Unsafe to call _ctx.IsCancelled() without a completed + // AsyncNotifyWhenDone tag. See CallData::Create. + return false; } //--------------------------------------------------------------------- //--------------------------------------------------------------------- bool CallData::IsActive() { - return _status != CallStatus::Finished && _status != CallStatus::Finishing && !IsCancelled(); + return _status != CallStatus::Finished && _status != CallStatus::Finishing; } //--------------------------------------------------------------------- @@ -134,18 +139,8 @@ namespace grpc_labview if (!ok && _status != CallStatus::Finished) { - if (_status == CallStatus::WaitingForConnection) - { - // Ugh. When using the grpc async APIs, you are required to call AsyncNotifyWhenDone if you want to call IsCancelled - // on the ServerContext. However, the tag registered with AsyncNotifyWhenDone is only notified if a RPC call actually - // starts, and you must call AsyncNotifyWhenDone before the call starts or tag will not be notified either. Generally, - // it is acceptable to just leak this one tag on server shutdown. However, because we maintain a shared pointer to the - // server, we will end up leaking everything if we don't clean up this tag. As a work around, we delete the tag here. - // This leaves a dangling tag pointer in the completion queue, but it never does anything with the tag. It only delivers - // the tag from the Next call which we know will never be triggered so this should be safe. - delete _callFinishedTag; - _callFinishedTag = nullptr; - } + // RequestCall ok=false means the call never started (typically + // CQ shutdown). Finish completing with ok=false is handled below. _status = CallStatus::Finishing; } @@ -211,41 +206,6 @@ namespace grpc_labview _stream.Finish(_callStatus, new CompletionQueueTag(shared_from_this())); } - //--------------------------------------------------------------------- - //--------------------------------------------------------------------- - void CallData::FinishFromCompletionQueue() - { - std::lock_guard lock(_stateMutex); - - _callFinishedTag = nullptr; - - // The call was completed normally from LV code. - if (_status == CallStatus::Finishing || _status == CallStatus::Finished) - { - return; - } - - // If FinishFromCompletionQueue is called and we are not already finishing because the user completed - // the call from LV, then it means either the server is shutting down or the call was cancelled. In either - // case there is no point in calling Finish on the stream so just mark the call as finished. - _status = CallStatus::Finished; - } - - //--------------------------------------------------------------------- - //--------------------------------------------------------------------- - CallFinishedTag::CallFinishedTag(std::shared_ptr callData) - { - _callData = callData; - } - - //--------------------------------------------------------------------- - //--------------------------------------------------------------------- - void CallFinishedTag::Proceed(bool ok) - { - _callData->FinishFromCompletionQueue(); - delete this; - } - //--------------------------------------------------------------------- //--------------------------------------------------------------------- ReadNextTag::ReadNextTag(std::shared_ptr callData) : @@ -328,4 +288,4 @@ namespace grpc_labview { serverStartStatus = 0; } -} \ No newline at end of file +} diff --git a/src/grpc_client.cc b/src/grpc_client.cc index 5cc700b2..7c9923f2 100644 --- a/src/grpc_client.cc +++ b/src/grpc_client.cc @@ -15,6 +15,28 @@ #include #include +namespace +{ + // RAII cleanup that runs on scope exit (normal OR exception) unless dismissed. + // The client completers dismiss it on the normal path, so it only fires when an + // exception (e.g. from response copying) would otherwise skip removing the call + // from ActiveClientCalls / the pointer manager -- which would leave a dangling + // raw pointer in ActiveClientCalls or leak the ClientCall. Never throws. + template + class ScopeGuard + { + public: + explicit ScopeGuard(F fn) : _fn(std::move(fn)) {} + ~ScopeGuard() { if (_active) { try { _fn(); } catch (...) {} } } + void dismiss() { _active = false; } + ScopeGuard(const ScopeGuard&) = delete; + ScopeGuard& operator=(const ScopeGuard&) = delete; + private: + F _fn; + bool _active = true; + }; +} + namespace grpc_labview { //--------------------------------------------------------------------- @@ -444,6 +466,16 @@ LIBRARY_EXPORT int32_t CompleteClientUnaryCall2( return -1; } + // Ensure the call is reaped even if response processing below throws. + ScopeGuard reap([&]{ + { + std::unique_lock lock(clientCall->_client->clientLock); + auto it = clientCall->_client->ActiveClientCalls.find(clientCall.get()); + if (it != clientCall->_client->ActiveClientCalls.end()) + clientCall->_client->ActiveClientCalls.erase(it); + } + grpc_labview::gPointerManager.UnregisterPointer(callId); + }); grpc_labview::gPointerManager.UnregisterPointer(callId); @@ -458,6 +490,14 @@ LIBRARY_EXPORT int32_t CompleteClientUnaryCall2( } else { + // On a failed/timed-out call, cancel the context so gRPC tears down the + // HTTP/2 stream (RST_STREAM) instead of leaving it to drain. Otherwise the + // event engine keeps reading and buffering the late-arriving response for a + // call nobody consumes -- a per-timeout leak in the gRPC-core receive path + // (MaybeMakeReadSlices / ProcessDataAfterMetadata). Cancel is a no-op if the + // stream already closed. + clientCall->Cancel(); + result = -(1000 + clientCall->_status.error_code()); if (errorMessage != nullptr) { @@ -474,6 +514,7 @@ LIBRARY_EXPORT int32_t CompleteClientUnaryCall2( clientCall->_client->ActiveClientCalls.erase(call); } lock.unlock(); + reap.dismiss(); return result; } catch (const std::exception& e) { grpc_labview::SetErrorMessage(errorMessage, e.what()); @@ -787,6 +828,18 @@ LIBRARY_EXPORT int32_t FinishClientCompleteClientStreamingCall( { return -1; } + + // Ensure the call is reaped even if response processing below throws. + ScopeGuard reap([&]{ + { + std::unique_lock lock(call->_client->clientLock); + auto it = call->_client->ActiveClientCalls.find(call.get()); + if (it != call->_client->ActiveClientCalls.end()) + call->_client->ActiveClientCalls.erase(it); + } + grpc_labview::gPointerManager.UnregisterPointer(callId); + }); + int32_t result = 0; if (call->_status.ok()) { @@ -795,6 +848,11 @@ LIBRARY_EXPORT int32_t FinishClientCompleteClientStreamingCall( } else { + // Cancel on failure/timeout so gRPC tears down the stream and frees the + // event-engine receive buffers instead of leaving them to drain (same + // mechanism as the unary path). No-op if the stream already closed. + call->Cancel(); + result = -(1000 + call->_status.error_code()); if (errorMessage != nullptr) { @@ -812,6 +870,7 @@ LIBRARY_EXPORT int32_t FinishClientCompleteClientStreamingCall( } lock.unlock(); grpc_labview::gPointerManager.UnregisterPointer(callId); + reap.dismiss(); return result; } catch (const std::exception& e) { grpc_labview::SetErrorMessage(errorMessage, e.what()); @@ -867,6 +926,17 @@ LIBRARY_EXPORT int32_t ClientCompleteStreamingCall( return -1; } + // Ensure the call is reaped even if Finish()/response processing below throws. + ScopeGuard reap([&]{ + { + std::unique_lock lock(call->_client->clientLock); + auto it = call->_client->ActiveClientCalls.find(call.get()); + if (it != call->_client->ActiveClientCalls.end()) + call->_client->ActiveClientCalls.erase(it); + } + grpc_labview::gPointerManager.UnregisterPointer(callId); + }); + // We've already got a shared_ptr for this token, so calling DestroyToken now // will just prevent any other API calls from grabbing the pointer. grpc_labview::gPointerManager.UnregisterPointer(callId); @@ -875,6 +945,11 @@ LIBRARY_EXPORT int32_t ClientCompleteStreamingCall( int32_t result = 0; if (!call->_status.ok()) { + // Cancel on failure/timeout so gRPC tears down the stream and frees the + // event-engine receive buffers instead of leaving them to drain (same + // mechanism as the unary path). No-op if the stream already closed. + call->Cancel(); + result = -(1000 + call->_status.error_code()); if (errorMessage != nullptr) { @@ -891,6 +966,7 @@ LIBRARY_EXPORT int32_t ClientCompleteStreamingCall( call->_client->ActiveClientCalls.erase(client_call); } lock.unlock(); + reap.dismiss(); return result; } catch (const std::exception&) { return grpc_labview::TranslateException(); diff --git a/src/grpc_server.h b/src/grpc_server.h index 7ba79d8f..579179cd 100644 --- a/src/grpc_server.h +++ b/src/grpc_server.h @@ -46,7 +46,6 @@ namespace grpc_labview class LabVIEWgRPCServer; class LVMessage; class CallData; - class CallFinishedTag; class MessageElementMetadata; struct MessageMetadata; @@ -123,7 +122,6 @@ namespace grpc_labview void Proceed(bool ok) override; bool Write(int8_t* cluster); void FinishFromLabVIEW(); - void FinishFromCompletionQueue(); bool IsCancelled(); bool IsActive(); bool ReadNext(int8_t* cluster); @@ -141,7 +139,6 @@ namespace grpc_labview grpc::GenericServerAsyncReaderWriter _stream; grpc::ByteBuffer _rb; grpc::Status _callStatus; - CallFinishedTag* _callFinishedTag; std::shared_ptr _request; std::shared_ptr _response; @@ -159,18 +156,6 @@ namespace grpc_labview CallStatus _status; }; - //--------------------------------------------------------------------- - //--------------------------------------------------------------------- - class CallFinishedTag : public CallDataBase - { - public: - CallFinishedTag(std::shared_ptr callData); - void Proceed(bool ok) override; - - private: - std::shared_ptr _callData; - }; - //--------------------------------------------------------------------- // Completion queue tag that keeps CallData alive //---------------------------------------------------------------------