diff --git a/GrpcInterface/Python/rips/tests/test_error_reporting.py b/GrpcInterface/Python/rips/tests/test_error_reporting.py index 38eec50b08..629cb760a6 100644 --- a/GrpcInterface/Python/rips/tests/test_error_reporting.py +++ b/GrpcInterface/Python/rips/tests/test_error_reporting.py @@ -1,5 +1,6 @@ import sys import os +import threading import time import grpc @@ -172,3 +173,51 @@ def test_heartbeat_reports_exited_process_with_diagnostic(): message = str(excinfo.value) assert "exited unexpectedly" in message assert "-11" in message + + +def test_server_answers_pings_while_main_thread_is_busy(rips_instance, initialize_test): + """ResInsight processes API requests on its main thread, so a long request + used to leave liveness pings unanswered for as long as it ran, making a busy + server look like a dead one to the client heartbeat (issue #14683). + + Pings must be answered promptly while a blocking request is in flight.""" + from Definitions_pb2 import Empty + + stop = threading.Event() + latencies = [] + errors = [] + + def ping_loop(): + while not stop.is_set(): + started = time.monotonic() + try: + rips_instance.app.GetVersion(Empty(), timeout=30.0) + latencies.append(time.monotonic() - started) + except grpc.RpcError as exc: # pragma: no cover - failure diagnostics + errors.append(exc) + time.sleep(0.02) + + thread = threading.Thread(target=ping_loop, daemon=True) + thread.start() + try: + # A case load blocks the main thread for as long as it takes to read the grid. + case_path = dataroot.PATH + "/TEST10K_FLT_LGR_NNC/TEST10K_FLT_LGR_NNC.EGRID" + started = time.monotonic() + rips_instance.project.load_case(path=case_path) + elapsed = time.monotonic() - started + finally: + stop.set() + thread.join(timeout=10.0) + + if elapsed < 0.2: + pytest.skip(f"blocking call was too short to be meaningful ({elapsed:.3f} s)") + + assert not errors, f"pings failed while the server was busy: {errors}" + + # A ping issued just after the blocking call starts would previously only be + # answered once that call finished, i.e. a latency of roughly `elapsed`. + assert max(latencies) < 0.5 * elapsed, ( + f"pings were starved by the busy main thread: max latency " + f"{max(latencies):.3f} s for a {elapsed:.3f} s blocking call" + ) + assert len(latencies) >= 3, "expected several pings during the blocking call" diff --git a/GrpcInterface/RiaGrpcAppService.cpp b/GrpcInterface/RiaGrpcAppService.cpp index abec461eab..5f0f248538 100644 --- a/GrpcInterface/RiaGrpcAppService.cpp +++ b/GrpcInterface/RiaGrpcAppService.cpp @@ -68,7 +68,17 @@ grpc::Status std::vector RiaGrpcAppService::createCallbacks() { using Self = RiaGrpcAppService; - return { new RiaGrpcUnaryCallback( this, &Self::GetVersion, &Self::RequestGetVersion ), + + // GetVersion reads compile-time constants only. It is flagged to run directly on the gRPC + // completion queue thread so that clients can check liveness even while the main thread is + // busy with a long-running request (see issue #14683). All other methods touch application + // state and must be processed on the main thread. + const bool runsOnServerThread = true; + + return { new RiaGrpcUnaryCallback( this, + &Self::GetVersion, + &Self::RequestGetVersion, + runsOnServerThread ), new RiaGrpcUnaryCallback( this, &Self::Exit, &Self::RequestExit ), new RiaGrpcUnaryCallback( this, &Self::GetRuntimeInfo, diff --git a/GrpcInterface/RiaGrpcCallbacks.h b/GrpcInterface/RiaGrpcCallbacks.h index 004c8eaca1..cdcee94e27 100644 --- a/GrpcInterface/RiaGrpcCallbacks.h +++ b/GrpcInterface/RiaGrpcCallbacks.h @@ -62,6 +62,13 @@ public: virtual void onProcessRequest() = 0; virtual void onFinishRequest() {} + // Callbacks that touch no application state can be served directly on the gRPC completion + // queue thread, instead of being queued for the main thread. This keeps liveness requests + // answerable while the main thread is occupied by a long-running request. Only override for + // methods that read no project/PDM data - anything touching the data model must be processed + // on the main thread. + virtual bool runsOnServerThread() const { return false; } + inline CallState callState() const; inline const Status& status() const; inline void setNextCallState( CallState state ); @@ -118,11 +125,15 @@ public: using MethodRequestT = std::function< void( ServiceT&, ServerContext*, RequestT*, ResponseWriterT*, CompletionQueue*, ServerCompletionQueue*, void* )>; - RiaGrpcUnaryCallback( ServiceT* service, MethodImplT methodImpl, MethodRequestT methodRequest ); + RiaGrpcUnaryCallback( ServiceT* service, + MethodImplT methodImpl, + MethodRequestT methodRequest, + bool runsOnServerThread = false ); RiaGrpcCallbackInterface* createNewFromThis() const override; void createRequestHandler( ServerCompletionQueue* completionQueue ) override; void onProcessRequest() override; + bool runsOnServerThread() const override { return m_runsOnServerThread; } protected: QString methodType() const override; @@ -132,6 +143,7 @@ private: ResponseWriterT m_responder; MethodImplT m_methodImpl; MethodRequestT m_methodRequest; + bool m_runsOnServerThread; }; //================================================================================================== diff --git a/GrpcInterface/RiaGrpcCallbacks.inl b/GrpcInterface/RiaGrpcCallbacks.inl index b0a708ed0d..627d92366c 100644 --- a/GrpcInterface/RiaGrpcCallbacks.inl +++ b/GrpcInterface/RiaGrpcCallbacks.inl @@ -93,11 +93,13 @@ ReplyT& RiaGrpcServiceCallback::reply() template RiaGrpcUnaryCallback::RiaGrpcUnaryCallback( ServiceT* service, MethodImplT methodImpl, - MethodRequestT methodRequest ) + MethodRequestT methodRequest, + bool runsOnServerThread ) : RiaGrpcServiceCallback( service ) , m_responder( &m_context ) , m_methodImpl( methodImpl ) , m_methodRequest( methodRequest ) + , m_runsOnServerThread( runsOnServerThread ) { } @@ -107,7 +109,10 @@ RiaGrpcUnaryCallback::RiaGrpcUnaryCallback( ServiceT template RiaGrpcCallbackInterface* RiaGrpcUnaryCallback::createNewFromThis() const { - return new RiaGrpcUnaryCallback( this->m_service, this->m_methodImpl, this->m_methodRequest ); + return new RiaGrpcUnaryCallback( this->m_service, + this->m_methodImpl, + this->m_methodRequest, + m_runsOnServerThread ); } //-------------------------------------------------------------------------------------------------- diff --git a/GrpcInterface/RiaGrpcServer.cpp b/GrpcInterface/RiaGrpcServer.cpp index 1340bbb6ef..4d03c511fc 100644 --- a/GrpcInterface/RiaGrpcServer.cpp +++ b/GrpcInterface/RiaGrpcServer.cpp @@ -38,6 +38,8 @@ #include +#include +#include #include using grpc::CompletionQueue; @@ -78,7 +80,14 @@ private: std::list m_unprocessedRequests; std::list m_allocatedCallbakcs; std::mutex m_requestMutex; - std::thread m_thread; + // Guards m_allocatedCallbakcs, which is touched both from the main thread and from the + // completion queue thread when serving callbacks flagged as runsOnServerThread(). + std::mutex m_allocatedCallbackMutex; + // Serializes shutdown against an in-flight dispatch on the completion queue thread, so that + // no request handler can be registered on a queue that is about to be shut down. + std::mutex m_shutdownMutex; + std::atomic m_isShuttingDown{ false }; + std::thread m_thread; }; //-------------------------------------------------------------------------------------------------- @@ -222,6 +231,14 @@ void RiaGrpcServerImpl::quit() { RiaLogging::info( "Shutting down gRPC server" ); + // Stop serving callbacks on the completion queue thread before shutting the queue down. + // Once the queue is shut down, no new request handler may be registered on it. Taking the + // lock also waits for any dispatch already in progress on that thread. + { + std::lock_guard shutdownLock( m_shutdownMutex ); + m_isShuttingDown = true; + } + // See the following link for details on how to shut down a GRPC server // https://github.com/grpc/grpc/blob/master/include/grpcpp/server_builder.h#L147 @@ -282,12 +299,31 @@ void RiaGrpcServerImpl::waitForNextRequest() while ( m_completionQueue->Next( &tag, &ok ) ) { - std::lock_guard requestLock( m_requestMutex ); - RiaGrpcCallbackInterface* method = static_cast( tag ); + RiaGrpcCallbackInterface* method = static_cast( tag ); if ( !ok ) { method->setNextCallState( RiaGrpcCallbackInterface::FINISH_REQUEST ); } + + // Callbacks that read no application state are served here instead of being queued for + // the main thread. ResInsight processes queued requests from the main thread, so a + // long-running request (large export, schedule generation, heavy grid computation) or an + // open progress dialog would otherwise leave liveness requests unanswered for as long as + // that operation runs, making a busy ResInsight look like a dead one (issue #14683). + // + // During shutdown the completion queue is draining and no new request handlers may be + // registered on it, so fall back to queueing (those callbacks are deleted in quit()). + if ( method->runsOnServerThread() ) + { + std::lock_guard shutdownLock( m_shutdownMutex ); + if ( !m_isShuttingDown ) + { + process( method ); + continue; + } + } + + std::lock_guard requestLock( m_requestMutex ); m_unprocessedRequests.push_back( method ); } } @@ -303,6 +339,7 @@ void RiaGrpcServerImpl::process( RiaGrpcCallbackInterface* method ) { method->createRequestHandler( m_completionQueue.get() ); + std::lock_guard callbackLock( m_allocatedCallbackMutex ); m_allocatedCallbakcs.push_back( method ); } else if ( method->callState() == RiaGrpcCallbackInterface::INIT_REQUEST_STARTED ) @@ -323,7 +360,10 @@ void RiaGrpcServerImpl::process( RiaGrpcCallbackInterface* method ) process( method->createNewFromThis() ); - m_allocatedCallbakcs.remove( method ); + { + std::lock_guard callbackLock( m_allocatedCallbackMutex ); + m_allocatedCallbakcs.remove( method ); + } delete method; }