make group::flow::setup_partitions callable from multiple threads

project-tsurugi/tsurugi-issues#801

src/jogasaki/executor/exchange/group/flow.cpp

@@ -100,6 +100,7 @@ takatori::util::sequence_view<std::shared_ptr<model::task>> flow::create_tasks()
 }
 
 flow::sinks_sources flow::setup_partitions(std::size_t partitions) {
+    std::unique_lock lk{mutex_};
     sinks_.reserve(partitions);
     for(std::size_t i=0; i < partitions; ++i) {
         sinks_.emplace_back(std::make_unique<group::sink>(downstream_partitions_, info_, context()));

src/jogasaki/executor/exchange/group/flow.h

@@ -55,10 +55,10 @@ class flow : public shuffle::flow {
     using field_index_type = meta::record_meta::field_index_type;
 
     ~flow() override;
-    flow(flow const& other) = default;
-    flow& operator=(flow const& other) = default;
-    flow(flow&& other) noexcept = default;
-    flow& operator=(flow&& other) noexcept = default;
+    flow(flow const& other) = delete;
+    flow& operator=(flow const& other) = delete;
+    flow(flow&& other) noexcept = delete;
+    flow& operator=(flow&& other) noexcept = delete;
 
     /**
      * @brief create new instance with empty schema (for testing)
@@ -118,6 +118,7 @@ class flow : public shuffle::flow {
     class request_context* context_{};
     step* owner_{};
     std::size_t downstream_partitions_{default_partitions};
+    std::mutex mutex_{};
 };
 
 }  // namespace jogasaki::executor::exchange::group