make the operations on responses_ in endpoint::worker_common thread-safe

src/tateyama/endpoint/common/worker_common.h

@@ -21,6 +21,7 @@
 #include <functional>
 #include <map>
 #include <vector>
+#include <mutex>
 
 #include <tateyama/status.h>
 #include <tateyama/api/server/request.h>
@@ -198,25 +199,34 @@ class worker_common {
             notify_client(res.get(), tateyama::proto::diagnostics::Code::INVALID_REQUEST, ss.str());
             return false;
         }
-        if (auto itr = responses_.find(slot); itr != responses_.end()) {
-            if (auto ptr = itr->second.lock(); ptr) {
-                ptr->set_cancel();
+        {
+            std::lock_guard<std::mutex> lock(mtx_responses_);
+            if (auto itr = responses_.find(slot); itr != responses_.end()) {
+                if (auto ptr = itr->second.lock(); ptr) {
+                    ptr->set_cancel();
+                }
             }
         }
         return true;
     }
 
     void register_response(std::size_t slot, const std::shared_ptr<tateyama::endpoint::common::response>& response) noexcept {
+        std::lock_guard<std::mutex> lock(mtx_responses_);
+        if (auto itr = responses_.find(slot); itr != responses_.end()) {
+            responses_.erase(itr);
+        }
         responses_.emplace(slot, response);
     }
     void remove_response(std::size_t slot) noexcept {
+        std::lock_guard<std::mutex> lock(mtx_responses_);
         responses_.erase(slot);
     }
 
 private:
     tateyama::session::session_variable_set session_variable_set_;
     const std::shared_ptr<tateyama::session::resource::session_context_impl> session_context_;
     std::map<std::size_t, std::weak_ptr<tateyama::endpoint::common::response>> responses_{};
+    std::mutex mtx_responses_{};
 
     std::string_view connection_label(connection_type con) {
         switch (con) {

src/tateyama/endpoint/ipc/ipc_response.h

@@ -112,10 +112,6 @@ class alignas(64) ipc_response : public tateyama::endpoint::common::response {
         session_id_ = id;
     }
 
-    [[nodiscard]] bool check_cancel() const override {
-        return false;
-    }
-
 private:
     std::shared_ptr<server_wire_container> server_wire_;
     std::size_t index_;

src/tateyama/endpoint/stream/stream.h

@@ -454,7 +454,7 @@ class connection_socket
     std::mutex num_mutex_{};
     std::condition_variable num_condition_{};
     std::size_t timeout_;
-    
+
     friend class stream_socket;
 
     [[nodiscard]] bool is_socket_available() {

src/tateyama/endpoint/stream/stream_response.h

@@ -102,10 +102,6 @@ class alignas(64) stream_response : public tateyama::endpoint::common::response
         session_id_ = id;
     }
 
-    [[nodiscard]] bool check_cancel() const override {
-        return false;
-    }
-
 private:
     std::shared_ptr<stream_socket> stream_;
     std::uint16_t index_;

src/tateyama/proto/endpoint/request.proto

@@ -40,11 +40,6 @@ message Handshake {
     WireInformation wire_information = 11;
 }
 
-// cancel operation.
-message Cancel {
-    // no special properties.
-}
-
 // client information
 message ClientInformation {
     // the connection label.
@@ -81,3 +76,8 @@ message WireInformation {
         uint64 maximum_concurrent_result_sets = 1;
     }
 }
+
+// cancel operation.
+message Cancel {
+    // no special properties.
+}