fix

Author: ywy2090

cpp/wedpr-computing/ppc-mpc/src/Common.h

@@ -58,6 +58,7 @@ struct JobStatus
 const int MPC_SUCCESS = 0;
 const int MPC_RUNNING = 1;
 const int MPC_DUPLICATED = 2;
+const int MPC_KILLED = 3;
 const int MPC_FAILED = -1;
 
 const std::string PATH_SEPARATOR = "/";

cpp/wedpr-computing/ppc-mpc/src/MPCService.cpp

@@ -47,16 +47,25 @@ using namespace ppc::tools;
 using namespace ppc::storage;
 using namespace ppc::rpc;
 
-bool MPCService::addJobIfNotExist(const JobInfo& jobInfo)
+bool MPCService::addJobIfNotRunning(const JobInfo& jobInfo)
 {
     std::lock_guard<std::mutex> l(x_job2Status);
     auto it = m_job2Status.find(jobInfo.jobId);
     if (it != m_job2Status.end())
     {
         MPC_LOG(INFO) << LOG_DESC("[MPCService][addJob]") 
                 << "job already exists"
-                << LOG_KV("jobId", jobInfo.jobId);
-        return false;
+                << LOG_KV("jobId", jobInfo.jobId)
+                << LOG_KV("code", it->code)
+                << LOG_KV("message", it->message);
+
+        if (it->second.code == MPC_RUNNING) 
+        {
+            // job is running
+            return false;
+        }
+       
+        m_job2Status.erase(it);
     }
 
     JobStatus jobStatus;
@@ -145,6 +154,31 @@ void MPCService::onFailed(const std::string &jobId, const std::string &msg)
                 ;
 }
 
+
+void MPCService::onKill(const std::string &jobId, const std::string &msg)
+{
+    std::lock_guard<std::mutex> l(x_job2Status);
+    auto it = m_job2Status.find(jobId);
+    if (it == m_job2Status.end())
+    {
+        MPC_LOG(ERROR) << LOG_DESC("[MPCService][onKill]") 
+                << "job does not exist"
+                << LOG_KV("jobId", jobId);
+        return;
+    }
+
+    JobStatus &jobStatus = it->second;
+    jobStatus.code = MPC_KILLED;
+    jobStatus.message = msg;
+    jobStatus.timeCostMs = utcSteadyTime() - jobStatus.startTimeMs;
+
+    MPC_LOG(INFO) << LOG_DESC("[MPCService][onKill]") 
+                << "job is killed"
+                << LOG_KV("jobId", jobId)
+                << LOG_KV("timeCostMs", jobStatus.timeCostMs)
+                ;
+}
+
 void MPCService::doRun(const JobInfo& jobInfo) 
 {
     auto startT = utcSteadyTime();
@@ -263,7 +297,7 @@ void MPCService::runMpcRpc(Json::Value const& request, RespFunc func)
     try
     {
         auto jobInfo = paramsToJobInfo(request);
-        auto r = addJobIfNotExist(jobInfo);
+        auto r = addJobIfNotRunning(jobInfo);
 
         if (r)
         {  
@@ -301,7 +335,7 @@ void MPCService::asyncRunMpcRpc(Json::Value const& request, RespFunc func)
     try
     {
         auto jobInfo = paramsToJobInfo(request);
-        auto r = addJobIfNotExist(jobInfo);
+        auto r = addJobIfNotRunning(jobInfo);
 
         if (r)
         {  
@@ -362,6 +396,7 @@ void MPCService::doKill(Json::Value const& request, Json::Value& response)
         if (outExitStatus == 0)
         {
             message = "Kill mpc job successfully";
+            onKill(jobId, message);
             MPC_LOG(INFO) << LOG_DESC("[MPCService][doKill]") << LOG_DESC(message);
             response["code"] = MPC_SUCCESS;
             response["message"] = "success";

cpp/wedpr-computing/ppc-mpc/src/MPCService.h

@@ -59,6 +59,7 @@ class MPCService: std::enable_shared_from_this<MPCService>
 
     void onSuccess(const std::string &jobId, const std::string &msg);
     void onFailed(const std::string &jobId, const std::string &msg);
+    void onKill(const std::string &jobId, const std::string &msg);
 
     void execCommand(const std::string cmd, int& outExitStatus, std::string& outResult);
 
@@ -74,7 +75,7 @@ class MPCService: std::enable_shared_from_this<MPCService>
         m_threadPool = threadPool;
     }
 
-    bool addJobIfNotExist(const JobInfo& jobInfo);
+    bool addJobIfNotRunning(const JobInfo& jobInfo);
 
     bool queryJobStatus(const std::string &jobId, JobStatus &jobStatus);