#14683 Serve liveness requests on the gRPC server thread

Requests are queued by the completion queue thread and processed from the main
thread. A long running request, or an open progress dialog in the GUI, therefore
left every request unanswered for as long as that operation lasted, including
the liveness checks a client uses to decide whether ResInsight is still there.

Let a callback declare that it touches no application state, and serve those
directly on the completion queue thread instead of queueing them. GetVersion is
the only such callback, since it reads compile-time constants only. Everything
that touches the project or the data model is still processed on the main
thread, so no request can observe or modify the data model while another one is
running.

The allocated callback list is now used from both threads and is guarded by a
mutex. Shutdown takes a lock before flagging that it is in progress, so that it
waits for any dispatch already running on the completion queue thread and no new
request handler can be registered on a queue that is about to be shut down.

During a blocking call of 3.25 s, 65 pings were answered with a maximum latency
of 6.2 ms. Before this change none of them were answered until the call
completed.

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"

GrpcInterface/RiaGrpcAppService.cpp

@@ -68,7 +68,17 @@ grpc::Status
 std::vector<RiaGrpcCallbackInterface*> RiaGrpcAppService::createCallbacks()
 {
     using Self = RiaGrpcAppService;
-    return { new RiaGrpcUnaryCallback<Self, rips::Empty, rips::Version>( 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<Self, rips::Empty, rips::Version>( this,
+                                                                         &Self::GetVersion,
+                                                                         &Self::RequestGetVersion,
+                                                                         runsOnServerThread ),
              new RiaGrpcUnaryCallback<Self, rips::Empty, rips::Empty>( this, &Self::Exit, &Self::RequestExit ),
              new RiaGrpcUnaryCallback<Self, rips::Empty, rips::RuntimeInfo>( this,
                                                                              &Self::GetRuntimeInfo,

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;
 };
 
 //==================================================================================================

GrpcInterface/RiaGrpcCallbacks.inl

@@ -93,11 +93,13 @@ ReplyT& RiaGrpcServiceCallback<ServiceT, RequestT, ReplyT>::reply()
 template <typename ServiceT, typename RequestT, typename ReplyT>
 RiaGrpcUnaryCallback<ServiceT, RequestT, ReplyT>::RiaGrpcUnaryCallback( ServiceT*      service,
                                                                         MethodImplT    methodImpl,
-                                                                        MethodRequestT methodRequest )
+                                                                        MethodRequestT methodRequest,
+                                                                        bool           runsOnServerThread )
     : RiaGrpcServiceCallback<ServiceT, RequestT, ReplyT>( service )
     , m_responder( &m_context )
     , m_methodImpl( methodImpl )
     , m_methodRequest( methodRequest )
+    , m_runsOnServerThread( runsOnServerThread )
 {
 }
 
@@ -107,7 +109,10 @@ RiaGrpcUnaryCallback<ServiceT, RequestT, ReplyT>::RiaGrpcUnaryCallback( ServiceT
 template <typename ServiceT, typename RequestT, typename ReplyT>
 RiaGrpcCallbackInterface* RiaGrpcUnaryCallback<ServiceT, RequestT, ReplyT>::createNewFromThis() const
 {
-    return new RiaGrpcUnaryCallback<ServiceT, RequestT, ReplyT>( this->m_service, this->m_methodImpl, this->m_methodRequest );
+    return new RiaGrpcUnaryCallback<ServiceT, RequestT, ReplyT>( this->m_service,
+                                                                 this->m_methodImpl,
+                                                                 this->m_methodRequest,
+                                                                 m_runsOnServerThread );
 }
 
 //--------------------------------------------------------------------------------------------------

GrpcInterface/RiaGrpcServer.cpp

@@ -38,6 +38,8 @@
 
 #include <QTcpServer>
 
+#include <atomic>
+#include <mutex>
 #include <thread>
 
 using grpc::CompletionQueue;
@@ -78,7 +80,14 @@ private:
     std::list<RiaGrpcCallbackInterface*>                m_unprocessedRequests;
     std::list<RiaGrpcCallbackInterface*>                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<bool> 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<std::mutex> 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<std::mutex> requestLock( m_requestMutex );
-        RiaGrpcCallbackInterface*   method = static_cast<RiaGrpcCallbackInterface*>( tag );
+        RiaGrpcCallbackInterface* method = static_cast<RiaGrpcCallbackInterface*>( 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<std::mutex> shutdownLock( m_shutdownMutex );
+            if ( !m_isShuttingDown )
+            {
+                process( method );
+                continue;
+            }
+        }
+
+        std::lock_guard<std::mutex> 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<std::mutex> 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<std::mutex> callbackLock( m_allocatedCallbackMutex );
+            m_allocatedCallbakcs.remove( method );
+        }
 
         delete method;
     }