Repository: alibaba/GraphScope (language: python)

Issue:
feat(Interactive): Restart the actors when changing graph
With AdminService introduced by [this PR](https://github.com/alibaba/GraphScope/pull/3322), Interactive are now able to switch between graphs at runtime(BUT will only serve on on graph at one moment).

During graph switching, all actors should be stopped, and should not process any new requests.


Discussion / hints:

Fix patch:
diff --git a/docs/flex/interactive/development/admin_service.md b/docs/flex/interactive/development/admin_service.md
index 241b78137896..e24f9578181e 100644
--- a/docs/flex/interactive/development/admin_service.md
+++ b/docs/flex/interactive/development/admin_service.md
@@ -942,7 +942,20 @@ curl -X DELETE -H "Content-Type: application/json" "http://[host]/v1/graph/{grap
 
 #### Description 
 
-Start the query service on a graph.
+Start the query service on a graph. The `graph_name` param can be empty, indicating restarting on current running graph.
+
+1. After the AdminService receives this request, the current actor scope for query actors will be cancelled.
+2. During the scope cancellation process of the query actors or after scope cancellation is completed, all requests sent to the query_service will fail and be rejected. 
+The response of the http request will be like
+```json
+{
+  "code": 500,
+  "message" : "Unable to send message, the target actor has been canceled!"
+}
+```
+3. After the previous graph is closed and new graph is opened, the new query actors will be available in a new scope. 
+4. The query service is now ready to serve requests on the new graph.
+ 
 
 #### HTTP Request
 - **Method**: POST
diff --git a/flex/bin/bulk_loader.cc b/flex/bin/bulk_loader.cc
index 13104a5148cc..54112a176077 100644
--- a/flex/bin/bulk_loader.cc
+++ b/flex/bin/bulk_loader.cc
@@ -85,8 +85,8 @@ int main(int argc, char** argv) {
   }
   std::filesystem::path serial_path = data_dir_path / "schema";
   if (std::filesystem::exists(serial_path)) {
-    LOG(ERROR) << "data directory is not empty";
-    return -1;
+    LOG(WARNING) << "data directory is not empty";
+    return 0;
   }
 
   auto loader = gs::LoaderFactory::CreateFragmentLoader(
diff --git a/flex/engines/http_server/actor/admin_actor.act.cc b/flex/engines/http_server/actor/admin_actor.act.cc
index 4a91f0a5c9cb..0a9f7989468e 100644
--- a/flex/engines/http_server/actor/admin_actor.act.cc
+++ b/flex/engines/http_server/actor/admin_actor.act.cc
@@ -277,7 +277,8 @@ seastar::future<query_result> admin_actor::update_procedure(
   }
 }
 
-// Manage the service of one graph.
+// Start service on a graph first means stop all current running actors, then
+// switch graph and and create new actors with a unused scope_id.
 seastar::future<query_result> admin_actor::start_service(
     query_param&& query_param) {
   // parse query_param.content as json and get graph_name
@@ -296,49 +297,59 @@ seastar::future<query_result> admin_actor::start_service(
           << server::WorkDirManipulator::GetRunningGraph();
     }
     LOG(WARNING) << "Starting service with graph: " << graph_name;
+  } catch (std::exception& e) {
+    LOG(ERROR) << "Fail to Start service: ";
+    return seastar::make_exception_future<query_result>(
+        std::runtime_error(e.what()));
+  }
+
+  auto schema_result = server::WorkDirManipulator::GetGraphSchema(graph_name);
+  if (!schema_result.ok()) {
+    LOG(ERROR) << "Fail to get graph schema: "
+               << schema_result.status().error_message() << ", " << graph_name;
+    return seastar::make_exception_future<query_result>(std::runtime_error(
+        "Fail to get graph schema: " + schema_result.status().error_message() +
+        ", " + graph_name));
+  }
+  auto& schema_value = schema_result.value();
+  auto data_dir = server::WorkDirManipulator::GetDataDirectory(graph_name);
+  if (!data_dir.ok()) {
+    LOG(ERROR) << "Fail to get data directory: "
+               << data_dir.status().error_message();
+    return seastar::make_exception_future<query_result>(std::runtime_error(
+        "Fail to get data directory: " + data_dir.status().error_message()));
+  }
+  auto data_dir_value = data_dir.value();
+
+  // First Stop query_handler's actors.
+
+  auto& hqps_service = HQPSService::get();
+  return hqps_service.stop_query_actors().then([this, graph_name, schema_value,
+                                                data_dir_value, &hqps_service] {
+    LOG(INFO) << "Successfully stopped query handler";
 
-    auto schema_result = server::WorkDirManipulator::GetGraphSchema(graph_name);
-    if (!schema_result.ok()) {
-      LOG(ERROR) << "Fail to get graph schema: "
-                 << schema_result.status().error_message() << ", "
-                 << graph_name;
-      return seastar::make_exception_future<query_result>(std::runtime_error(
-          "Fail to get graph schema: " +
-          schema_result.status().error_message() + ", " + graph_name));
-    }
-    auto data_dir = server::WorkDirManipulator::GetDataDirectory(graph_name);
-    if (!data_dir.ok()) {
-      LOG(ERROR) << "Fail to get data directory: "
-                 << data_dir.status().error_message();
-      return seastar::make_exception_future<query_result>(std::runtime_error(
-          "Fail to get data directory: " + data_dir.status().error_message()));
-    }
     {
       std::lock_guard<std::mutex> lock(mtx_);
       auto& db = gs::GraphDB::get();
       LOG(INFO) << "Update service running on graph:" << graph_name;
-      auto& schema_value = schema_result.value();
+
       // use the previous thread num
       auto thread_num = db.SessionNum();
       db.Close();
-      if (!db.Open(schema_value, data_dir.value(), thread_num).ok()) {
+      if (!db.Open(schema_value, data_dir_value, thread_num).ok()) {
         LOG(ERROR) << "Fail to load graph from data directory: "
-                   << data_dir.value();
+                   << data_dir_value;
         return seastar::make_exception_future<query_result>(std::runtime_error(
-            "Fail to load graph from data directory: " + data_dir.value()));
+            "Fail to load graph from data directory: " + data_dir_value));
       }
       server::WorkDirManipulator::SetRunningGraph(graph_name);
     }
-
+    hqps_service.start_query_actors();  // start on a new scope.
+    LOG(INFO) << "Successfully restart query actors";
     LOG(INFO) << "Successfully started service with graph: " << graph_name;
-
     return seastar::make_ready_future<query_result>(
         "Successfully start service");
-  } catch (std::exception& e) {
-    LOG(ERROR) << "Fail to Start service: ";
-    return seastar::make_exception_future<query_result>(
-        std::runtime_error(e.what()));
-  }
+  });
 }
 
 // get service status
