| // ********************************************************************** |
| // @@@ START COPYRIGHT @@@ |
| // |
| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| // |
| // @@@ END COPYRIGHT @@@ |
| // ********************************************************************** |
| |
| #include "QmmQmm.h" |
| #include "QueryRewriteServer.h" |
| #include "Ipc.h" |
| #include "ComCextdecs.h" |
| #include "ComRtUtils.h" |
| #include "PortProcessCalls.h" |
| |
| #ifdef NA_NSK |
| extern "C" |
| { |
| #include "cextdecs.h(NODENUMBER_TO_NODENAME_, \ |
| REMOTEPROCESSORSTATUS, \ |
| PROCESSHANDLE_DECOMPOSE_, \ |
| PROCESSSTRING_SCAN_, \ |
| PROCESS_SPAWN_, \ |
| FILENAME_RESOLVE_, \ |
| FILENAME_TO_PATHNAME_)" |
| } |
| |
| #else |
| #include "cextdecs/cextdecs.h" |
| #endif |
| |
| using namespace QR; |
| |
| IpcServerClass* QmpStub::qmpServerClass_ = NULL; |
| IpcServerClass* QmsStub::qmsServerClass_ = NULL; |
| |
| void QmmGuaReceiveControlConnection::actOnSystemMessage |
| (short messageNum, |
| IpcMessageBufferPtr sysMsg, |
| IpcMessageObjSize sysMsgLen, |
| short clientFileNumber, |
| const GuaProcessHandle& clientPhandle, |
| GuaConnectionToClient* connection) |
| { |
| // @ZXros -- these declns go with ifdef'ed code below |
| #ifdef NA_NSK |
| short result; |
| short cpu; |
| short pin; |
| Int32 segmentNumber; |
| #endif |
| switch (messageNum) |
| { |
| case ZSYS_VAL_SMSG_OPEN: |
| { |
| NAString qmmName("qmm"); |
| QmmMessageStream* msgStream = |
| new QmmMessageStream(const_cast<IpcEnvironment*> |
| (qmm_->getEnvironment()), |
| qmmName, qmm_); |
| msgStream->addRecipient(connection); |
| //connection->receive(msgStream); |
| msgStream->receive(FALSE); |
| } |
| break; |
| |
| case ZSYS_VAL_SMSG_CLOSE: |
| case ZSYS_VAL_SMSG_PROCDEATH: |
| qmm_->handleClientExit((short *)&(clientPhandle.phandle_), messageNum); |
| if (connection) |
| connection->setFatalError(NULL); //@ZX |
| break; |
| |
| case ZSYS_VAL_SMSG_CPUDOWN: |
| { |
| zsys_ddl_smsg_cpudown_def* msg = (zsys_ddl_smsg_cpudown_def*)sysMsg; |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "Cpu %d on qmm's segment has gone down.", msg->z_cpunumber); |
| qmm_->getQmsStub(1, msg->z_cpunumber) |
| ->disable(QmsStub::CPU_NOT_REACHABLE); |
| } |
| break; |
| |
| case ZSYS_VAL_SMSG_CPUUP: |
| { |
| zsys_ddl_smsg_cpuup_def* msg = (zsys_ddl_smsg_cpuup_def*)sysMsg; |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "Cpu %d on qmm's segment has come back up.", msg->z_cpunumber); |
| //qmm_->getQmsStub(1, msg->z_cpunumber)->start(); |
| qmm_->getQmsStub(1, msg->z_cpunumber)->scheduleRestart(); |
| } |
| break; |
| |
| case ZSYS_VAL_SMSG_REMOTECPUDOWN: |
| { |
| zsys_ddl_smsg_remotecpudown_def* msg = |
| (zsys_ddl_smsg_remotecpudown_def*)sysMsg; |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "Cpu %d on segment %d has gone down.", msg->z_cpunumber, msg->z_nodenumber); |
| |
| qmm_->getQmsStub(msg->z_nodenumber, msg->z_cpunumber) |
| ->disable(QmsStub::CPU_NOT_REACHABLE); |
| } |
| break; |
| |
| case ZSYS_VAL_SMSG_REMOTECPUUP: |
| { |
| zsys_ddl_smsg_remotecpuup_def* msg = |
| (zsys_ddl_smsg_remotecpuup_def*)sysMsg; |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "Cpu %d on segment %d has come back up.", msg->z_cpunumber, msg->z_nodenumber); |
| //qmm_->getQmsStub(msg->z_nodenumber, msg->z_cpunumber)->start(); |
| qmm_->getQmsStub(msg->z_nodenumber, msg->z_cpunumber)->scheduleRestart(); |
| } |
| break; |
| |
| case ZSYS_VAL_SMSG_NODEDOWN: |
| { |
| zsys_ddl_smsg_nodedown_def* msg = (zsys_ddl_smsg_nodedown_def*)sysMsg; |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "Segment %d has gone down.", msg->z_nodenumber); |
| for (short cpu = 0; cpu < CPUS_PER_SEGMENT; cpu++) |
| qmm_->getQmsStub(msg->z_nodenumber, cpu) |
| ->disable(QmsStub::SEGMENT_NOT_REACHABLE); |
| } |
| break; |
| |
| case ZSYS_VAL_SMSG_NODEUP: |
| { |
| zsys_ddl_smsg_nodeup_def* msg = (zsys_ddl_smsg_nodeup_def*)sysMsg; |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "Segment %d has come back up.", msg->z_nodenumber); |
| for (short cpu = 0; cpu < CPUS_PER_SEGMENT; cpu++) |
| //qmm_->getQmsStub(msg->z_nodenumber, cpu)->start(); |
| qmm_->getQmsStub(msg->z_nodenumber, cpu)->scheduleRestart(); |
| } |
| break; |
| |
| default: |
| break; |
| } |
| |
| // See if we have any processes due to be restarted. |
| QRProcessStub::checkRestarts(); |
| } |
| |
| // Static singleton instance of the class. |
| Qmm* Qmm::instance_ = NULL; |
| |
| Qmm::Qmm(CollHeap* heap) |
| : qmsPool_(NULL), |
| qmsCount_(0), |
| qmp_(NULL), |
| heap_(heap) |
| { |
| // Establish the IPC heap and cache the IpcEnvironment ptr from |
| // MvQueryRewriteServer. |
| MvQueryRewriteServer::setHeap(heap); |
| ipcEnv_ = MvQueryRewriteServer::getIpcEnv(); |
| |
| // Set static IpcServerClass for QmsStub to use when instantiating qms |
| // processes. |
| QmsStub::setQmsServerClass(new(heap_) IpcServerClass(ipcEnv_, |
| IPC_SQLQMS_SERVER, |
| IPC_SPAWN_OSS_PROCESS)); |
| } |
| |
| void Qmm::handleClientExit(const short* phandle, short messageNum) |
| { |
| short cpu, pin; |
| Int32 segmentNumber; |
| short result = 0; |
| NABoolean qmpDied = FALSE; |
| |
| Int32 lc_pin; |
| Int32 lc_cpu; |
| Int32 lc_seg; |
| |
| result = XPROCESSHANDLE_DECOMPOSE_ ((SB_Phandle_Type *)phandle, &lc_cpu, &lc_pin, &segmentNumber); |
| cpu = lc_cpu; |
| pin = lc_pin; |
| |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "dead process is on cpu <%d> and pin <%d>", cpu, pin); |
| |
| |
| if (qmp_ && *qmp_ == *(SB_Phandle_Type *)phandle) |
| { |
| qmpDied = TRUE; |
| } |
| |
| if (!qmpDied) |
| { |
| // When a qms dies we are delivered a null connection pointer and handle. |
| // When this happens, we check the status of our qms processes until we find |
| // one that's no longer alive. If >1 close messages arrive close together, |
| // we may not get the same one referenced by the message, but that one will |
| // be restarted in response to a subsequent message. |
| NABoolean found = FALSE; |
| QmsStub* qmsStub; |
| for (Int32 i=0; i<qmsCount_ && !found; i++) |
| { |
| qmsStub = qmsPool_[i]; |
| NABoolean processDoesNotExist = FALSE; |
| |
| char procName[200]; |
| short result = -1; |
| |
| SB_Phandle_Type procHandle = qmsStub->getProcessHandle(); |
| NAProcessHandle processHandle(&procHandle); |
| Int32 guaRetcode = processHandle.decompose(); |
| if (!guaRetcode) |
| { |
| memset(procName, 0, sizeof(procName)); |
| memcpy(procName, processHandle.getPhandleString(), processHandle.getPhandleStringLen()); |
| result = msg_mon_get_process_info (procName, &lc_cpu, &lc_pin); |
| } |
| |
| if (result || (lc_cpu < 0) || (lc_pin < 0)) |
| processDoesNotExist = TRUE; |
| |
| if (qmsStub->getStatus() == QmsStub::RUNNING && processDoesNotExist) |
| { |
| found = TRUE; |
| QRLogger::log(CAT_QR_IPC, LL_FATAL, |
| "*** QMS on cpu %d of segment %d has died, " |
| "attempting to restart...", |
| qmsStub->getCpuNumber(), qmsStub->getSegmentNumber()); |
| qmsStub->disable(QmsStub::NOT_RUNNING); |
| qmsStub->scheduleRestart(); |
| } |
| } |
| |
| if (!found) |
| { |
| if (messageNum == ZSYS_VAL_SMSG_CLOSE) |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "*** Unknown client process has ended. ***"); |
| else if (messageNum == ZSYS_VAL_SMSG_PROCDEATH) |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "*** Unknown child process has died. ***"); |
| else |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "*** Unknown process has terminated, message=%d. ***", messageNum); |
| } |
| |
| return; |
| } |
| |
| if (messageNum == ZSYS_VAL_SMSG_CLOSE) |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "*** CLOSE message received for process %d,%d on segment %d ***", |
| cpu, pin, segmentNumber); |
| else if (messageNum == ZSYS_VAL_SMSG_PROCDEATH) |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "*** PROCDEATH message received for process %d,%d on segment %d ***", |
| cpu, pin, segmentNumber); |
| else |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "*** Process %d,%d on segment %d has terminated, message=%d. ***", |
| cpu, pin, segmentNumber, messageNum); |
| |
| if (qmp_ && *qmp_ == *(SB_Phandle_Type *)phandle) |
| { |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "*** QMP has died, will attempt to restart..."); |
| qmp_->scheduleRestart(); |
| } |
| else if (*getQmsStub(segmentNumber, cpu) == *(SB_Phandle_Type *)phandle) |
| { |
| // This probably never gets executed; process handle usually delivered |
| // for a qms termination is null, and is handled in loop above that looks |
| // for one that is no longer running. |
| QRLogger::log(CAT_QR_IPC, LL_ERROR, |
| "*** QMS on cpu %d of segment %d has died, " |
| "will attempt to restart...", |
| cpu, segmentNumber); |
| getQmsStub(segmentNumber, cpu)->disable(QmsStub::NOT_RUNNING); |
| getQmsStub(segmentNumber, cpu)->scheduleRestart(); |
| } |
| else |
| QRLogger::log(CAT_QR_IPC, LL_WARN, |
| "Could not identify terminated process as either QMP or QMS"); |
| } |
| |
| // Only used for Windows. |
| void Qmm::allocateQms() |
| { |
| char segmentName[SEGMENT_NAME_LEN + 1]; |
| short segmentNameLen = 0; |
| short result = NODENUMBER_TO_NODENAME_(-1, segmentName, SEGMENT_NAME_LEN, |
| &segmentNameLen); |
| segmentName[segmentNameLen] = '\0'; |
| QRLogger::log(CAT_QR_IPC, LL_DEBUG, |
| "Result for caller's segment is %d, segment name is %s", |
| result, segmentName); |
| qmsCount_ = 1; |
| qmsPool_ = new(heap_) QmsStub*[qmsCount_]; |
| qmsPool_[0] = new(heap_) QmsStub(1, segmentName, 0, 0, TRUE, TRUE, |
| /*qmsMsgStream_,*/ ipcEnv_, heap_); |
| } |
| |
| void Qmm::allocateQmsPool() |
| { |
| short qmsInx = 0; |
| Int32 maxCpuNum = 0; |
| Int32 lv_ret = 0; |
| Int32 nodeCount = 0; |
| Int32 nodeMax = 0; |
| MS_Mon_Node_Info_Entry_Type *nodeInfo = NULL; |
| |
| // Get the number of nodes to know how much info space to allocate |
| lv_ret = msg_mon_get_node_info(&nodeCount, 0, NULL); |
| if ((lv_ret == 0) && (nodeCount > 0)) |
| { |
| // Allocate the space for node info entries |
| nodeInfo = (MS_Mon_Node_Info_Entry_Type *) new(heap_) |
| char[nodeCount * sizeof(MS_Mon_Node_Info_Entry_Type)]; |
| if (nodeInfo) |
| { |
| // Get the node info |
| memset(nodeInfo, 0, sizeof(nodeInfo)); |
| nodeMax = nodeCount; |
| lv_ret = msg_mon_get_node_info(&nodeCount, nodeMax, nodeInfo); |
| if (lv_ret == 0) |
| { |
| // Find number of storage nodes by checking the storage bit. |
| // The computed value is used as node Id where QMSs will be created. |
| // QMSs should be running on the storage nodes only. |
| for ( Int32 i=0; i<nodeCount; i++ ) { |
| if (( nodeInfo[i].type & MS_Mon_ZoneType_Storage ) && |
| ( !nodeInfo[i].spare_node )) |
| maxCpuNum++; |
| } |
| } |
| NADELETEBASIC(nodeInfo,heap_); |
| } |
| } |
| |
| qmsCount_ = maxCpuNum; |
| qmsPool_ = new(heap_) QmsStub*[qmsCount_]; |
| |
| // Create a qms stub for each possible cpu in the cluster |
| for (short cpu = 0; cpu < maxCpuNum; cpu++) |
| { |
| qmsPool_[qmsInx++] = new(heap_) QmsStub(1, (char *)"\\NSK", cpu, 0, TRUE, TRUE, |
| /*qmsMsgStream_,*/ ipcEnv_, heap_); |
| } |
| |
| } // allocateQmsPool() |
| |
| // Make sure all QMSs started, and if not, retry up to maxRetries times, each |
| // time invoking a delay of delaySeconds seconds prior to starting it on the |
| // first cpu in the array on which it isn't running. When QMM dies and comes |
| // back up, sometimes the old one's QMSs haven't received the notification and |
| // self-terminated before the new QMM tries to create mew Q<Ss, and a "duplicate |
| // process name" error results. |
| void Qmm::checkAndRetryQms(Int16 maxRetries, Int16 delaySeconds) |
| { |
| NABoolean needRetry = TRUE; |
| for (Int16 i=1; i<=maxRetries && needRetry; i++) |
| { |
| NABoolean allOK = TRUE; |
| needRetry = FALSE; |
| for (Int16 cpuInx=0; cpuInx<qmsCount_; cpuInx++) |
| { |
| if (qmsPool_[cpuInx]->getStatus() == QmsStub::NOT_RUNNING) |
| { |
| if (allOK) // first one found that isn't running? |
| { |
| allOK = FALSE; |
| DELAY(delaySeconds * 100); |
| } |
| if (!qmsPool_[cpuInx]->start()) |
| needRetry = TRUE; // at least 1 qms didn't start |
| } |
| } |
| } |
| } |
| |
| void Qmm::startQmp(short cpu) |
| { |
| if (qmp_) |
| delete qmp_; |
| qmp_ = new(heap_) QmpStub(*ipcEnv_, qmpStartOpt_, cpu, heap_); |
| } |
| |
| NABoolean QmpStub::start() |
| { |
| char* baseProcName = MvQueryRewriteServer::getProcessName(IPC_SQLQMP_SERVER, NULL, cpu_); |
| |
| SB_Phandle_Type *phandle; |
| phandle = get_phandle_with_retry(baseProcName); |
| |
| if (phandle != NULL) |
| { |
| QRLogger::log(CAT_QR_IPC, LL_INFO, |
| "qmm found existing qmp process, not starting new one."); |
| setProcessHandle(*phandle); |
| return TRUE; |
| } |
| |
| switch ((Int32)qmpStartOpt_) |
| { |
| case SPAWN: |
| spawnProcess(ipcEnv_, cpu_); |
| break; |
| case SERVER: |
| allocateProcess(ipcEnv_, cpu_); |
| break; |
| case NONE: |
| // don't start qmp here; start manually |
| break; |
| default: |
| assertLogAndThrow1(CAT_QR_IPC, LL_ERROR, |
| FALSE, QmmException, |
| "Unknown option for starting QMP -- %d", |
| qmpStartOpt_); |
| break; |
| } |
| |
| return TRUE; |
| } |
| |
| void QmpStub::allocateProcess(IpcEnvironment& ipcEnv, short cpu) |
| { |
| if (!qmpServerClass_) |
| qmpServerClass_ = new(heap_) IpcServerClass(&ipcEnv, |
| IPC_SQLQMP_SERVER, |
| IPC_SPAWN_OSS_PROCESS); |
| ComDiagsArea* diagsArea = NULL; |
| IpcAllocateDiagsArea(diagsArea, heap_); |
| |
| // Have to supply name of segment (minus leading \, which gets prepended), or a |
| // fault will occur while populating the diagnostics area if there is an error. |
| char localSegmentName[SEGMENT_NAME_LEN + 1]; |
| short localSegmentNameLen; |
| short result = NODENUMBER_TO_NODENAME_(-1,localSegmentName, |
| SEGMENT_NAME_LEN, |
| &localSegmentNameLen); |
| assertLogAndThrow1(CAT_QR_IPC, LL_ERROR, |
| result==0, QmmException, |
| "Could not get name of local segment, error is %d", result); |
| localSegmentName[localSegmentNameLen] = '\0'; |
| qmpServer_ = qmpServerClass_->allocateServerProcess(&diagsArea, heap_, |
| localSegmentName+1, cpu, |
| IPC_PRIORITY_DONT_CARE, 1, |
| TRUE, TRUE); |
| if (qmpServer_) |
| { |
| |
| QRLogger::log(CAT_QR_IPC, LL_DEBUG, |
| "QMP process started on cpu #%d", qmpServer_->getServerId().getCpuNum()); |
| setProcessHandle(qmpServer_->getServerId().getPhandle().phandle_); |
| } |
| else |
| { |
| QRLogger::log(CAT_QR_IPC, LL_ERROR, |
| "Failed to allocate server process for QMP on cpu %d ", cpu); |
| QRLogger::logDiags(diagsArea, CAT_QR_IPC); |
| } |
| } |
| |
| void QmpStub::spawnProcess(IpcEnvironment& ipcEnv, short cpu) //, ComDiagsArea **diags, CollHeap *diagsHeap) |
| { |
| SB_Phandle_Type p_handle; |
| const char* progFile = "tdm_arkqmp"; |
| |
| #define MAX_PROC_ARGS 10 |
| #define SET_ARGV(argv,argc,argval) {argv[argc] = (char *) calloc(strlen(argval), 1); \ |
| strcpy(argv[argc++], argval); } |
| |
| Int32 largc = 0; |
| char *largv[MAX_PROC_ARGS]; |
| |
| Int32 server_nid = cpu; |
| Int32 server_pid = 0; |
| Int32 server_oid = 0; |
| //char process_name[100]; |
| char prog[MS_MON_MAX_PROCESS_PATH]; |
| |
| SET_ARGV(largv, largc, progFile); |
| SET_ARGV(largv, largc, "-oss"); |
| //SET_ARGV(largv, largc, NULL); |
| |
| strcpy(prog, getenv("TRAF_HOME")); |
| strcat(prog, "/export/bin32/"); |
| strcat(prog, progFile); |
| |
| char* tmpprocess_name = MvQueryRewriteServer::getProcessName(IPC_SQLQMP_SERVER, NULL, cpu, heap_); |
| |
| char process_name[100]; |
| strcpy(process_name, tmpprocess_name); |
| |
| msg_mon_start_process( |
| prog, /* prog */ |
| process_name, /* name */ |
| process_name, /* output process name */ |
| largc, /* args */ |
| largv, |
| &p_handle, |
| 0, /* open */ |
| &server_oid, /* oid */ |
| MS_ProcessType_Generic, /* process type */ |
| 0, /* priority */ |
| 0, /* debug */ |
| 0, /* backup */ |
| &server_nid, /* nid */ |
| &server_pid, |
| NULL, |
| NULL); /* pid */ |
| |
| setProcessHandle(p_handle); |
| } |
| |
| void Qmm::relayPendingPubsToQms() |
| { |
| QRXmlMessageObj* xmlMsgObj; |
| for (CollIndex i=0; i<pendingPubs_.entries(); i++) |
| { |
| xmlMsgObj = pendingPubs_[i]; |
| QRLogger::log(CAT_QR_IPC, LL_DEBUG, |
| "XML of publish message is:\n%s", xmlMsgObj->getData()); |
| for (Int32 qmsInx=0; qmsInx<qmsCount_; qmsInx++) |
| { |
| qmsPool_[qmsInx]->publish(xmlMsgObj); |
| } |
| } |
| |
| QRLogger::log(CAT_QR_IPC, LL_DEBUG, |
| "Sent %d Publish requests to QMS list", pendingPubs_.entries()); |
| for (CollIndex i=0; i<pendingPubs_.entries(); i++) |
| pendingPubs_[i]->decrRefCount(); |
| pendingPubs_.clear(); |
| } |
| |
| void Qmm::executeMessageLoop() |
| { |
| // Have to allocate this from heap, because ~IpcEnvironment deletes it. |
| QmmGuaReceiveControlConnection* conn = |
| new(heap_) QmmGuaReceiveControlConnection(ipcEnv_, this); |
| ipcEnv_->setControlConnection(conn); |
| NAString qmmName("qmm"); |
| QRMessageStream msgStream(ipcEnv_, qmmName); |
| //QRMessageRequest request(msgStream); |
| QRMessageObj* responseObj = NULL; |
| |
| while (!conn->getConnection()) |
| conn->wait(IpcInfiniteTimeout); |
| IpcConnection* firstClient = NULL; |
| |
| if (listenOpt_ == WAITONALL) |
| { |
| firstClient = conn->getConnection(); |
| msgStream.addRecipient(firstClient); |
| while (TRUE) |
| { |
| IpcAllConnections* allConns = ipcEnv_->getAllConnections(); |
| allConns->waitOnAll(); |
| } |
| } |
| else if (listenOpt_ == WAITCC) |
| { |
| WaitReturnStatus waitStatus; |
| while (TRUE) |
| { |
| //debugMessage1("IPC heap usage = %d", ipcEnv_.getHeap()->getAllocSize()); |
| waitStatus = conn->wait(getWaitTimeout()); |
| if (waitStatus == WAIT_OK) |
| { |
| QRProcessStub::checkRestarts(); |
| } |
| } |
| } |
| else |
| { |
| assertLogAndThrow1(CAT_QR_IPC, LL_ERROR, |
| listenOpt_ == RECEIVE, QmmException, |
| "Unknown listen opt -- %d", listenOpt_); |
| firstClient = conn->getConnection(); |
| msgStream.addRecipient(firstClient); |
| |
| while (TRUE) |
| { |
| msgStream.receive(); |
| //IpcAllConnections* allConns = ipcEnv_->getAllConnections(); |
| //allConns->waitOnAll(); |
| try |
| { |
| //responseObj = processRequestMessage(request); |
| responseObj = processRequestMessage(&msgStream); |
| } |
| catch(...) |
| {} |
| |
| // Either an exception was caught above, or a response was not generated |
| // for some other reason (shouldn't happen). Either way, create a status |
| // message indicating an internal error. |
| if (!responseObj) |
| responseObj = new QRStatusMessageObj(QR::InternalError); |
| |
| msgStream.clearAllObjects(); |
| msgStream.setType(responseObj->getType()); // so correct msg type logged |
| msgStream << *responseObj; |
| msgStream.send(); |
| responseObj->decrRefCount(); |
| relayPendingPubsToQms(); |
| } |
| } |
| } |
| |
| QRRequestResult Qmm::handlePublishRequest(QRMessageStream* msgStream) |
| { |
| // Have to use the global heap because IpcMessageObj is not derived from |
| // NABasicObject. |
| QRXmlMessageObj* xmlMsgObj = new QRXmlMessageObj(NULL, PUBLISH_REQUEST); |
| *msgStream >> *xmlMsgObj; |
| pendingPubs_.insert(xmlMsgObj); |
| return QR::Success; |
| } |
| |
| QRRequestResult Qmm::handleAllocateRequest(QRMessageStream* msgStream) |
| { |
| return QR::Unable; |
| } |
| |
| //QRMessageObj* Qmm::processRequestMessage(QRMessageRequest& request) |
| QRMessageObj* Qmm::processRequestMessage(QRMessageStream* msgStream) |
| { |
| QRRequestResult result; |
| XMLFormattedString resultXML; |
| QRMessageObj* responseMsgPtr = NULL; |
| //QRMessageTypeEnum requestType; |
| |
| if (!msgStream->moreObjects()) |
| { |
| QRLogger::log(CAT_QR_IPC, LL_ERROR, |
| "QMM received an empty message stream."); |
| return new QRStatusMessageObj(ProtocolError); |
| } |
| |
| // @ZX: Although we allow the possibility of multiple request objects per |
| // message, the returned response object will indicate the status of |
| // only the last one. Should probably return a status for each request |
| // item (or at least delete the overwritten ones so they don't leak). |
| // @ZX: Note that the message stream is cleared when we encounter an unknown |
| // or unhandled message type. How can an object be extracted from a stream |
| // if you don't understand its type? |
| while (msgStream->moreObjects()) |
| { |
| switch (msgStream->getNextObjType()) |
| { |
| // Can't use heap_ for the allocation of the response message because |
| // IpcMessageObj is not derived from NABasicObject. |
| |
| case PUBLISH_REQUEST: |
| result = handlePublishRequest(msgStream); |
| if (result == QR::Success) |
| QRLogger::log(CAT_QR_IPC, LL_INFO, "PUBLISH was successful."); |
| else |
| QRLogger::log(CAT_QR_IPC, LL_ERROR, |
| "PUBLISH failed, status is %d", result); |
| responseMsgPtr = new QRStatusMessageObj(result); |
| break; |
| |
| case ALLOCATE_REQUEST: |
| result = handleAllocateRequest(msgStream); |
| if (result == QR::Success) |
| QRLogger::log(CAT_QR_IPC, LL_INFO, "ALLOCATE was successful."); |
| else |
| QRLogger::log(CAT_QR_IPC, LL_ERROR, |
| "ALLOCATE failed, status is %d", result); |
| responseMsgPtr = new QRStatusMessageObj(result); |
| break; |
| |
| case DEFAULTS_REQUEST: |
| QRLogger::log(CAT_QR_IPC, LL_ERROR, |
| "'DEFAULTS' request not yet handled by QMM"); |
| responseMsgPtr = new QRStatusMessageObj(QR::InvalidRequest); |
| msgStream->clearAllObjects(); |
| break; |
| |
| default: |
| QRLogger::log(CAT_QR_IPC, LL_ERROR, |
| "QMM received unexpected message type: %d", msgStream->getNextObjType()); |
| responseMsgPtr = new QRStatusMessageObj(QR::InvalidRequest); |
| msgStream->clearAllObjects(); |
| break; |
| } |
| } |
| |
| return responseMsgPtr; |
| } // processRequestMessage |
| |
| IpcTimeout Qmm::getWaitTimeout() |
| { |
| const NAList<QRProcessStub*> restartList = QRProcessStub::getRestartList(); |
| CollIndex restartListEntries = restartList.entries(); |
| if (restartListEntries == 0) |
| return IpcInfiniteTimeout; |
| |
| // Find the time of the next scheduled restart. Init to first entry in list; |
| // we exited above is list was empty. |
| Int64 earliestTimestamp = restartList[0]->getLockoutEndTS(); |
| for (CollIndex i=1; i<restartListEntries; i++) |
| { |
| if (restartList[i]->getLockoutEndTS() < earliestTimestamp) |
| earliestTimestamp = restartList[i]->getLockoutEndTS(); |
| } |
| |
| // Timestamp is microsecond resolution, IpcTimeout is in 10ms units (100 = |
| // 1 second). |
| Int64 microsecondsTillNext = earliestTimestamp - NA_JulianTimestamp() |
| + 1000000; // 1-sec fudge factor |
| if (microsecondsTillNext < 0) // already past time somehow? |
| microsecondsTillNext = 0; |
| QRLogger::log(CAT_QR_IPC, LL_DEBUG, |
| "Wait timeout set to %d seconds", microsecondsTillNext/1000000); |
| return (IpcTimeout)(microsecondsTillNext / 1000); |
| } |
| |
| |
| NAList<QRProcessStub*> QRProcessStub::restartList_((CollHeap*)NULL); |
| |
| QRProcessStub::QRProcessStub(CollHeap* heap) |
| : lockoutEndTS_(0), |
| retryNumber_(0), |
| heap_(heap) |
| { |
| // Initialize the process handle to its null representation, all -1s. |
| nullProcessHandle(); |
| } |
| |
| void QRProcessStub::setLockout() |
| { |
| static Int32 seconds[] = {0, 10, 20, 30, 60, 120}; |
| Int32 lockoutSeconds = (retryNumber_ >= (sizeof seconds / sizeof seconds[0])) |
| ? 180 |
| : seconds[retryNumber_]; |
| QRLogger::log(CAT_QR_IPC, LL_DEBUG, |
| "Will not attempt another restart of this process for %d seconds.", |
| lockoutSeconds); |
| lockoutEndTS_ = NA_JulianTimestamp() + (lockoutSeconds * 1000000); |
| } |
| |
| void QRProcessStub::scheduleRestart() |
| { |
| if (lockoutEndTS_ > NA_JulianTimestamp()) |
| { |
| // Not time yet, add to restart list. When restarted and removed from |
| // restart list, will enter another (longer) lockout period. |
| restartList_.insert(this); |
| } |
| else |
| { |
| // Either there was no lockout or it has expired. Restart process, set |
| // retry count to 1. Only when a process dies again when it is in its |
| // lockout period do we queue the restart and extend the next lockout. |
| // If the restart fails, increment the retry count and put the stub in |
| // the restart list. |
| if (start()) |
| retryNumber_ = 1; |
| else |
| { |
| retryNumber_++; |
| restartList_.insert(this); |
| } |
| setLockout(); |
| } |
| } |
| |
| void QRProcessStub::checkRestarts() |
| { |
| // Go through the list backwards so indexing is not affected by removal of |
| // elements. CollIndex is an unsigned type, so have to start at entries() |
| // (1 past last valid index) instead of entries()-1, so we don't try to assign |
| // it a negative value in the case of an empty list. |
| QRProcessStub* processStub; |
| for (CollIndex i=restartList_.entries(); i>0; i--) |
| { |
| processStub = restartList_[i-1]; |
| if (processStub->lockoutEndTS_ < NA_JulianTimestamp()) |
| { |
| // If the process starts successfully, remove it from the list, else |
| // leave it there to try again next time. In either case, bump up the |
| // lockout period before the next time it can be restarted. |
| if (processStub->start()) |
| restartList_.removeAt(i-1); |
| processStub->retryNumber_++; |
| processStub->setLockout(); |
| } |
| } |
| } |
| |
| void QRProcessStub::nullProcessHandle() |
| { |
| } |
| |
| QmsStub::QmsStub(short segmentNumber, char* segmentName, short cpuNumber, |
| short segmentStatus, NABoolean cpuExists, |
| NABoolean cpuReachable, // QRMessageStream* qmsMsgStream, |
| IpcEnvironment* ipcEnv, |
| CollHeap* heap) |
| : QRProcessStub(heap), |
| segmentNumber_(segmentNumber), |
| cpuNumber_(cpuNumber), |
| segmentStatus_(segmentStatus), |
| qmsServer_(NULL), |
| status_(UNINITIALIZED), |
| //qmsMsgStream_(qmsMsgStream) |
| qmsMsgStream_(new(heap_) QRMessageStream(ipcEnv, "qmm", heap, PUBLISH_REQUEST)) |
| { |
| strcpy(segmentName_, segmentName); |
| |
| if (!cpuExists) |
| { |
| status_ = CPU_NOT_PRESENT; |
| return; |
| } |
| else if (!cpuReachable) |
| { |
| status_ = CPU_NOT_REACHABLE; |
| QRLogger::log(CAT_QR_IPC, LL_DEBUG, |
| "Processor %d on segment %s is not reachable.", cpuNumber, segmentName); |
| return; |
| } |
| else if (segmentStatus != 0) |
| { |
| status_ = SEGMENT_NOT_REACHABLE; // Segment status logged by caller |
| return; |
| } |
| |
| #ifndef NA_WINNT |
| qmsProcessName_ = MvQueryRewriteServer::getProcessName(IPC_SQLQMS_SERVER, segmentName_, cpuNumber, heap_); |
| #else |
| qmsProcessName_ = new(heap_) char[PROCESSNAME_STRING_LEN]; |
| sprintf(qmsProcessName_, "%s.%s%02d", segmentName_, QMS_PROCESS_PREFIX, cpuNumber); |
| #endif |
| |
| // Launch the QMS process. |
| start(); |
| } |
| |
| NABoolean QmsStub::start() |
| { |
| // Qmm class should have called static function to set this. |
| assertLogAndThrow(CAT_QR_IPC, LL_ERROR, |
| qmsServerClass_, QmmException, "qmsServerClass_ is NULL"); |
| |
| // Nodeup message is accompanied by a remote cpu up message for one processor, |
| // so there will be a redundant attempt to start qms on that cpu a 2nd time. |
| if (qmsServer_) |
| { |
| QRLogger::log(CAT_QR_IPC, LL_DEBUG, |
| "QmsStub::start() -- QMS process already running on cpu %d " |
| "of segment %s", cpuNumber_, segmentName_); |
| if (status_ == RUNNING) |
| { |
| QRLogger::log(CAT_QR_IPC, LL_INFO, "-- will not restart QMS process."); |
| return TRUE; |
| } |
| else |
| { |
| // Exists, but status not what we want. Get rid of it, new one will |
| // be started below. |
| QRLogger::log(CAT_QR_IPC, LL_DEBUG, |
| "-- QMS process has status %d, will restart.", status_); |
| disable(NOT_RUNNING); |
| } |
| } |
| |
| ComDiagsArea* diagsArea = NULL; |
| IpcAllocateDiagsArea(diagsArea, heap_); |
| |
| // Pass segmentName_+1 because IpcGuardianServer::spawnProcess() assumes the |
| // node name does not include the leading \, and adds one. |
| |
| qmsServer_ = qmsServerClass_->allocateServerProcess |
| (&diagsArea, heap_, segmentName_+1, |
| cpuNumber_, IPC_PRIORITY_DONT_CARE, 1, |
| TRUE, TRUE, 2, NULL, |
| qmsProcessName_); |
| |
| if (qmsServer_) |
| { |
| qmsMsgStream_->addRecipient(qmsServer_->getControlConnection()); |
| status_ = RUNNING; |
| |
| setProcessHandle(qmsServer_->getServerId().getPhandle().phandle_); |
| QRLogger::log(CAT_QR_IPC, LL_DEBUG, |
| "QMS process started on cpu #%d", qmsServer_->getServerId().getCpuNum()); |
| MvQueryRewriteServer::initQms(qmsServer_, heap_); |
| return TRUE; |
| } |
| else |
| { |
| status_ = NOT_RUNNING; |
| QRLogger::log(CAT_QR_IPC, LL_ERROR, |
| "Failed to allocate server process for QMS on cpu %d of segment %s", |
| cpuNumber_, segmentName_); |
| QRLogger::logDiags(diagsArea, CAT_QR_IPC); |
| return FALSE; |
| } |
| } |
| |
| void QmsStub::disable(Status reason) |
| { |
| setStatus(reason); |
| |
| // This stub may not have an active process. For example, a segment down will |
| // cause this to be called for each stub, including processors not present in |
| // the segment (e.g., a segment with only 8 instead of 16 processors). |
| if (qmsServer_) |
| { |
| qmsMsgStream_->deleteRecipient(qmsServer_->getControlConnection()); |
| qmsServer_->release(); |
| qmsServer_ = NULL; |
| nullProcessHandle(); |
| } |
| } |
| |
| void QmsStub::publish(QRXmlMessageObj* xmlMsgObj) |
| { |
| if (!qmsServer_) |
| return; |
| |
| qmsMsgStream_->clearAllObjects(); |
| qmsMsgStream_->setType(PUBLISH_REQUEST); |
| *qmsMsgStream_ << *xmlMsgObj; |
| qmsMsgStream_->send(); |
| } |
| |
| //void QmmMessageStream::actOnSend(IpcConnection* connection) |
| //{ |
| //} |
| |
| void QmmMessageStream::actOnReceive(IpcConnection* connection) |
| { |
| QRLogger::log(CAT_QR_IPC, LL_INFO, "Reached QmmMessageStream::actOnReceive()"); |
| QRMessageObj* responseObj = NULL; |
| try |
| { |
| responseObj = qmm_->processRequestMessage(this); |
| } |
| catch(...) |
| {} |
| |
| QRMessageStream::actOnReceive(connection); |
| respond(responseObj); |
| |
| qmm_->relayPendingPubsToQms(); |
| |
| // See if we have any processes due to be restarted. |
| QRProcessStub::checkRestarts(); |
| } |
