Add download stop API for download service

engine/commands/cortex_upd_cmd.cc

@@ -2,7 +2,6 @@
 #include "httplib.h"
 #include "nlohmann/json.hpp"
 #include "server_stop_cmd.h"
-#include "services/download_service.h"
 #include "utils/archive_utils.h"
 #include "utils/file_manager_utils.h"
 #include "utils/logging_utils.h"
@@ -395,7 +394,7 @@ bool CortexUpdCmd::HandleGithubRelease(const nlohmann::json& assets,
                                           .localPath = local_path,
                                       }}}};
 
-      auto result = DownloadService().AddDownloadTask(
+      auto result = download_service_->AddDownloadTask(
           download_task, [](const DownloadTask& finishedTask) {
             // try to unzip the downloaded file
             CTL_INF("Downloaded engine path: "
@@ -460,7 +459,7 @@ bool CortexUpdCmd::GetNightly(const std::string& v) {
                        .localPath = localPath,
                    }}};
 
-  auto result = DownloadService().AddDownloadTask(
+  auto result = download_service_->AddDownloadTask(
       download_task, [](const DownloadTask& finishedTask) {
         // try to unzip the downloaded file
         CTL_INF("Downloaded engine path: "

engine/commands/cortex_upd_cmd.h

@@ -5,10 +5,7 @@
 #include <unistd.h>
 #endif
 
-#include "httplib.h"
-#include "nlohmann/json.hpp"
 #include "utils/file_manager_utils.h"
-#include "utils/logging_utils.h"
 
 namespace commands {
 #ifndef CORTEX_VARIANT
@@ -81,9 +78,14 @@ bool ReplaceBinaryInflight(const std::filesystem::path& src,
 // - Nightly: Enables retrieval of the latest nightly build and specific versions using the -v flag
 class CortexUpdCmd {
  public:
+  explicit CortexUpdCmd(std::shared_ptr<DownloadService> download_service)
+      : download_service_{download_service} {};
+
   void Exec(const std::string& v);
 
  private:
+  std::shared_ptr<DownloadService> download_service_;
+
   bool GetStable(const std::string& v);
   bool GetBeta(const std::string& v);
   bool HandleGithubRelease(const nlohmann::json& assets,

engine/commands/server_stop_cmd.cc

@@ -1,6 +1,5 @@
 #include "server_stop_cmd.h"
 #include "httplib.h"
-#include "trantor/utils/Logger.h"
 #include "utils/logging_utils.h"
 
 namespace commands {
@@ -18,4 +17,4 @@ void ServerStopCmd::Exec() {
   }
 }
 
-};  // namespace commands
+};  // namespace commands

engine/commands/server_stop_cmd.h

@@ -1,9 +1,10 @@
 #pragma once
+
 #include <string>
 
 namespace commands {
 
-class ServerStopCmd{
+class ServerStopCmd {
  public:
   ServerStopCmd(std::string host, int port);
   void Exec();
@@ -12,4 +13,4 @@ class ServerStopCmd{
   std::string host_;
   int port_;
 };
-}  // namespace commands
+}  // namespace commands

engine/common/download_task.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <json/json.h>
 #include <filesystem>
 #include <nlohmann/json.hpp>
 #include <sstream>
@@ -70,6 +71,28 @@ struct DownloadTask {
     return output.str();
   }
 
+  Json::Value ToJsonCpp() const {
+    Json::Value root;
+    root["id"] = id;
+    root["type"] = DownloadTypeToString(type);
+
+    Json::Value itemsArray(Json::arrayValue);
+    for (const auto& item : items) {
+      Json::Value itemObj;
+      itemObj["id"] = item.id;
+      itemObj["downloadUrl"] = item.downloadUrl;
+      itemObj["localPath"] = item.localPath.string();
+      itemObj["checksum"] = item.checksum.value_or("N/A");
+      itemObj["bytes"] = Json::Value::UInt64(item.bytes.value_or(0));
+      itemObj["downloadedBytes"] =
+          Json::Value::UInt64(item.downloadedBytes.value_or(0));
+      itemsArray.append(itemObj);
+    }
+    root["items"] = itemsArray;
+
+    return root;
+  }
+
   json ToJson() const {
     json dl_items = json::array();
 

engine/common/event.h

@@ -34,7 +34,7 @@ std::string DownloadEventTypeToString(DownloadEventType type) {
     case DownloadEventType::DownloadStarted:
       return "DownloadStarted";
     case DownloadEventType::DownloadStopped:
-      return "DownloadPaused";
+      return "DownloadStopped";
     case DownloadEventType::DownloadUpdated:
       return "DownloadUpdated";
     case DownloadEventType::DownloadSuccess:

engine/config/gguf_parser.cc

@@ -588,7 +588,8 @@ void GGUFHandler::ModelConfigFromMetadata() {
   model_config_.model = name;
   model_config_.id = name;
   model_config_.version = std::to_string(version);
-  model_config_.max_tokens = std::min<int>(kDefaultMaxContextLength, max_tokens);
+  model_config_.max_tokens =
+      std::min<int>(kDefaultMaxContextLength, max_tokens);
   model_config_.ctx_len = std::min<int>(kDefaultMaxContextLength, max_tokens);
   model_config_.ngl = ngl;
 }

engine/controllers/command_line_parser.cc

@@ -32,10 +32,12 @@ constexpr const auto kEngineGroup = "Engines";
 constexpr const auto kSystemGroup = "System";
 constexpr const auto kSubcommands = "Subcommands";
 }  // namespace
+
 CommandLineParser::CommandLineParser()
     : app_("Cortex.cpp CLI"),
-      model_service_{ModelService(std::make_shared<DownloadService>())},
-      engine_service_{EngineService(std::make_shared<DownloadService>())} {}
+      download_service_{std::make_shared<DownloadService>()},
+      model_service_{ModelService(download_service_)},
+      engine_service_{EngineService(download_service_)} {}
 
 bool CommandLineParser::SetupCommand(int argc, char** argv) {
   app_.usage("Usage:\n" + commands::GetCortexBinary() +
@@ -452,7 +454,7 @@ void CommandLineParser::SetupSystemCommands() {
       return;
     }
 #endif
-    commands::CortexUpdCmd cuc;
+    auto cuc = commands::CortexUpdCmd(download_service_);
     cuc.Exec(cml_data_.cortex_version);
     cml_data_.check_upd = false;
   });

engine/controllers/command_line_parser.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <memory>
 #include "CLI/CLI.hpp"
 #include "services/engine_service.h"
 #include "services/model_service.h"
@@ -29,6 +30,7 @@ class CommandLineParser {
   void ModelUpdate(CLI::App* parent);
 
   CLI::App app_;
+  std::shared_ptr<DownloadService> download_service_;
   EngineService engine_service_;
   ModelService model_service_;
 

engine/controllers/models.cc

@@ -27,14 +27,14 @@ void Models::PullModel(const HttpRequestPtr& req,
   }
 
   auto handle_model_input =
-      [&, model_handle]() -> cpp::result<std::string, std::string> {
+      [&, model_handle]() -> cpp::result<DownloadTask, std::string> {
     CTL_INF("Handle model input, model handle: " + model_handle);
     if (string_utils::StartsWith(model_handle, "https")) {
-      return model_service_->HandleUrl(model_handle, true);
+      return model_service_->HandleDownloadUrlAsync(model_handle);
     } else if (model_handle.find(":") != std::string::npos) {
       auto model_and_branch = string_utils::SplitBy(model_handle, ":");
-      return model_service_->DownloadModelFromCortexso(
-          model_and_branch[0], model_and_branch[1], true);
+      return model_service_->DownloadModelFromCortexsoAsync(
+          model_and_branch[0], model_and_branch[1]);
     }
 
     return cpp::fail("Invalid model handle or not supported!");
@@ -50,6 +50,39 @@ void Models::PullModel(const HttpRequestPtr& req,
   } else {
     Json::Value ret;
     ret["message"] = "Model start downloading!";
+    ret["task"] = result.value().ToJsonCpp();
+    auto resp = cortex_utils::CreateCortexHttpJsonResponse(ret);
+    resp->setStatusCode(k200OK);
+    callback(resp);
+  }
+}
+
+void Models::AbortPullModel(
+    const HttpRequestPtr& req,
+    std::function<void(const HttpResponsePtr&)>&& callback) {
+  if (!http_util::HasFieldInReq(req, callback, "taskId")) {
+    return;
+  }
+  auto task_id = (*(req->getJsonObject())).get("taskId", "").asString();
+  if (task_id.empty()) {
+    Json::Value ret;
+    ret["result"] = "Bad Request";
+    auto resp = cortex_utils::CreateCortexHttpJsonResponse(ret);
+    resp->setStatusCode(k400BadRequest);
+    callback(resp);
+    return;
+  }
+
+  auto result = model_service_->AbortDownloadModel(task_id);
+  if (result.has_error()) {
+    Json::Value ret;
+    ret["message"] = result.error();
+    auto resp = cortex_utils::CreateCortexHttpJsonResponse(ret);
+    resp->setStatusCode(k400BadRequest);
+    callback(resp);
+  } else {
+    Json::Value ret;
+    ret["message"] = "Task stopped!";
     auto resp = cortex_utils::CreateCortexHttpJsonResponse(ret);
     resp->setStatusCode(k200OK);
     callback(resp);

engine/controllers/models.h

@@ -10,6 +10,7 @@ class Models : public drogon::HttpController<Models, false> {
  public:
   METHOD_LIST_BEGIN
   METHOD_ADD(Models::PullModel, "/pull", Post);
+  METHOD_ADD(Models::AbortPullModel, "/pull", Delete);
   METHOD_ADD(Models::ListModel, "", Get);
   METHOD_ADD(Models::GetModel, "/{1}", Get);
   METHOD_ADD(Models::UpdateModel, "/{1}", Patch);
@@ -20,6 +21,7 @@ class Models : public drogon::HttpController<Models, false> {
   METHOD_ADD(Models::StopModel, "/stop", Post);
 
   ADD_METHOD_TO(Models::PullModel, "/v1/models/pull", Post);
+  ADD_METHOD_TO(Models::PullModel, "/v1/models/pull", Delete);
   ADD_METHOD_TO(Models::ListModel, "/v1/models", Get);
   ADD_METHOD_TO(Models::GetModel, "/v1/models/{1}", Get);
   ADD_METHOD_TO(Models::UpdateModel, "/v1/models/{1}", Patch);
@@ -35,6 +37,8 @@ class Models : public drogon::HttpController<Models, false> {
 
   void PullModel(const HttpRequestPtr& req,
                  std::function<void(const HttpResponsePtr&)>&& callback);
+  void AbortPullModel(const HttpRequestPtr& req,
+                      std::function<void(const HttpResponsePtr&)>&& callback);
   void ListModel(const HttpRequestPtr& req,
                  std::function<void(const HttpResponsePtr&)>&& callback) const;
   void GetModel(const HttpRequestPtr& req,

engine/controllers/processManager.cc → engine/controllers/process_manager.cc

@@ -1,12 +1,21 @@
-#include "processManager.h"
+#include "process_manager.h"
 #include "utils/cortex_utils.h"
+#include "utils/logging_utils.h"
 
 #include <trantor/utils/Logger.h>
 #include <cstdlib>
 
-void processManager::destroy(
+void ProcessManager::destroy(
     const HttpRequestPtr& req,
     std::function<void(const HttpResponsePtr&)>&& callback) {
+
+  auto destroy_result = download_service_->Destroy();
+  if (destroy_result.has_error()) {
+    CTL_ERR("Failed to destroy download service: " + destroy_result.error());
+  } else {
+    CTL_INF("Download service stopped!");
+  }
+
   app().quit();
   Json::Value ret;
   ret["message"] = "Program is exitting, goodbye!";

engine/controllers/process_manager.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include <drogon/HttpController.h>
+#include <drogon/HttpTypes.h>
+#include "services/download_service.h"
+
+using namespace drogon;
+
+class ProcessManager : public drogon::HttpController<ProcessManager, false> {
+ public:
+  METHOD_LIST_BEGIN
+  METHOD_ADD(ProcessManager::destroy, "/destroy", Delete);
+  METHOD_LIST_END
+
+  explicit ProcessManager(std::shared_ptr<DownloadService> download_service)
+      : download_service_{download_service} {}
+
+  void destroy(const HttpRequestPtr& req,
+               std::function<void(const HttpResponsePtr&)>&& callback);
+
+ private:
+  std::shared_ptr<DownloadService> download_service_;
+};

engine/main.cc

@@ -5,6 +5,7 @@
 #include "controllers/engines.h"
 #include "controllers/events.h"
 #include "controllers/models.h"
+#include "controllers/process_manager.h"
 #include "cortex-common/cortexpythoni.h"
 #include "services/model_service.h"
 #include "utils/archive_utils.h"
@@ -92,10 +93,12 @@ void RunServer() {
   auto engine_ctl = std::make_shared<Engines>(engine_service);
   auto model_ctl = std::make_shared<Models>(model_service);
   auto event_ctl = std::make_shared<Events>(event_queue_ptr);
+  auto pm_ctl = std::make_shared<ProcessManager>(download_service);
 
   drogon::app().registerController(engine_ctl);
   drogon::app().registerController(model_ctl);
   drogon::app().registerController(event_ctl);
+  drogon::app().registerController(pm_ctl);
 
   LOG_INFO << "Server started, listening at: " << config.apiServerHost << ":"
            << config.apiServerPort;