Merge pull request #171 from project-tsurugi/wip/i_592

change the method of storing worker threads that were denied TCP/IP c…
project-tsurugi · Jan 19, 2024 · ff3cde9 · ff3cde9
2 parents ff45739 + 991a83e
commit ff3cde9
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 14 deletions.
diff --git a/src/tateyama/endpoint/stream/bootstrap/stream_listener.h b/src/tateyama/endpoint/stream/bootstrap/stream_listener.h
@@ -103,6 +103,7 @@ class stream_listener {
 
         arrive_and_wait();
         while(true) {
+            undertakers_.erase(std::remove_if(std::begin(undertakers_), std::end(undertakers_), [](std::unique_ptr<stream_worker>& worker){ return worker->wait_for() == std::future_status::ready; }), std::cend(undertakers_));
             std::shared_ptr<stream_socket> stream{};
             try {
                 stream = connection_socket_->accept();
@@ -128,16 +129,8 @@ class stream_listener {
                 if (!found) {
                     auto worker_decline = std::make_unique<stream_worker>(*router_, session_id, std::move(stream), status_->database_info(), true);
                     auto *worker_decline_ptr = worker_decline.get();
-                    worker_decline->invoke([&]{worker_decline->run([&](){
-                        std::unique_lock lock{mutex_};
-                        if (auto itr = undertakers_.find(worker_decline_ptr); itr != undertakers_.end()) {
-                            undertakers_.erase(itr);
-                        }
-                    });});
-                    {
-                        std::unique_lock lock{mutex_};
-                        undertakers_.emplace(std::move(worker_decline));
-                    }
+                    undertakers_.emplace_back(std::move(worker_decline));
+                    worker_decline_ptr->invoke([&]{worker_decline_ptr->run();});
                     LOG_LP(ERROR) << "the number of sessions exceeded the limit (" << workers_.size() << ")";
                     continue;
                 }
@@ -170,7 +163,7 @@ class stream_listener {
     const std::shared_ptr<status_info::resource::bridge> status_{};
     std::unique_ptr<connection_socket> connection_socket_{};
     std::vector<std::unique_ptr<stream_worker>> workers_{};
-    std::set<std::unique_ptr<stream_worker>, tateyama::endpoint::common::pointer_comp<stream_worker>> undertakers_{};
+    std::vector<std::unique_ptr<stream_worker>> undertakers_{};
     std::mutex mutex_{};
 
     boost::barrier sync{2};

diff --git a/src/tateyama/endpoint/stream/bootstrap/stream_worker.cpp b/src/tateyama/endpoint/stream/bootstrap/stream_worker.cpp
@@ -23,7 +23,7 @@
 
 namespace tateyama::endpoint::stream::bootstrap {
 
-void stream_worker::run(const std::function<void(void)>& clean_up)
+void stream_worker::run()
 {
     {
         std::uint16_t slot{};
@@ -43,7 +43,6 @@ void stream_worker::run(const std::function<void(void)>& clean_up)
             } else {
                 LOG_LP(INFO) << "illegal procedure (receive a request in spite of a decline case)";  // should not reach here
             }
-            clean_up();
             return;
         }
 

diff --git a/src/tateyama/endpoint/stream/bootstrap/stream_worker.h b/src/tateyama/endpoint/stream/bootstrap/stream_worker.h
@@ -55,7 +55,7 @@ class stream_worker : public tateyama::endpoint::common::worker_common {
     stream_worker& operator = (stream_worker const&) = delete;
     stream_worker& operator = (stream_worker&&) = delete;
 
-    void run(const std::function<void(void)>& clean_up = [](){});
+    void run();
     friend class stream_provider;
 
  private: