Plausible improvements

Author: eddyashton

include/ccf/endpoint.h

@@ -264,7 +264,7 @@ namespace ccf::endpoints
     // committed (ie - assigned a transaction ID)
     LocallyCommittedEndpointFunction locally_committed_func;
 
-    // Functor which is invoked to modify the response after it is reaches a
+    // Functor which is invoked to modify the response after it reaches a
     // terminal consensus state (ie - it is either globally committed, or
     // invalidated)
     ConsensusCommittedEndpointFunction consensus_committed_func;

src/consensus/aft/raft.h

@@ -2607,7 +2607,7 @@ namespace aft
       if (commit_callbacks != nullptr)
       {
         const auto term = get_term_internal(idx);
-        commit_callbacks->execute_callbacks({term, idx});
+        commit_callbacks->trigger_callbacks({term, idx});
       }
 
       RAFT_DEBUG_FMT("Commit on {}: {}", state->node_id, idx);

src/http/http_session.h

@@ -38,13 +38,12 @@ namespace http
       std::unique_ptr<ccf::tls::Context> ctx,
       const ccf::http::ParserConfiguration& configuration,
       const std::shared_ptr<ErrorReporter>& error_reporter_,
-      const std::shared_ptr<ccf::CommitCallbackSubsystem>& commit_callbacks =
-        nullptr) :
+      const std::shared_ptr<ccf::CommitCallbackSubsystem>& commit_callbacks_) :
       HTTPSession(session_id_, writer_factory, std::move(ctx)),
       request_parser(*this, configuration),
       rpc_map(std::move(rpc_map_)),
       error_reporter(error_reporter_),
-      commit_callbacks(commit_callbacks),
+      commit_callbacks(commit_callbacks_),
       interface_id(std::move(interface_id_))
     {}
 
@@ -186,25 +185,29 @@ namespace http
           // maintain session consistency
           ccf::tasks::Resumable paused_task = ccf::tasks::pause_current_task();
 
+          // shared_from_this returns a base session type
+          std::shared_ptr<ccf::ThreadedSession> self = shared_from_this();
+
           // Register for a callback when this TxID is committed (or
           // invalidated)
           commit_callbacks->add_callback(
             tx_id,
-            [this, rpc_ctx, paused_task, committed_func](
+            [self, rpc_ctx, paused_task, committed_func](
               ccf::TxID transaction_id, ccf::FinalTxStatus status) {
               // Let the handler modify the response
               committed_func(rpc_ctx, transaction_id, status);
 
               // Write the response
-              this->send_response(
+              send_response_impl(
+                *self,
                 rpc_ctx->get_response_http_status(),
                 rpc_ctx->get_response_headers(),
                 rpc_ctx->get_response_trailers(),
                 std::move(rpc_ctx->take_response_body()));
 
               if (rpc_ctx->terminate_session)
               {
-                close_session();
+                self->close_session();
               }
 
               // Resume processing work for this session
@@ -240,11 +243,12 @@ namespace http
       }
     }
 
-    bool send_response(
+    static bool send_response_impl(
+      ccf::ThreadedSession& session,
       ccf::http_status status_code,
       ccf::http::HeaderMap&& headers,
       ccf::http::HeaderMap&& trailers,
-      std::vector<uint8_t>&& body) override
+      std::vector<uint8_t>&& body)
     {
       if (!trailers.empty())
       {
@@ -263,9 +267,23 @@ namespace http
         false /* Don't overwrite any existing content-length header */
       );
 
-      send_data(response.build_response());
+      session.send_data(response.build_response());
       return true;
     }
+
+    bool send_response(
+      ccf::http_status status_code,
+      ccf::http::HeaderMap&& headers,
+      ccf::http::HeaderMap&& trailers,
+      std::vector<uint8_t>&& body) override
+    {
+      return send_response_impl(
+        *this,
+        status_code,
+        std::move(headers),
+        std::move(trailers),
+        std::move(body));
+    }
   };
 
   class HTTPClientSession : public HTTPSession,

src/node/commit_callback_subsystem.h

@@ -17,14 +17,16 @@ namespace ccf
 
     std::optional<ccf::TxID> known_commit = std::nullopt;
 
-    std::mutex callbacks_mutex;
+    // Use a recursive mutex so that `add_callback` may safely be called while a
+    // callback is executing (and the mutex is locked)
+    std::recursive_mutex callbacks_mutex;
 
   public:
     CommitCallbackSubsystem() = default;
 
     void add_callback(ccf::TxID tx_id, CommitCallback&& callback) override
     {
-      std::lock_guard<std::mutex> guard(callbacks_mutex);
+      std::lock_guard<std::recursive_mutex> guard(callbacks_mutex);
 
       if (known_commit.has_value())
       {
@@ -49,9 +51,9 @@ namespace ccf
         std::make_pair(tx_id, std::move(callback)));
     }
 
-    void execute_callbacks(ccf::TxID committed)
+    void trigger_callbacks(ccf::TxID committed)
     {
-      std::lock_guard<std::mutex> guard(callbacks_mutex);
+      std::lock_guard<std::recursive_mutex> guard(callbacks_mutex);
 
       known_commit = committed;