MCOL-5555 Add support for startreadonly command.

This patch adds support for `startreadonly` command which waits
until all active cpimport jobs are done and then puts controller node to readonly
mode.

versioning/BRM/masterdbrmnode.cpp

@@ -24,6 +24,7 @@
 #include <unistd.h>
 #include <sys/types.h>
 #include <sstream>
+#include <thread>
 
 #include "sessionmanager.h"
 #include "socketclosed.h"
@@ -477,6 +478,17 @@ void MasterDBRMNode::msgProcessor()
       case GET_UNCOMMITTED_LBIDS: doGetUncommittedLbids(msg, p); continue;
     }
 
+    switch (cmd)
+    {
+      case NEW_CPIMPORT_JOB: doNewCpimportJob(p); continue;
+
+      case FINISH_CPIMPORT_JOB: doFinishCpimportJob(msg, p); continue;
+
+      case START_READONLY: doStartReadOnly(p->sock); continue;
+
+      case FORCE_CLEAR_CPIMPORT_JOBS: doForceClearAllCpimportJobs(p->sock); continue;
+    }
+
     /* Process TableLock calls */
     switch (cmd)
     {
@@ -1149,11 +1161,66 @@ void MasterDBRMNode::doResume(messageqcpp::IOSocket* sock)
   {
   }
 }
+void MasterDBRMNode::doForceClearAllCpimportJobs(messageqcpp::IOSocket* sock)
+{
+#ifdef BMR_VERBOSE
+  std::cout << "doForceClearCpimprtJobs" << std::endl;
+#endif
+  ByteStream reply;
+  std::unique_lock lk(cpimportMutex);
+  sm.clearAllCpimportJobs();
+
+  reply << (uint8_t)ERR_OK;
+  try
+  {
+    sock->write(reply);
+  }
+  catch (exception&)
+  {
+  }
+}
+
+void MasterDBRMNode::doStartReadOnly(messageqcpp::IOSocket* sock)
+{
+#ifdef BRM_VERBOSE
+  std::cout << "doStartReadonly" << std::endl;
+#endif
+  ByteStream reply;
+
+  // Spawn a new thread and detach it, we cannot block `msgProcessor`.
+  std::thread readonlyAdjuster(
+      [this]()
+      {
+        std::unique_lock lk(cpimportMutex);
+        if (sm.getCpimportJobsCount() != 0)
+        {
+          waitToFinishJobs = true;
+          // Wait until all cpimort jobs are done.
+          cpimportJobsCond.wait(lk, [this]() { return sm.getCpimportJobsCount() == 0; });
+          setReadOnly(true);
+          waitToFinishJobs = false;
+        }
+        else
+        {
+          setReadOnly(true);
+        }
+      });
+
+  readonlyAdjuster.detach();
+
+  reply << (uint8_t)ERR_OK;
+  try
+  {
+    sock->write(reply);
+  }
+  catch (exception&)
+  {
+  }
+}
 
 void MasterDBRMNode::doSetReadOnly(messageqcpp::IOSocket* sock, bool b)
 {
   ByteStream reply;
-
   setReadOnly(b);
   reply << (uint8_t)ERR_OK;
 
@@ -1358,6 +1425,103 @@ void MasterDBRMNode::doSysCatVerID(ByteStream& msg, ThreadParams* p)
   }
 }
 
+void MasterDBRMNode::doNewCpimportJob(ThreadParams* p)
+{
+#ifdef BRM_VERBOSE
+  std::cerr << "doNewCpimportJob" << std::endl;
+#endif
+  ByteStream reply;
+  uint32_t jobId;
+
+  std::unique_lock lk(cpimportMutex);
+  // That means that controller node is waiting untill all active cpimport jobs are done.
+  // We cannot block `msgProcessor` and cannot create a new job, so exit with `readonly` code.
+  if (waitToFinishJobs)
+  {
+    reply << (uint8_t)ERR_READONLY;
+    try
+    {
+      p->sock->write(reply);
+    }
+    catch (...)
+    {
+    }
+    return;
+  }
+
+  try
+  {
+    jobId = sm.newCpimportJob();
+  }
+  catch (exception&)
+  {
+    reply.reset();
+    reply << (uint8_t)ERR_FAILURE;
+    try
+    {
+      p->sock->write(reply);
+    }
+    catch (...)
+    {
+    }
+
+    return;
+  }
+
+  reply << (uint8_t)ERR_OK;
+  reply << (uint32_t)jobId;
+  try
+  {
+    p->sock->write(reply);
+  }
+  catch (exception&)
+  {
+  }
+}
+
+void MasterDBRMNode::doFinishCpimportJob(ByteStream& msg, ThreadParams* p)
+{
+#ifdef BRM_VERBOSE
+  std::cout << "doFinishCpimportJob" << std::endl;
+#endif
+  ByteStream reply;
+  uint32_t cpimportJob;
+  uint8_t cmd;
+  std::unique_lock lk(cpimportMutex);
+
+  msg >> cmd;
+  msg >> cpimportJob;
+  try
+  {
+    sm.finishCpimortJob(cpimportJob);
+  }
+  catch (exception&)
+  {
+    reply << (uint8_t)ERR_FAILURE;
+    try
+    {
+      p->sock->write(reply);
+    }
+    catch (...)
+    {
+    }
+
+    return;
+  }
+
+  reply << (uint8_t)ERR_OK;
+  try
+  {
+    p->sock->write(reply);
+  }
+  catch (...)
+  {
+  }
+
+  if (waitToFinishJobs && sm.getCpimportJobsCount() == 0)
+    cpimportJobsCond.notify_one();
+}
+
 void MasterDBRMNode::doNewTxnID(ByteStream& msg, ThreadParams* p)
 {
   ByteStream reply;