database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-07-29 08:21:15 +03:00
Code Activity

MCOL-5555 Add support for startreadonly command.

Browse Source 
This patch adds support for `startreadonly` command which waits
until all active cpimport jobs are done and then puts controller node to readonly
mode.
This commit is contained in:
Denis Khalikov
2023-10-02 23:46:53 +03:00
committed by Leonid Fedorov
parent 99542e90ca
commit 441cd9d34f
11 changed files with 403 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
writeengine/bulk/cpimport.cpp
View File

@ -59,6 +59,7 @@ namespace
char* pgmName = 0;
const std::string IMPORT_PATH_CWD(".");
bool bDebug = false;
uint32_t cpimportJobId = 0;
//@bug 4643: cpimport job ended during setup w/o any err msg.
// Added a try/catch with logging to main() in case
@ -192,6 +193,7 @@ void printUsage()
//------------------------------------------------------------------------------
void handleSigTerm(int i)
{
BRMWrapper::getInstance()->finishCpimportJob(cpimportJobId);
std::cout << "Received SIGTERM to terminate the process..." << std::endl;
BulkStatus::setJobStatus(EXIT_FAILURE);
}
@ -201,12 +203,32 @@ void handleSigTerm(int i)
//------------------------------------------------------------------------------
void handleControlC(int i)
{
BRMWrapper::getInstance()->finishCpimportJob(cpimportJobId);
if (!BulkLoad::disableConsoleOutput())
std::cout << "Received Control-C to terminate the process..." << std::endl;
BulkStatus::setJobStatus(EXIT_FAILURE);
}
//------------------------------------------------------------------------------
// Signal handler to catch SIGTERM signal to terminate the process
//------------------------------------------------------------------------------
void handleSigSegv(int i)
{
BRMWrapper::getInstance()->finishCpimportJob(cpimportJobId);
std::cout << "Received SIGSEGV to terminate the process..." << std::endl;
BulkStatus::setJobStatus(EXIT_FAILURE);
}
//------------------------------------------------------------------------------
// Signal handler to catch SIGTERM signal to terminate the process
//------------------------------------------------------------------------------
void handleSigAbrt(int i)
{
BRMWrapper::getInstance()->finishCpimportJob(cpimportJobId);
std::cout << "Received SIGABRT to terminate the process..." << std::endl;
BulkStatus::setJobStatus(EXIT_FAILURE);
}
//------------------------------------------------------------------------------
// If error occurs during startup, this function is called to log the specified
@ -214,6 +236,7 @@ void handleControlC(int i)
//------------------------------------------------------------------------------
void startupError(const std::string& errMsg, bool showHint)
{
BRMWrapper::getInstance()->finishCpimportJob(cpimportJobId);
// Log to console
if (!BulkLoad::disableConsoleOutput())
cerr << errMsg << endl;
@ -275,6 +298,17 @@ void setupSignalHandlers()
memset(&act, 0, sizeof(act));
act.sa_handler = handleSigTerm;
sigaction(SIGTERM, &act, 0);
// catch SIGSEGV signal to terminate the program
memset(&act, 0, sizeof(act));
act.sa_handler = handleSigSegv;
sigaction(SIGSEGV, &act, 0);
// catch SIGABRT signal to terminate the program
memset(&act, 0, sizeof(act));
act.sa_handler = handleSigAbrt;
sigaction(SIGABRT, &act, 0);
}
//------------------------------------------------------------------------------
@ -1270,6 +1304,16 @@ int main(int argc, char** argv)
new boost::thread(utils::MonitorProcMem(0, checkPct, SUBSYSTEM_ID_WE_BULK));
}
rc = BRMWrapper::getInstance()->newCpimportJob(cpimportJobId);
if (rc != NO_ERROR)
{
WErrorCodes ec;
std::ostringstream oss;
oss << "Error in creating new cpimport job on Controller node; " << ec.errorString(rc)
<< "; cpimport is terminating.";
startupError(oss.str(), false);
}
//--------------------------------------------------------------------------
// This is the real business
//--------------------------------------------------------------------------
@ -1320,6 +1364,7 @@ int main(int argc, char** argv)
rc = ERR_UNKNOWN;
}
BRMWrapper::getInstance()->finishCpimportJob(cpimportJobId);
// Free up resources allocated by MY_INIT() above.
my_end(0);