diff --git a/flex/engines/http_server/handler/hqps_http_handler.cc b/flex/engines/http_server/handler/hqps_http_handler.cc
index d4baefa0f667..82102867586f 100644
--- a/flex/engines/http_server/handler/hqps_http_handler.cc
+++ b/flex/engines/http_server/handler/hqps_http_handler.cc
@@ -16,150 +16,266 @@
 #include "flex/engines/http_server/options.h"
 #include "flex/engines/http_server/service/hqps_service.h"
 
-#include <seastar/core/alien.hh>
-#include <seastar/core/print.hh>
-#include <seastar/http/handlers.hh>
-#include "flex/engines/http_server/generated/actor/codegen_actor_ref.act.autogen.h"
-#include "flex/engines/http_server/generated/actor/executor_ref.act.autogen.h"
 #include "flex/engines/http_server/types.h"
 
 namespace server {
 
-class hqps_ic_handler : public seastar::httpd::handler_base {
- public:
-  hqps_ic_handler(uint32_t group_id, uint32_t shard_concurrency)
-      : shard_concurrency_(shard_concurrency), executor_idx_(0) {
-    executor_refs_.reserve(shard_concurrency_);
+hqps_ic_handler::hqps_ic_handler(uint32_t init_group_id, uint32_t max_group_id,
+                                 uint32_t group_inc_step,
+                                 uint32_t shard_concurrency)
+    : cur_group_id_(init_group_id),
+      max_group_id_(max_group_id),
+      group_inc_step_(group_inc_step),
+      shard_concurrency_(shard_concurrency),
+      executor_idx_(0) {
+  executor_refs_.reserve(shard_concurrency_);
+  hiactor::scope_builder builder;
+  builder.set_shard(hiactor::local_shard_id())
+      .enter_sub_scope(hiactor::scope<executor_group>(0))
+      .enter_sub_scope(hiactor::scope<hiactor::actor_group>(cur_group_id_));
+  for (unsigned i = 0; i < shard_concurrency_; ++i) {
+    executor_refs_.emplace_back(builder.build_ref<executor_ref>(i));
+  }
+}
+
+hqps_ic_handler::~hqps_ic_handler() = default;
+
+seastar::future<> hqps_ic_handler::cancel_current_scope() {
+  hiactor::scope_builder builder;
+  builder.set_shard(hiactor::local_shard_id())
+      .enter_sub_scope(hiactor::scope<executor_group>(0))
+      .enter_sub_scope(hiactor::scope<hiactor::actor_group>(cur_group_id_));
+  return hiactor::actor_engine()
+      .cancel_scope_request(builder, false)
+      .then([this] {
+        LOG(INFO) << "Cancel IC scope successfully!";
+        // clear the actor refs
+        executor_refs_.clear();
+        return seastar::make_ready_future<>();
+      });
+}
+
+bool hqps_ic_handler::create_actors() {
+  if (executor_refs_.size() > 0) {
+    LOG(ERROR) << "The actors have been already created!";
+    return false;
+  }
+
+  VLOG(10) << "Create actors with a different sub scope id: " << cur_group_id_;
+  if (cur_group_id_ + group_inc_step_ > max_group_id_) {
+    LOG(ERROR) << "The max group id is reached, cannot create more actors!";
+    return false;
+  }
+  if (cur_group_id_ + group_inc_step_ < cur_group_id_) {
+    LOG(ERROR) << "overflow detected!";
+    return false;
+  }
+  cur_group_id_ += group_inc_step_;
+  hiactor::scope_builder builder;
+  builder.set_shard(hiactor::local_shard_id())
+      .enter_sub_scope(hiactor::scope<executor_group>(0))
+      .enter_sub_scope(hiactor::scope<hiactor::actor_group>(cur_group_id_));
+  for (unsigned i = 0; i < shard_concurrency_; ++i) {
+    executor_refs_.emplace_back(builder.build_ref<executor_ref>(i));
+  }
+  return true;
+}
+
+seastar::future<std::unique_ptr<seastar::httpd::reply>> hqps_ic_handler::handle(
+    const seastar::sstring& path, std::unique_ptr<seastar::httpd::request> req,
+    std::unique_ptr<seastar::httpd::reply> rep) {
+  auto dst_executor = executor_idx_;
+  executor_idx_ = (executor_idx_ + 1) % shard_concurrency_;
+
+  return executor_refs_[dst_executor]
+      .run_hqps_procedure_query(query_param{std::move(req->content)})
+      .then_wrapped(
+          [rep = std::move(rep)](seastar::future<query_result>&& fut) mutable {
+            if (__builtin_expect(fut.failed(), false)) {
+              rep->set_status(
+                  seastar::httpd::reply::status_type::internal_server_error);
+              try {
+                std::rethrow_exception(fut.get_exception());
+              } catch (std::exception& e) {
+                rep->write_body("bin", seastar::sstring(e.what()));
+              }
+              rep->done();
+              return seastar::make_ready_future<
+                  std::unique_ptr<seastar::httpd::reply>>(std::move(rep));
+            }
+            auto result = fut.get0();
+            rep->write_body("bin", std::move(result.content));
+            rep->done();
+            return seastar::make_ready_future<
+                std::unique_ptr<seastar::httpd::reply>>(std::move(rep));
+          });
+}
+
+// a handler for handl adhoc query.
+
+hqps_adhoc_query_handler::hqps_adhoc_query_handler(
+    uint32_t init_adhoc_group_id, uint32_t init_codegen_group_id,
+    uint32_t max_group_id, uint32_t group_inc_step, uint32_t shard_concurrency)
+    : cur_adhoc_group_id_(init_adhoc_group_id),
+      cur_codegen_group_id_(init_codegen_group_id),
+      max_group_id_(max_group_id),
+      group_inc_step_(group_inc_step),
+      shard_concurrency_(shard_concurrency),
+      executor_idx_(0) {
+  executor_refs_.reserve(shard_concurrency_);
+  {
     hiactor::scope_builder builder;
     builder.set_shard(hiactor::local_shard_id())
         .enter_sub_scope(hiactor::scope<executor_group>(0))
-        .enter_sub_scope(hiactor::scope<hiactor::actor_group>(group_id));
+        .enter_sub_scope(
+            hiactor::scope<hiactor::actor_group>(cur_adhoc_group_id_));
     for (unsigned i = 0; i < shard_concurrency_; ++i) {
       executor_refs_.emplace_back(builder.build_ref<executor_ref>(i));
     }
   }
-  ~hqps_ic_handler() override = default;
-
-  seastar::future<std::unique_ptr<seastar::httpd::reply>> handle(
-      const seastar::sstring& path,
-      std::unique_ptr<seastar::httpd::request> req,
-      std::unique_ptr<seastar::httpd::reply> rep) override {
-    auto dst_executor = executor_idx_;
-    executor_idx_ = (executor_idx_ + 1) % shard_concurrency_;
-
-    return executor_refs_[dst_executor]
-        .run_hqps_procedure_query(query_param{std::move(req->content)})
-        .then_wrapped([rep = std::move(rep)](
-                          seastar::future<query_result>&& fut) mutable {
-          if (__builtin_expect(fut.failed(), false)) {
-            rep->set_status(
-                seastar::httpd::reply::status_type::internal_server_error);
-            try {
-              std::rethrow_exception(fut.get_exception());
-            } catch (std::exception& e) {
-              rep->write_body("bin", seastar::sstring(e.what()));
-            }
-            rep->done();
-            return seastar::make_ready_future<
-                std::unique_ptr<seastar::httpd::reply>>(std::move(rep));
-          }
-          auto result = fut.get0();
-          rep->write_body("bin", std::move(result.content));
-          rep->done();
-          return seastar::make_ready_future<
-              std::unique_ptr<seastar::httpd::reply>>(std::move(rep));
-        });
+  {
+    hiactor::scope_builder builder;
+    builder.set_shard(hiactor::local_shard_id())
+        .enter_sub_scope(hiactor::scope<executor_group>(0))
+        .enter_sub_scope(
+            hiactor::scope<hiactor::actor_group>(cur_codegen_group_id_));
+    codegen_actor_refs_.emplace_back(builder.build_ref<codegen_actor_ref>(0));
   }
+}
+hqps_adhoc_query_handler::~hqps_adhoc_query_handler() = default;
+
+seastar::future<> hqps_adhoc_query_handler::cancel_current_scope() {
+  hiactor::scope_builder adhoc_builder;
+  adhoc_builder.set_shard(hiactor::local_shard_id())
+      .enter_sub_scope(hiactor::scope<executor_group>(0))
+      .enter_sub_scope(
+          hiactor::scope<hiactor::actor_group>(cur_adhoc_group_id_));
+  hiactor::scope_builder codegen_builder;
+  codegen_builder.set_shard(hiactor::local_shard_id())
+      .enter_sub_scope(hiactor::scope<executor_group>(0))
+      .enter_sub_scope(
+          hiactor::scope<hiactor::actor_group>(cur_codegen_group_id_));
+  return hiactor::actor_engine()
+      .cancel_scope_request(adhoc_builder, false)
+      .then([this, codegen_builder] {
+        LOG(INFO) << "Cancel adhoc scope successfully!";
+        return hiactor::actor_engine().cancel_scope_request(codegen_builder,
+                                                            false);
+      })
+      .then([this] {
+        LOG(INFO) << "Cancel codegen scope successfully!";
+        // clear the actor refs
+        executor_refs_.clear();
+        codegen_actor_refs_.clear();
+        LOG(INFO) << "Clear actor refs successfully!";
+        return seastar::make_ready_future<>();
+      });
+}
 
- private:
-  const uint32_t shard_concurrency_;
-  uint32_t executor_idx_;
-  std::vector<executor_ref> executor_refs_;
-};
+bool hqps_adhoc_query_handler::create_actors() {
+  if (executor_refs_.size() > 0 || codegen_actor_refs_.size() > 0) {
+    LOG(ERROR) << "The actors have been already created!";
+    return false;
+  }
+  // Check whether cur_adhoc_group_id + group_inc_step_ is larger than
+  // max_group_id_, considering overflow
+  if (cur_adhoc_group_id_ + group_inc_step_ > max_group_id_ ||
+      cur_codegen_group_id_ + group_inc_step_ > max_group_id_) {
+    LOG(ERROR) << "The max group id is reached, cannot create more actors!";
+    return false;
+  }
+  if (cur_adhoc_group_id_ + group_inc_step_ < cur_adhoc_group_id_ ||
+      cur_codegen_group_id_ + group_inc_step_ < cur_codegen_group_id_) {
+    LOG(ERROR) << "overflow detected!";
+    return false;
+  }
 
-// a handler for handl adhoc query.
-class hqps_adhoc_query_handler : public seastar::httpd::handler_base {
- public:
-  hqps_adhoc_query_handler(uint32_t group_id, uint32_t codegen_actor_group_id,
-                           uint32_t shard_concurrency)
-      : shard_concurrency_(shard_concurrency), executor_idx_(0) {
-    executor_refs_.reserve(shard_concurrency_);
-    {
-      hiactor::scope_builder builder;
-      builder.set_shard(hiactor::local_shard_id())
-          .enter_sub_scope(hiactor::scope<executor_group>(0))
-          .enter_sub_scope(hiactor::scope<hiactor::actor_group>(group_id));
-      for (unsigned i = 0; i < shard_concurrency_; ++i) {
-        executor_refs_.emplace_back(builder.build_ref<executor_ref>(i));
-      }
-    }
-    {
-      hiactor::scope_builder builder;
-      builder.set_shard(hiactor::local_shard_id())
-          .enter_sub_scope(hiactor::scope<executor_group>(0))
-          .enter_sub_scope(
-              hiactor::scope<hiactor::actor_group>(codegen_actor_group_id));
-      codegen_actor_ref_ = builder.build_ref<codegen_actor_ref>(0);
+  {
+    cur_adhoc_group_id_ += group_inc_step_;
+    hiactor::scope_builder builder;
+    builder.set_shard(hiactor::local_shard_id())
+        .enter_sub_scope(hiactor::scope<executor_group>(0))
+        .enter_sub_scope(
+            hiactor::scope<hiactor::actor_group>(cur_adhoc_group_id_));
+    for (unsigned i = 0; i < shard_concurrency_; ++i) {
+      executor_refs_.emplace_back(builder.build_ref<executor_ref>(i));
     }
   }
-  ~hqps_adhoc_query_handler() override = default;
-
-  seastar::future<std::unique_ptr<seastar::httpd::reply>> handle(
-      const seastar::sstring& path,
-      std::unique_ptr<seastar::httpd::request> req,
-      std::unique_ptr<seastar::httpd::reply> rep) override {
-    auto dst_executor = executor_idx_;
-    executor_idx_ = (executor_idx_ + 1) % shard_concurrency_;
-
-    return codegen_actor_ref_.do_codegen(query_param{std::move(req->content)})
-        .then([this, dst_executor](auto&& param) {
-          return executor_refs_[dst_executor].run_hqps_adhoc_query(
-              std::move(param));
-        })
-        .then_wrapped([rep = std::move(rep)](
-                          seastar::future<query_result>&& fut) mutable {
-          if (__builtin_expect(fut.failed(), false)) {
-            rep->set_status(
-                seastar::httpd::reply::status_type::internal_server_error);
-            try {
-              std::rethrow_exception(fut.get_exception());
-            } catch (std::exception& e) {
-              rep->write_body("bin", seastar::sstring(e.what()));
+  {
+    cur_codegen_group_id_ += group_inc_step_;
+    hiactor::scope_builder builder;
+    builder.set_shard(hiactor::local_shard_id())
+        .enter_sub_scope(hiactor::scope<executor_group>(0))
+        .enter_sub_scope(
+            hiactor::scope<hiactor::actor_group>(cur_codegen_group_id_));
+    codegen_actor_refs_.emplace_back(builder.build_ref<codegen_actor_ref>(0));
+  }
+  return true;
+}
+
+seastar::future<std::unique_ptr<seastar::httpd::reply>>
+hqps_adhoc_query_handler::handle(const seastar::sstring& path,
+                                 std::unique_ptr<seastar::httpd::request> req,
+                                 std::unique_ptr<seastar::httpd::reply> rep) {
+  auto dst_executor = executor_idx_;
+  executor_idx_ = (executor_idx_ + 1) % shard_concurrency_;
+
+  return codegen_actor_refs_[0]
+      .do_codegen(query_param{std::move(req->content)})
+      .then([this, dst_executor](auto&& param) {
+        return executor_refs_[dst_executor].run_hqps_adhoc_query(
+            std::move(param));
+      })
+      .then_wrapped(
+          [rep = std::move(rep)](seastar::future<query_result>&& fut) mutable {
+            if (__builtin_expect(fut.failed(), false)) {
+              rep->set_status(
+                  seastar::httpd::reply::status_type::internal_server_error);
+              try {
+                std::rethrow_exception(fut.get_exception());
+              } catch (std::exception& e) {
+                rep->write_body("bin", seastar::sstring(e.what()));
+              }
+              rep->done();
+              return seastar::make_ready_future<
+                  std::unique_ptr<seastar::httpd::reply>>(std::move(rep));
             }
+            auto result = fut.get0();
+            rep->write_body("bin", std::move(result.content));
             rep->done();
             return seastar::make_ready_future<
                 std::unique_ptr<seastar::httpd::reply>>(std::move(rep));
-          }
-          auto result = fut.get0();
-          rep->write_body("bin", std::move(result.content));
-          rep->done();
-          return seastar::make_ready_future<
-              std::unique_ptr<seastar::httpd::reply>>(std::move(rep));
-        });
-  }
+          });
+}
 
- private:
-  const uint32_t shard_concurrency_;
-  uint32_t executor_idx_;
-  std::vector<executor_ref> executor_refs_;
-  codegen_actor_ref codegen_actor_ref_;
-};
-
-class hqps_exit_handler : public seastar::httpd::handler_base {
- public:
-  seastar::future<std::unique_ptr<seastar::httpd::reply>> handle(
-      const seastar::sstring& path,
-      std::unique_ptr<seastar::httpd::request> req,
-      std::unique_ptr<seastar::httpd::reply> rep) override {
-    HQPSService::get().set_exit_state();
-    rep->write_body("bin", seastar::sstring{"HQPS service is exiting ..."});
-    return seastar::make_ready_future<std::unique_ptr<seastar::httpd::reply>>(
-        std::move(rep));
-  }
-};
+seastar::future<std::unique_ptr<seastar::httpd::reply>>
+hqps_exit_handler::handle(const seastar::sstring& path,
+                          std::unique_ptr<seastar::httpd::request> req,
+                          std::unique_ptr<seastar::httpd::reply> rep) {
+  HQPSService::get().set_exit_state();
+  rep->write_body("bin", seastar::sstring{"HQPS service is exiting ..."});
+  return seastar::make_ready_future<std::unique_ptr<seastar::httpd::reply>>(
+      std::move(rep));
+}
 
 hqps_http_handler::hqps_http_handler(uint16_t http_port)
-    : http_port_(http_port) {}
+    : http_port_(http_port) {
+  ic_handler_ = new hqps_ic_handler(ic_query_group_id, max_group_id,
+                                    group_inc_step, shard_query_concurrency);
+  adhoc_query_handler_ = new hqps_adhoc_query_handler(
+      ic_adhoc_group_id, codegen_group_id, max_group_id, group_inc_step,
+      shard_adhoc_concurrency);
+  exit_handler_ = new hqps_exit_handler();
+}
+
+hqps_http_handler::~hqps_http_handler() {
+  if (is_running()) {
+    stop();
+  }
+  delete ic_handler_;
+  delete adhoc_query_handler_;
+  delete exit_handler_;
+}
 
 uint16_t hqps_http_handler::get_port() const { return http_port_; }
 
@@ -194,20 +310,36 @@ void hqps_http_handler::stop() {
   running_.store(false);
 }
 
+seastar::future<> hqps_http_handler::stop_query_actors() {
+  // First cancel the scope.
+  return ic_handler_->cancel_current_scope()
+      .then([this] {
+        LOG(INFO) << "Cancel ic scope";
+        return adhoc_query_handler_->cancel_current_scope();
+      })
+      .then([this] {
+        LOG(INFO) << "Cancel adhoc scope";
+        return seastar::make_ready_future<>();
+      });
+}
+
+void hqps_http_handler::start_query_actors() {
+  ic_handler_->create_actors();
+  adhoc_query_handler_->create_actors();
+  LOG(INFO) << "Restart all actors";
+}
+
 seastar::future<> hqps_http_handler::set_routes() {
   return server_.set_routes([this](seastar::httpd::routes& r) {
-    auto procedure_handler =
-        new hqps_ic_handler(ic_query_group_id, shard_query_concurrency);
     r.add(seastar::httpd::operation_type::POST,
-          seastar::httpd::url("/interactive/query"), procedure_handler);
+          seastar::httpd::url("/interactive/query"), ic_handler_);
     r.add(seastar::httpd::operation_type::POST,
-          seastar::httpd::url("/v1/query"), procedure_handler);
+          seastar::httpd::url("/v1/query"), ic_handler_);
     r.add(seastar::httpd::operation_type::POST,
           seastar::httpd::url("/interactive/adhoc_query"),
-          new hqps_adhoc_query_handler(ic_adhoc_group_id, codegen_group_id,
-                                       shard_adhoc_concurrency));
+          adhoc_query_handler_);
     r.add(seastar::httpd::operation_type::POST,
-          seastar::httpd::url("/interactive/exit"), new hqps_exit_handler());
+          seastar::httpd::url("/interactive/exit"), exit_handler_);
     return seastar::make_ready_future<>();
   });
 }
diff --git a/flex/engines/http_server/handler/hqps_http_handler.h b/flex/engines/http_server/handler/hqps_http_handler.h
index dcc0af897693..5fe603507166 100644
--- a/flex/engines/http_server/handler/hqps_http_handler.h
+++ b/flex/engines/http_server/handler/hqps_http_handler.h
@@ -15,13 +15,77 @@
 #ifndef ENGINES_HTTP_SERVER_HANDLER_HQPS_HTTP_HANDLER_H_
 #define ENGINES_HTTP_SERVER_HANDLER_HQPS_HTTP_HANDLER_H_
 
+#include <seastar/core/alien.hh>
+#include <seastar/core/print.hh>
+#include <seastar/http/handlers.hh>
 #include <seastar/http/httpd.hh>
+#include "flex/engines/http_server/generated/actor/codegen_actor_ref.act.autogen.h"
+#include "flex/engines/http_server/generated/actor/executor_ref.act.autogen.h"
 
 namespace server {
 
+class hqps_ic_handler : public seastar::httpd::handler_base {
+ public:
+  hqps_ic_handler(uint32_t init_group_id, uint32_t max_group_id,
+                  uint32_t group_inc_step, uint32_t shard_concurrency);
+  ~hqps_ic_handler() override;
+
+  bool create_actors();
+
+  seastar::future<> cancel_current_scope();
+
+  seastar::future<std::unique_ptr<seastar::httpd::reply>> handle(
+      const seastar::sstring& path,
+      std::unique_ptr<seastar::httpd::request> req,
+      std::unique_ptr<seastar::httpd::reply> rep) override;
+
+ private:
+  uint32_t cur_group_id_;
+  const uint32_t max_group_id_, group_inc_step_;
+  const uint32_t shard_concurrency_;
+  uint32_t executor_idx_;
+  std::vector<executor_ref> executor_refs_;
+};
+
+class hqps_adhoc_query_handler : public seastar::httpd::handler_base {
+ public:
+  hqps_adhoc_query_handler(uint32_t init_adhoc_group_id,
+                           uint32_t init_codegen_group_id,
+                           uint32_t max_group_id, uint32_t group_inc_step,
+                           uint32_t shard_concurrency);
+
+  ~hqps_adhoc_query_handler() override;
+
+  seastar::future<> cancel_current_scope();
+
+  bool create_actors();
+
+  seastar::future<std::unique_ptr<seastar::httpd::reply>> handle(
+      const seastar::sstring& path,
+      std::unique_ptr<seastar::httpd::request> req,
+      std::unique_ptr<seastar::httpd::reply> rep) override;
+
+ private:
+  uint32_t cur_adhoc_group_id_, cur_codegen_group_id_;
+  const uint32_t group_inc_step_, max_group_id_;
+  const uint32_t shard_concurrency_;
+  uint32_t executor_idx_;
+  std::vector<executor_ref> executor_refs_;
+  std::vector<codegen_actor_ref> codegen_actor_refs_;
+};
+
+class hqps_exit_handler : public seastar::httpd::handler_base {
+ public:
+  seastar::future<std::unique_ptr<seastar::httpd::reply>> handle(
+      const seastar::sstring& path,
+      std::unique_ptr<seastar::httpd::request> req,
+      std::unique_ptr<seastar::httpd::reply> rep) override;
+};
+
 class hqps_http_handler {
  public:
   hqps_http_handler(uint16_t http_port);
+  ~hqps_http_handler();
 
   void start();
   void stop();
@@ -30,6 +94,10 @@ class hqps_http_handler {
 
   bool is_running() const;
 
+  seastar::future<> stop_query_actors();
+
+  void start_query_actors();
+
  private:
   seastar::future<> set_routes();
 
@@ -37,6 +105,10 @@ class hqps_http_handler {
   const uint16_t http_port_;
   seastar::httpd::http_server_control server_;
   std::atomic<bool> running_{false};
+
+  hqps_ic_handler* ic_handler_;
+  hqps_adhoc_query_handler* adhoc_query_handler_;
+  hqps_exit_handler* exit_handler_;
 };
 
 }  // namespace server
diff --git a/flex/engines/http_server/options.h b/flex/engines/http_server/options.h
index 819937367415..e6afd886cf24 100644
--- a/flex/engines/http_server/options.h
+++ b/flex/engines/http_server/options.h
@@ -17,15 +17,21 @@
 #define ENGINES_HTTP_SERVER_OPTIONS_H_
 
 #include <cstdint>
+#include <limits>
 
 namespace server {
 
 /// make update executors with higher priority.
-const uint32_t ic_query_group_id = 1;
-const uint32_t ic_update_group_id = 2;
-const uint32_t ic_adhoc_group_id = 3;
-const uint32_t codegen_group_id = 4;
-const uint32_t interactive_admin_group_id = 5;
+const uint32_t interactive_admin_group_id = 1;
+const uint32_t ic_query_group_id = 2;
+const uint32_t ic_update_group_id = 3;
+const uint32_t ic_adhoc_group_id = 4;
+const uint32_t codegen_group_id = 5;
+
+const uint32_t max_group_id = std::numeric_limits<uint32_t>::max();
+const uint32_t group_inc_step =
+    4;  // should equal to number of non-admin groups.
+// Each time we cancel a scope, we will increase the group id by this step.
 
 extern uint32_t shard_query_concurrency;
 extern uint32_t shard_update_concurrency;
diff --git a/flex/engines/http_server/service/hqps_service.cc b/flex/engines/http_server/service/hqps_service.cc
index 1b65ab5d983e..8dd394ab0b72 100644
--- a/flex/engines/http_server/service/hqps_service.cc
+++ b/flex/engines/http_server/service/hqps_service.cc
@@ -110,4 +110,22 @@ void HQPSService::run_and_wait_for_exit() {
 
 void HQPSService::set_exit_state() { running_.store(false); }
 
+seastar::future<> HQPSService::stop_query_actors() {
+  if (query_hdl_) {
+    return query_hdl_->stop_query_actors();
+  } else {
+    std::cerr << "Query handler has not been inited!" << std::endl;
+    return seastar::make_exception_future<>(
+        std::runtime_error("Query handler has not been inited!"));
+  }
+}
+
+void HQPSService::start_query_actors() {
+  if (query_hdl_) {
+    query_hdl_->start_query_actors();
+  } else {
+    std::cerr << "Query handler has not been inited!" << std::endl;
+    return;
+  }
+}
 }  // namespace server
diff --git a/flex/engines/http_server/service/hqps_service.h b/flex/engines/http_server/service/hqps_service.h
index 7a011144e915..1d71f0111fc3 100644
--- a/flex/engines/http_server/service/hqps_service.h
+++ b/flex/engines/http_server/service/hqps_service.h
@@ -53,6 +53,14 @@ class HQPSService {
 
   void set_exit_state();
 
+  // Actually stop the actors, the service is still on, but returns error code
+  // for each request.
+  seastar::future<> stop_query_actors();
+
+  // Actually create new actors with a different scope_id,
+  // Because we don't know whether the previous scope_id can be reused.
+  void start_query_actors();
+
  private:
   HQPSService() = default;
 
diff --git a/flex/engines/http_server/workdir_manipulator.cc b/flex/engines/http_server/workdir_manipulator.cc
index 32813e4d64d0..0c00581fe179 100644
--- a/flex/engines/http_server/workdir_manipulator.cc
+++ b/flex/engines/http_server/workdir_manipulator.cc
@@ -861,7 +861,12 @@ seastar::future<seastar::sstring> WorkDirManipulator::generate_procedure(
     const nlohmann::json& json) {
   LOG(INFO) << "Generate procedure: " << json.dump();
   auto codegen_bin = gs::find_codegen_bin();
-  auto work_directory = std::string(server::CodegenProxy::DEFAULT_CODEGEN_DIR);
+  auto temp_codegen_directory =
+      std::string(server::CodegenProxy::DEFAULT_CODEGEN_DIR);
+  // mkdir -p temp_codegen_directory
+  if (!std::filesystem::exists(temp_codegen_directory)) {
+    std::filesystem::create_directory(temp_codegen_directory);
+  }
   // dump json["query"] to file.
   auto query = json["query"].get<std::string>();
   auto name = json["name"].get<std::string>();
@@ -869,9 +874,9 @@ seastar::future<seastar::sstring> WorkDirManipulator::generate_procedure(
   auto bounded_graph = json["bound_graph"].get<std::string>();
   std::string query_file;
   if (type == "cypher" || type == "CYPHER") {
-    query_file = work_directory + "/" + name + ".cypher";
+    query_file = temp_codegen_directory + "/" + name + ".cypher";
   } else if (type == "CPP" || type == "cpp") {
-    query_file = work_directory + "/" + name + ".cpp";
+    query_file = temp_codegen_directory + "/" + name + ".cpp";
   } else {
     return seastar::make_exception_future<seastar::sstring>(
         "Procedure type is not supported: " + type);
@@ -902,7 +907,7 @@ seastar::future<seastar::sstring> WorkDirManipulator::generate_procedure(
   auto schema_path = GetGraphSchemaPath(bounded_graph);
   auto engine_config = get_engine_config_path();
 
-  return CodegenProxy::CallCodegenCmd(query_file, name, work_directory,
+  return CodegenProxy::CallCodegenCmd(query_file, name, temp_codegen_directory,
                                       output_dir, schema_path, engine_config,
                                       codegen_bin)
       .then_wrapped([name, output_dir](auto&& f) {


Test patch:
diff --git a/flex/tests/hqps/hqps_admin_test.sh b/flex/tests/hqps/hqps_admin_test.sh
index fef925977719..b500819e3984 100644
--- a/flex/tests/hqps/hqps_admin_test.sh
+++ b/flex/tests/hqps/hqps_admin_test.sh
@@ -100,7 +100,7 @@ start_engine_service(){
     info "Start engine service success"
 }
 
-run_movie_test(){
+run_admin_test(){
   echo "run movie test"
   pushd ${FLEX_HOME}/build/
   cmd="GLOG_v=10 ./tests/hqps/admin_http_test ${ADMIN_PORT} ${QUERY_PORT} ${GRAPH_SCHEMA_YAML} ${GRAPH_BULK_LOAD_YAML} ${RAW_CSV_FILES} ${TEST_CYPHER_QUERIES}"
@@ -112,7 +112,7 @@ run_movie_test(){
 
 kill_service
 start_engine_service
-run_movie_test
+run_admin_test
 kill_service