From afa5a6a7bf3f046d4ec1356650804d325742574e Mon Sep 17 00:00:00 2001 From: "tomas@poseidon.ndb.mysql.com" <> Date: Mon, 13 Dec 2004 00:48:06 +0100 Subject: [PATCH] fix bug#7193 added start backup parameter to enable setting wait state and set different timeouts depending on waitstate moved listen of backup event out of backup and into separete thread thread created at connect() and destroyed at disconnect() added start backup command options "nowait" "wait completed" "wait started" fixed log level settings --- ndb/include/mgmapi/mgmapi.h | 6 +- ndb/src/common/debugger/EventLogger.cpp | 8 +- ndb/src/mgmapi/mgmapi.cpp | 14 +- ndb/src/mgmclient/CommandInterpreter.cpp | 199 +++++++++++++++++++---- ndb/src/mgmsrv/MgmtSrvr.cpp | 28 +++- ndb/src/mgmsrv/MgmtSrvr.hpp | 6 +- ndb/src/mgmsrv/Services.cpp | 138 ++++++---------- ndb/test/src/NdbBackup.cpp | 3 +- 8 files changed, 263 insertions(+), 139 deletions(-) diff --git a/ndb/include/mgmapi/mgmapi.h b/ndb/include/mgmapi/mgmapi.h index 5329ded4f1..fd4baea3e1 100644 --- a/ndb/include/mgmapi/mgmapi.h +++ b/ndb/include/mgmapi/mgmapi.h @@ -666,11 +666,15 @@ extern "C" { * Start backup * * @param handle NDB management handle. + * @param wait_completed 0=don't wait for confirmation + 1=wait for backup started + 2=wait for backup completed * @param backup_id Backup id is returned from function. * @param reply Reply message. * @return -1 on error. */ - int ndb_mgm_start_backup(NdbMgmHandle handle, unsigned int* backup_id, + int ndb_mgm_start_backup(NdbMgmHandle handle, int wait_completed, + unsigned int* backup_id, struct ndb_mgm_reply* reply); /** diff --git a/ndb/src/common/debugger/EventLogger.cpp b/ndb/src/common/debugger/EventLogger.cpp index 8bb797b785..877f5dee24 100644 --- a/ndb/src/common/debugger/EventLogger.cpp +++ b/ndb/src/common/debugger/EventLogger.cpp @@ -1280,10 +1280,10 @@ EventLogger::getText(char * m_text, size_t m_text_len, case EventReport::BackupCompleted: BaseString::snprintf(m_text, m_text_len, - "%sBackup %d started from node %d completed\n" - " StartGCP: %d StopGCP: %d\n" - " #Records: %d #LogRecords: %d\n" - " Data: %d bytes Log: %d bytes", + "%sBackup %u started from node %u completed\n" + " StartGCP: %u StopGCP: %u\n" + " #Records: %u #LogRecords: %u\n" + " Data: %u bytes Log: %u bytes", theNodeId, theData[2], refToNode(theData[1]), theData[3], theData[4], theData[6], theData[8], theData[5], theData[7]); diff --git a/ndb/src/mgmapi/mgmapi.cpp b/ndb/src/mgmapi/mgmapi.cpp index 1a4a10f942..adf0d26d63 100644 --- a/ndb/src/mgmapi/mgmapi.cpp +++ b/ndb/src/mgmapi/mgmapi.cpp @@ -1545,7 +1545,8 @@ ndb_mgm_start(NdbMgmHandle handle, int no_of_nodes, const int * node_list) *****************************************************************************/ extern "C" int -ndb_mgm_start_backup(NdbMgmHandle handle, unsigned int* _backup_id, +ndb_mgm_start_backup(NdbMgmHandle handle, int wait_completed, + unsigned int* _backup_id, struct ndb_mgm_reply* /*reply*/) { SET_ERROR(handle, NDB_MGM_NO_ERROR, "Executing: ndb_mgm_start_backup"); @@ -1559,8 +1560,17 @@ ndb_mgm_start_backup(NdbMgmHandle handle, unsigned int* _backup_id, CHECK_CONNECTED(handle, -1); Properties args; + args.put("completed", wait_completed); const Properties *reply; - reply = ndb_mgm_call(handle, start_backup_reply, "start backup", &args); + { // start backup can take some time, set timeout high + Uint64 old_timeout= handle->read_timeout; + if (wait_completed == 2) + handle->read_timeout= 30*60*1000; // 30 minutes + else if (wait_completed == 1) + handle->read_timeout= 5*60*1000; // 5 minutes + reply = ndb_mgm_call(handle, start_backup_reply, "start backup", &args); + handle->read_timeout= old_timeout; + } CHECK_REPLY(reply, -1); BaseString result; diff --git a/ndb/src/mgmclient/CommandInterpreter.cpp b/ndb/src/mgmclient/CommandInterpreter.cpp index e4c66d0462..4855b716d9 100644 --- a/ndb/src/mgmclient/CommandInterpreter.cpp +++ b/ndb/src/mgmclient/CommandInterpreter.cpp @@ -154,7 +154,8 @@ private: const char * param); NdbMgmHandle m_mgmsrv; - bool connected; + NdbMgmHandle m_mgmsrv2; + bool m_connected; int m_verbose; int try_reconnect; int m_error; @@ -163,6 +164,7 @@ private: const char *rep_host; bool rep_connected; #endif + struct NdbThread* m_event_thread; }; @@ -261,7 +263,8 @@ static const char* helpText = "SHOW CONFIG Print configuration\n" "SHOW PARAMETERS Print configuration parameters\n" #endif -"START BACKUP Start backup\n" +"START BACKUP [NOWAIT | WAIT STARTED | WAIT COMPLETED]\n" +" Start backup (default WAIT COMPLETED)\n" "ABORT BACKUP <backup id> Abort backup\n" "SHUTDOWN Shutdown all processes in cluster and quit\n" "CLUSTERLOG ON [<severity>] ... Enable Cluster logging\n" @@ -386,13 +389,19 @@ CommandInterpreter::CommandInterpreter(const char *_host,int verbose) ndbout_c("Cannot create handle to management server."); exit(-1); } + m_mgmsrv2 = ndb_mgm_create_handle(); + if(m_mgmsrv2 == NULL) { + ndbout_c("Cannot create handle to management server."); + exit(-1); + } if (ndb_mgm_set_connectstring(m_mgmsrv, _host)) { printError(); exit(-1); } - connected = false; + m_connected= false; + m_event_thread= 0; try_reconnect = 0; #ifdef HAVE_GLOBAL_REPLICATION rep_host = NULL; @@ -406,8 +415,9 @@ CommandInterpreter::CommandInterpreter(const char *_host,int verbose) */ CommandInterpreter::~CommandInterpreter() { - connected = false; + disconnect(); ndb_mgm_destroy_handle(&m_mgmsrv); + ndb_mgm_destroy_handle(&m_mgmsrv2); } static bool @@ -430,7 +440,10 @@ void CommandInterpreter::printError() { if (ndb_mgm_check_connection(m_mgmsrv)) - connected= false; + { + m_connected= false; + disconnect(); + } ndbout_c("* %5d: %s", ndb_mgm_get_latest_error(m_mgmsrv), ndb_mgm_get_latest_error_msg(m_mgmsrv)); @@ -440,32 +453,90 @@ CommandInterpreter::printError() //***************************************************************************** //***************************************************************************** -bool +static int do_event_thread; +static void* +event_thread_run(void* m) +{ + NdbMgmHandle handle= *(NdbMgmHandle*)m; + + my_thread_init(); + + int filter[] = { 15, NDB_MGM_EVENT_CATEGORY_BACKUP, 0 }; + int fd = ndb_mgm_listen_event(handle, filter); + if (fd > 0) + { + char *tmp= 0; + char buf[1024]; + SocketInputStream in(fd,10); + do { + if (tmp == 0) NdbSleep_MilliSleep(10); + if((tmp = in.gets(buf, 1024))) + ndbout << tmp; + } while(do_event_thread); + } + + my_thread_end(); + NdbThread_Exit(0); + return 0; +} + +bool CommandInterpreter::connect() { - if(!connected) { + if(!m_connected) + { if(!ndb_mgm_connect(m_mgmsrv, try_reconnect-1, 5, 1)) { - connected = true; - if (m_verbose) + const char *host= ndb_mgm_get_connected_host(m_mgmsrv); + unsigned port= ndb_mgm_get_connected_port(m_mgmsrv); + if(!ndb_mgm_set_connectstring(m_mgmsrv2, + BaseString(host).appfmt(":%d",port).c_str()) + && + !ndb_mgm_connect(m_mgmsrv2, try_reconnect-1, 5, 1)) { - printf("Connected to Management Server at: %s:%d\n", - ndb_mgm_get_connected_host(m_mgmsrv), - ndb_mgm_get_connected_port(m_mgmsrv)); + m_connected= true; + if (m_verbose) + { + printf("Connected to Management Server at: %s:%d\n", + host, port); + } + { + do_event_thread= 1; + m_event_thread = NdbThread_Create(event_thread_run, + (void**)&m_mgmsrv2, + 32768, + "CommandInterpreted_event_thread", + NDB_THREAD_PRIO_LOW); + } + } + else + { + ndb_mgm_disconnect(m_mgmsrv); } } } - return connected; + return m_connected; } bool CommandInterpreter::disconnect() { - if (connected && (ndb_mgm_disconnect(m_mgmsrv) == -1)) { - ndbout_c("Could not disconnect from management server"); - printError(); + if (m_event_thread) { + void *res; + do_event_thread= 0; + NdbThread_WaitFor(m_event_thread, &res); + NdbThread_Destroy(&m_event_thread); + m_event_thread= 0; + ndb_mgm_disconnect(m_mgmsrv2); + } + if (m_connected) + { + if (ndb_mgm_disconnect(m_mgmsrv) == -1) { + ndbout_c("Could not disconnect from management server"); + printError(); + } + m_connected= false; } - connected = false; return true; } @@ -914,7 +985,8 @@ CommandInterpreter::executeShutdown(char* parameters) return result; } - connected = false; + m_connected= false; + disconnect(); ndbout << "NDB Cluster management server shutdown." << endl; return 0; } @@ -1882,21 +1954,68 @@ CommandInterpreter::executeEventReporting(int processId, * Backup *****************************************************************************/ int -CommandInterpreter::executeStartBackup(char* /*parameters*/) +CommandInterpreter::executeStartBackup(char* parameters) { struct ndb_mgm_reply reply; unsigned int backupId; - +#if 0 int filter[] = { 15, NDB_MGM_EVENT_CATEGORY_BACKUP, 0 }; int fd = ndb_mgm_listen_event(m_mgmsrv, filter); - int result = ndb_mgm_start_backup(m_mgmsrv, &backupId, &reply); + if (fd < 0) + { + ndbout << "Initializing start of backup failed" << endl; + printError(); + return fd; + } +#endif + Vector<BaseString> args; + { + BaseString(parameters).split(args); + for (unsigned i= 0; i < args.size(); i++) + if (args[i].length() == 0) + args.erase(i--); + else + args[i].ndb_toupper(); + } + int sz= args.size(); + + int result; + if (sz == 2 && + args[1] == "NOWAIT") + { + result = ndb_mgm_start_backup(m_mgmsrv, 0, &backupId, &reply); + } + else if (sz == 1 || + (sz == 3 && + args[1] == "WAIT" && + args[2] == "COMPLETED")) + { + ndbout_c("Waiting for completed, this may take several minutes"); + result = ndb_mgm_start_backup(m_mgmsrv, 2, &backupId, &reply); + } + else if (sz == 3 && + args[1] == "WAIT" && + args[2] == "STARTED") + { + ndbout_c("Waiting for started, this may take several minutes"); + result = ndb_mgm_start_backup(m_mgmsrv, 1, &backupId, &reply); + } + else + { + invalid_command(parameters); + return -1; + } + if (result != 0) { ndbout << "Start of backup failed" << endl; printError(); +#if 0 close(fd); +#endif return result; } - +#if 0 + ndbout_c("Waiting for completed, this may take several minutes"); char *tmp; char buf[1024]; { @@ -1923,29 +2042,39 @@ CommandInterpreter::executeStartBackup(char* /*parameters*/) ndbout << tmp; } } while(tmp && tmp[0] != 0); - + close(fd); +#endif return 0; } void CommandInterpreter::executeAbortBackup(char* parameters) { - strtok(parameters, " "); - struct ndb_mgm_reply reply; - char* id = strtok(NULL, "\0"); int bid = -1; - if(id == 0 || sscanf(id, "%d", &bid) != 1){ - ndbout << "Invalid arguments: expected <BackupId>" << endl; - return; + struct ndb_mgm_reply reply; + if (emptyString(parameters)) + goto executeAbortBackupError1; + + { + strtok(parameters, " "); + char* id = strtok(NULL, "\0"); + if(id == 0 || sscanf(id, "%d", &bid) != 1) + goto executeAbortBackupError1; } - int result = ndb_mgm_abort_backup(m_mgmsrv, bid, &reply); - if (result != 0) { - ndbout << "Abort of backup " << bid << " failed" << endl; - printError(); - } else { - ndbout << "Abort of backup " << bid << " ordered" << endl; + { + int result= ndb_mgm_abort_backup(m_mgmsrv, bid, &reply); + if (result != 0) { + ndbout << "Abort of backup " << bid << " failed" << endl; + printError(); + } else { + ndbout << "Abort of backup " << bid << " ordered" << endl; + } } + return; + executeAbortBackupError1: + ndbout << "Invalid arguments: expected <BackupId>" << endl; + return; } #ifdef HAVE_GLOBAL_REPLICATION diff --git a/ndb/src/mgmsrv/MgmtSrvr.cpp b/ndb/src/mgmsrv/MgmtSrvr.cpp index c5f08df8a6..d0e1207cef 100644 --- a/ndb/src/mgmsrv/MgmtSrvr.cpp +++ b/ndb/src/mgmsrv/MgmtSrvr.cpp @@ -2397,7 +2397,7 @@ MgmtSrvr::eventReport(NodeId nodeId, const Uint32 * theData) * Backup ***************************************************************************/ int -MgmtSrvr::startBackup(Uint32& backupId, bool waitCompleted) +MgmtSrvr::startBackup(Uint32& backupId, int waitCompleted) { bool next; NodeId nodeId = 0; @@ -2419,11 +2419,16 @@ MgmtSrvr::startBackup(Uint32& backupId, bool waitCompleted) req->backupDataLen = 0; int result; - if (waitCompleted) { - result = sendRecSignal(nodeId, WAIT_BACKUP_COMPLETED, signal, true); + if (waitCompleted == 2) { + result = sendRecSignal(nodeId, WAIT_BACKUP_COMPLETED, + signal, true, 30*60*1000 /*30 secs*/); + } + else if (waitCompleted == 1) { + result = sendRecSignal(nodeId, WAIT_BACKUP_STARTED, + signal, true, 5*60*1000 /*5 mins*/); } else { - result = sendRecSignal(nodeId, WAIT_BACKUP_STARTED, signal, true); + result = sendRecSignal(nodeId, NO_WAIT, signal, true); } if (result == -1) { return SEND_OR_RECEIVE_FAILED; @@ -2502,18 +2507,31 @@ MgmtSrvr::abortBackup(Uint32 backupId) void MgmtSrvr::backupCallback(BackupEvent & event) { + DBUG_ENTER("MgmtSrvr::backupCallback"); m_lastBackupEvent = event; switch(event.Event){ case BackupEvent::BackupFailedToStart: + DBUG_PRINT("info",("BackupEvent::BackupFailedToStart")); + theWaitState = NO_WAIT; + break; case BackupEvent::BackupAborted: + DBUG_PRINT("info",("BackupEvent::BackupAborted")); + theWaitState = NO_WAIT; + break; case BackupEvent::BackupCompleted: + DBUG_PRINT("info",("BackupEvent::BackupCompleted")); theWaitState = NO_WAIT; break; case BackupEvent::BackupStarted: if(theWaitState == WAIT_BACKUP_STARTED) + { + DBUG_PRINT("info",("BackupEvent::BackupStarted NO_WAIT")); theWaitState = NO_WAIT; + } else { + DBUG_PRINT("info",("BackupEvent::BackupStarted")); + } } - return; + DBUG_VOID_RETURN; } diff --git a/ndb/src/mgmsrv/MgmtSrvr.hpp b/ndb/src/mgmsrv/MgmtSrvr.hpp index 0cad99e9d8..21311ad78c 100644 --- a/ndb/src/mgmsrv/MgmtSrvr.hpp +++ b/ndb/src/mgmsrv/MgmtSrvr.hpp @@ -60,6 +60,8 @@ public: } void add_listener(const Event_listener&); + void update_max_log_level(const LogLevel&); + void update_log_level(const LogLevel&); void log(int eventType, const Uint32* theData, NodeId nodeId); @@ -67,6 +69,8 @@ public: Event_listener& operator[](unsigned i) { return m_clients[i]; } const Event_listener& operator[](unsigned i) const { return m_clients[i]; } + void lock() { m_clients.lock(); } + void unlock(){ m_clients.unlock(); } }; /** @@ -360,7 +364,7 @@ public: /** * Backup functionallity */ - int startBackup(Uint32& backupId, bool waitCompleted = false); + int startBackup(Uint32& backupId, int waitCompleted= 2); int abortBackup(Uint32 backupId); int performBackup(Uint32* backupId); diff --git a/ndb/src/mgmsrv/Services.cpp b/ndb/src/mgmsrv/Services.cpp index 5b8d12e44d..edd1e6ef07 100644 --- a/ndb/src/mgmsrv/Services.cpp +++ b/ndb/src/mgmsrv/Services.cpp @@ -176,6 +176,7 @@ ParserRow<MgmApiSession> commands[] = { MGM_ARG("args", String, Mandatory, "Args(space separated int's)"), MGM_CMD("start backup", &MgmApiSession::startBackup, ""), + MGM_ARG("completed", Int, Optional ,"Wait until completed"), MGM_CMD("abort backup", &MgmApiSession::abortBackup, ""), MGM_ARG("id", Int, Mandatory, "Backup id"), @@ -629,85 +630,30 @@ MgmApiSession::getVersion(Parser<MgmApiSession>::Context &, m_output->println("string: %s", NDB_VERSION_STRING); m_output->println(""); } -#if 0 - -/***************************************************************************** - * BACKUP - *****************************************************************************/ - -int completed; -MgmtSrvr::BackupEvent globalEvent; - -static void -completedCallback(const MgmtSrvr::BackupEvent & event){ - - ndbout << "WaitCallback" << endl; - // Save event in the latestEvent var - - switch(event.Event){ - case MgmtSrvr::BackupEvent::BackupCompleted: - case MgmtSrvr::BackupEvent::BackupFailedToStart: - globalEvent = event; - completed = 1; - break; - } -} void MgmApiSession::startBackup(Parser<MgmApiSession>::Context &, - Properties const &) { + Properties const &args) { + DBUG_ENTER("MgmApiSession::startBackup"); unsigned backupId; + Uint32 completed= 2; int result; - MgmtSrvr::BackupCallback prevCallback; - prevCallback = m_mgmsrv.setCallback(completedCallback); - completed = 0; - result = m_mgmsrv.startBackup(backupId); - if (result == 0){ - - // Wait for the callback to call our condition - // waitFor(); - while (completed == 0) - NdbSleep_SecSleep(0); - - if (globalEvent.Event == MgmtSrvr::BackupEvent::BackupFailedToStart) - result = globalEvent.FailedToStart.ErrorCode; - else - backupId = globalEvent.Completed.BackupId; - } - - // restore old callback - m_mgmsrv.setCallback(prevCallback); - - m_output->println("start backup reply"); - if(result != 0) - m_output->println("result: %s(%d)", get_error_text(result), result); - else{ - m_output->println("result: Ok"); - m_output->println("id: %d", backupId); - } - m_output->println(""); - -} -#endif - -void -MgmApiSession::startBackup(Parser<MgmApiSession>::Context &, - Properties const &) { - unsigned backupId; - int result; + args.get("completed", &completed); - result = m_mgmsrv.startBackup(backupId, true); + result = m_mgmsrv.startBackup(backupId, completed); m_output->println("start backup reply"); if(result != 0) + { m_output->println("result: %s", get_error_text(result)); + } else{ m_output->println("result: Ok"); m_output->println("id: %d", backupId); } m_output->println(""); - + DBUG_VOID_RETURN; } void @@ -804,19 +750,22 @@ MgmApiSession::setClusterLogLevel(Parser<MgmApiSession>::Context &, LogLevel::EventCategory category= (LogLevel::EventCategory)(cat-(int)CFG_MIN_LOGLEVEL); + m_mgmsrv.m_event_listner.lock(); if (m_mgmsrv.m_event_listner[0].m_logLevel.setLogLevel(category,level)) { m_output->println(reply); m_output->println("result: Invalid category %d", category); m_output->println(""); + m_mgmsrv.m_event_listner.unlock(); DBUG_VOID_RETURN; } + m_mgmsrv.m_event_listner.unlock(); - EventSubscribeReq req; - req.blockRef = 0; - req.noOfEntries = 1; - req.theData[0] = (category << 16) | level; - m_mgmsrv.m_log_level_requests.push_back(req); + { + LogLevel ll; + ll.setLogLevel(category,level); + m_mgmsrv.m_event_listner.update_max_log_level(ll); + } m_output->println(reply); m_output->println("result: Ok"); @@ -827,13 +776,13 @@ MgmApiSession::setClusterLogLevel(Parser<MgmApiSession>::Context &, void MgmApiSession::setLogLevel(Parser<MgmApiSession>::Context &, Properties const &args) { - Uint32 node = 0, level = 0, category; + Uint32 node = 0, level = 0, cat; BaseString errorString; SetLogLevelOrd logLevel; int result; logLevel.clear(); args.get("node", &node); - args.get("category", &category); + args.get("category", &cat); args.get("level", &level); /* XXX should use constants for this value */ @@ -844,12 +793,15 @@ MgmApiSession::setLogLevel(Parser<MgmApiSession>::Context &, return; } - EventSubscribeReq req; - req.blockRef = node; - req.noOfEntries = 1; - req.theData[0] = (category << 16) | level; - m_mgmsrv.m_log_level_requests.push_back(req); - + LogLevel::EventCategory category= + (LogLevel::EventCategory)(cat-(int)CFG_MIN_LOGLEVEL); + + { + LogLevel ll; + ll.setLogLevel(category,level); + m_mgmsrv.m_event_listner.update_max_log_level(ll); + } + m_output->println("set loglevel reply"); m_output->println("result: Ok"); m_output->println(""); @@ -1319,24 +1271,22 @@ Ndb_mgmd_event_service::log(int eventType, const Uint32* theData, NodeId nodeId) tmp.set_max(m_clients[i].m_logLevel); } m_clients.unlock(); - - if(!(tmp == m_logLevel)){ - m_logLevel = tmp; - EventSubscribeReq req; - req = tmp; - req.blockRef = 0; - m_mgmsrv->m_log_level_requests.push_back(req); - } + update_log_level(tmp); } - DBUG_VOID_RETURN; + DBUG_VOID_RETURN; } void -Ndb_mgmd_event_service::add_listener(const Event_listener& client){ - m_clients.push_back(client); - LogLevel tmp = m_logLevel; - tmp.set_max(client.m_logLevel); - +Ndb_mgmd_event_service::update_max_log_level(const LogLevel &log_level) +{ + LogLevel tmp= m_logLevel; + tmp.set_max(log_level); + update_log_level(tmp); +} + +void +Ndb_mgmd_event_service::update_log_level(const LogLevel &tmp) +{ if(!(tmp == m_logLevel)){ m_logLevel = tmp; EventSubscribeReq req; @@ -1346,14 +1296,22 @@ Ndb_mgmd_event_service::add_listener(const Event_listener& client){ } } +void +Ndb_mgmd_event_service::add_listener(const Event_listener& client){ + m_clients.push_back(client); + update_max_log_level(client.m_logLevel); +} + void Ndb_mgmd_event_service::stop_sessions(){ + m_clients.lock(); for(int i = m_clients.size() - 1; i >= 0; i--){ if(m_clients[i].m_socket >= 0){ NDB_CLOSE_SOCKET(m_clients[i].m_socket); m_clients.erase(i); } } + m_clients.unlock(); } void diff --git a/ndb/test/src/NdbBackup.cpp b/ndb/test/src/NdbBackup.cpp index 09f52bf0be..5e22468692 100644 --- a/ndb/test/src/NdbBackup.cpp +++ b/ndb/test/src/NdbBackup.cpp @@ -46,7 +46,8 @@ NdbBackup::start(unsigned int & _backup_id){ ndb_mgm_reply reply; reply.return_code = 0; - if (ndb_mgm_start_backup(handle, + if (ndb_mgm_start_backup(handle, + 2, // wait until completed &_backup_id, &reply) == -1) { g_err << "Could not start backup " << endl; -- 2.30.9