mirror of
https://github.com/OPM/ResInsight.git
synced 2026-10-24 22:27:55 -05:00
Kristian Bendiksen
35614d160e
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.
235 lines
9.7 KiB
C++
235 lines
9.7 KiB
C++
/////////////////////////////////////////////////////////////////////////////////
|
|
//
|
|
// Copyright (C) 2019- Equinor ASA
|
|
//
|
|
// ResInsight is free software: you can redistribute it and/or modify
|
|
// it under the terms of the GNU General Public License as published by
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
// (at your option) any later version.
|
|
//
|
|
// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY
|
|
// WARRANTY; without even the implied warranty of MERCHANTABILITY or
|
|
// FITNESS FOR A PARTICULAR PURPOSE.
|
|
//
|
|
// See the GNU General Public License at <http://www.gnu.org/licenses/gpl.html>
|
|
// for more details.
|
|
//
|
|
//////////////////////////////////////////////////////////////////////////////////
|
|
#pragma once
|
|
|
|
#include <QString>
|
|
#include <grpc/support/log.h>
|
|
#include <grpcpp/grpcpp.h>
|
|
#include <grpcpp/server_context.h>
|
|
#include <grpcpp/support/async_stream.h>
|
|
#include <grpcpp/support/async_unary_call.h>
|
|
#include <grpcpp/support/sync_stream.h>
|
|
|
|
using grpc::CompletionQueue;
|
|
using grpc::ServerAsyncReader;
|
|
using grpc::ServerAsyncResponseWriter;
|
|
using grpc::ServerAsyncWriter;
|
|
using grpc::ServerCompletionQueue;
|
|
using grpc::ServerContext;
|
|
using grpc::Status;
|
|
|
|
//==================================================================================================
|
|
//
|
|
// Base class for all gRPC-callbacks
|
|
//
|
|
//==================================================================================================
|
|
class RiaGrpcCallbackInterface
|
|
{
|
|
public:
|
|
enum CallState
|
|
{
|
|
CREATE_HANDLER,
|
|
INIT_REQUEST_STARTED,
|
|
INIT_REQUEST_COMPLETED,
|
|
PROCESS_REQUEST,
|
|
FINISH_REQUEST
|
|
};
|
|
|
|
public:
|
|
inline RiaGrpcCallbackInterface();
|
|
|
|
virtual ~RiaGrpcCallbackInterface() {}
|
|
virtual QString name() const = 0;
|
|
virtual RiaGrpcCallbackInterface* createNewFromThis() const = 0;
|
|
virtual void createRequestHandler( ServerCompletionQueue* completionQueue ) = 0;
|
|
virtual void onInitRequestStarted() {}
|
|
virtual void onInitRequestCompleted() {}
|
|
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 );
|
|
|
|
protected:
|
|
CallState m_state;
|
|
Status m_status;
|
|
};
|
|
|
|
//==================================================================================================
|
|
//
|
|
// Templated gRPC-callback base class
|
|
//
|
|
//==================================================================================================
|
|
template <typename ServiceT, typename RequestT, typename ReplyT>
|
|
class RiaGrpcServiceCallback : public RiaGrpcCallbackInterface
|
|
{
|
|
public:
|
|
RiaGrpcServiceCallback( ServiceT* service );
|
|
~RiaGrpcServiceCallback() override;
|
|
|
|
QString name() const override;
|
|
const RequestT& request() const;
|
|
ReplyT& reply();
|
|
|
|
protected:
|
|
virtual QString methodType() const = 0;
|
|
|
|
protected:
|
|
ServiceT* m_service;
|
|
RequestT m_request;
|
|
ReplyT m_reply;
|
|
};
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
template <typename ServiceT, typename RequestT, typename ReplyT>
|
|
RiaGrpcServiceCallback<ServiceT, RequestT, ReplyT>::~RiaGrpcServiceCallback()
|
|
{
|
|
}
|
|
|
|
//==================================================================================================
|
|
//
|
|
// Templated gRPC-callback for non-streaming services
|
|
//
|
|
//==================================================================================================
|
|
template <typename ServiceT, typename RequestT, typename ReplyT>
|
|
class RiaGrpcUnaryCallback : public RiaGrpcServiceCallback<ServiceT, RequestT, ReplyT>
|
|
{
|
|
public:
|
|
using ResponseWriterT = ServerAsyncResponseWriter<ReplyT>;
|
|
using MethodImplT = std::function<Status( ServiceT&, ServerContext*, const RequestT*, ReplyT* )>;
|
|
using MethodRequestT = std::function<
|
|
void( ServiceT&, ServerContext*, RequestT*, ResponseWriterT*, CompletionQueue*, ServerCompletionQueue*, void* )>;
|
|
|
|
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;
|
|
|
|
private:
|
|
ServerContext m_context;
|
|
ResponseWriterT m_responder;
|
|
MethodImplT m_methodImpl;
|
|
MethodRequestT m_methodRequest;
|
|
bool m_runsOnServerThread;
|
|
};
|
|
|
|
//==================================================================================================
|
|
//
|
|
// Templated server->client *streaming* gRPC-callback calling service implementation callbacks
|
|
//
|
|
// The streaming callback needs a state handler for setting up and maintaining order.
|
|
//
|
|
// A fully functional state handler for server->client streaming needs to implement the following methods:
|
|
// 1. Default Constructor
|
|
// 2. grpc::Status init(const grpc::Message* request)
|
|
//
|
|
//==================================================================================================
|
|
template <typename ServiceT, typename RequestT, typename ReplyT, typename StateHandlerT>
|
|
class RiaGrpcServerToClientStreamCallback : public RiaGrpcServiceCallback<ServiceT, RequestT, ReplyT>
|
|
{
|
|
public:
|
|
using ResponseWriterT = ServerAsyncWriter<ReplyT>;
|
|
using MethodImplT = std::function<Status( ServiceT&, ServerContext*, const RequestT*, ReplyT*, StateHandlerT* )>;
|
|
using MethodRequestT = std::function<
|
|
void( ServiceT&, ServerContext*, RequestT*, ResponseWriterT*, CompletionQueue*, ServerCompletionQueue*, void* )>;
|
|
|
|
RiaGrpcServerToClientStreamCallback( ServiceT* service,
|
|
MethodImplT methodImpl,
|
|
MethodRequestT methodRequest,
|
|
StateHandlerT* stateHandler );
|
|
|
|
RiaGrpcCallbackInterface* createNewFromThis() const override;
|
|
void createRequestHandler( ServerCompletionQueue* completionQueue ) override;
|
|
void onInitRequestCompleted() override;
|
|
void onProcessRequest() override;
|
|
|
|
protected:
|
|
QString methodType() const override;
|
|
|
|
private:
|
|
ServerContext m_context;
|
|
ResponseWriterT m_responder;
|
|
MethodImplT m_methodImpl;
|
|
MethodRequestT m_methodRequest;
|
|
std::unique_ptr<StateHandlerT> m_stateHandler;
|
|
};
|
|
|
|
//==================================================================================================
|
|
//
|
|
// Templated client->server *streaming* gRPC-callback calling service implementation callbacks
|
|
//
|
|
// The streaming callback needs a state handler for setting up and maintaining order.
|
|
//
|
|
// A fully functional state handler for client->server streaming needs to implement the following methods:
|
|
// 1. Default Constructor
|
|
// 2. grpc::Status init(const grpc::Message* request)
|
|
// 3. void finish() any updates required after completion
|
|
//
|
|
//==================================================================================================
|
|
template <typename ServiceT, typename RequestT, typename ReplyT, typename StateHandlerT>
|
|
class RiaGrpcClientToServerStreamCallback : public RiaGrpcServiceCallback<ServiceT, RequestT, ReplyT>
|
|
{
|
|
public:
|
|
using RequestReaderT = ServerAsyncReader<ReplyT, RequestT>;
|
|
using MethodImplT = std::function<Status( ServiceT&, ServerContext*, const RequestT*, ReplyT*, StateHandlerT* )>;
|
|
using MethodRequestT =
|
|
std::function<void( ServiceT&, ServerContext*, RequestReaderT*, CompletionQueue*, ServerCompletionQueue*, void* )>;
|
|
|
|
RiaGrpcClientToServerStreamCallback( ServiceT* service,
|
|
MethodImplT methodImpl,
|
|
MethodRequestT methodRequest,
|
|
StateHandlerT* stateHandler );
|
|
|
|
RiaGrpcCallbackInterface* createNewFromThis() const override;
|
|
void createRequestHandler( ServerCompletionQueue* completionQueue ) override;
|
|
void onInitRequestStarted() override;
|
|
void onInitRequestCompleted() override;
|
|
void onProcessRequest() override;
|
|
void onFinishRequest() override;
|
|
|
|
protected:
|
|
QString methodType() const override;
|
|
|
|
private:
|
|
ServerContext m_context;
|
|
RequestReaderT m_reader;
|
|
MethodImplT m_methodImpl;
|
|
MethodRequestT m_methodRequest;
|
|
std::unique_ptr<StateHandlerT> m_stateHandler;
|
|
};
|
|
|
|
#include "RiaGrpcCallbacks.inl"
|