| /********************************************************************** |
| // @@@ 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 @@@ |
| **********************************************************************/ |
| /* -*-C++-*- |
| ***************************************************************************** |
| * |
| * File: ExUdrServer.cpp |
| * Description: Client-side process management for UDR servers |
| * |
| * Created: 08/16/2000 |
| * Language: C++ |
| * |
| * |
| ***************************************************************************** |
| */ |
| |
| #include "ex_stdh.h" |
| #include "ExUdrServer.h" |
| #include "ExUdrClientIpc.h" |
| #include "ExpError.h" |
| #include "ExCextdecs.h" |
| #include "ComRtUtils.h" |
| #include "PortProcessCalls.h" |
| |
| #include "seabed/fs.h" |
| #include "seabed/ms.h" |
| |
| #ifdef UDR_DEBUG |
| extern const char *GetWorkRetcodeString(ExWorkProcRetcode r); |
| /* |
| static const char *GetStatusString(ExUdrServer::ExUdrServerStatus s) |
| { |
| switch (s) |
| { |
| case ExUdrServer::EX_UDR_SUCCESS: |
| return "Success"; |
| break; |
| case ExUdrServer::EX_UDR_WARNING: |
| return "Warning"; |
| break; |
| case ExUdrServer::EX_UDR_ERROR: |
| return "Error"; |
| break; |
| } |
| return "***UNKNOWN***"; |
| } */ |
| |
| #define UdrDebug0(s) \ |
| ( UdrPrintf(traceFile_,(s)) ) |
| #define UdrDebug1(s,a1) \ |
| ( UdrPrintf(traceFile_,(s),(a1)) ) |
| #define UdrDebug2(s,a1,a2) \ |
| ( UdrPrintf(traceFile_,(s),(a1),(a2)) ) |
| #define UdrDebug3(s,a1,a2,a3) \ |
| ( UdrPrintf(traceFile_,(s),(a1),(a2),(a3)) ) |
| #define UdrDebug4(s,a1,a2,a3,a4) \ |
| ( UdrPrintf(traceFile_,(s),(a1),(a2),(a3),(a4)) ) |
| #define UdrDebug5(s,a1,a2,a3,a4,a5) \ |
| ( UdrPrintf(traceFile_,(s),(a1),(a2),(a3),(a4),(a5)) ) |
| |
| #else |
| // |
| // Debug macros are no-ops in the release build |
| // |
| #define UdrDebug0(s) |
| #define UdrDebug1(s,a1) |
| #define UdrDebug2(s,a1,a2) |
| #define UdrDebug3(s,a1,a2,a3) |
| #define UdrDebug4(s,a1,a2,a3,a4) |
| #define UdrDebug5(s,a1,a2,a3,a4,a5) |
| |
| #endif // UDR_DEBUG |
| |
| // |
| // Helper functions allowing the ExUdrServer class and its callers to |
| // determine if a process ID is NULL and to nullify a process |
| // ID. Internally in our IpcProcessId objects we consider a valid |
| // process ID to be anything with a domain other than IPC_DOM_INVALID. |
| // The default IpcProcessId constructor sets the domain to |
| // IPC_DOM_INVALID so this constructor can be used to instantiate a |
| // NULL process ID. |
| // |
| NABoolean ProcessIdIsNull(const IpcProcessId &id) |
| { |
| NABoolean result = FALSE; |
| if (id.getDomain() == IPC_DOM_INVALID) |
| { |
| result = TRUE; |
| } |
| return result; |
| } |
| void InvalidateProcessId(IpcProcessId &id) |
| { |
| // |
| // The default IpcProcessId constructor is used to instantiate |
| // a process ID with domain IPC_DOM_INVALID. |
| // |
| IpcProcessId nullPid; |
| // |
| // Invoke the IpcProcessId assignment operator to invalidate |
| // id. Arguments to this operator are passed by reference, not on |
| // the stack. |
| // |
| id = nullPid; |
| } |
| |
| // ----------------------------------------------------------------------- |
| // ExUdrServer |
| // ----------------------------------------------------------------------- |
| ExUdrServer::ExUdrServer(IpcEnvironment *env, |
| const Int32 &userId, |
| const char *options, |
| const char *optionDelimiters, |
| const char *userName, |
| const char *userPassword, |
| IpcServerClass *serverClass) |
| : state_(EX_UDR_NOT_STARTED), |
| ipcEnvironment_(env), |
| udrServerClass_(serverClass), |
| ipcServer_(NULL), |
| serverProcessId_(), |
| startAttempts_(0), |
| userId_(userId), |
| userName_(NULL), |
| userPassword_(NULL), |
| options_(NULL), |
| optionDelimiters_(NULL), |
| refCount_(0), |
| dedicated_(FALSE), |
| inUse_(FALSE), |
| inUseConns_(NULL), |
| freeConns_(NULL) |
| #ifdef UDR_DEBUG |
| , traceFile_(NULL) |
| #endif |
| { |
| ex_assert(options && optionDelimiters, |
| "No runtime options specified for UDR server startup"); |
| |
| CollHeap *h = myIpcHeap(); |
| |
| Int32 len = str_len(options); |
| options_ = new (h) char[len + 1]; |
| str_cpy_all(options_, options, len + 1); |
| |
| len = str_len(optionDelimiters); |
| optionDelimiters_ = new (h) char[len + 1]; |
| str_cpy_all(optionDelimiters_, optionDelimiters, len + 1); |
| |
| if (userName) |
| { |
| len = str_len(userName); |
| userName_ = new (h) char[len + 1]; |
| str_cpy_all(userName_, userName, len + 1); |
| } |
| |
| if (userPassword) |
| { |
| len = str_len(userPassword); |
| userPassword_ = new (h) char[len + 1]; |
| str_cpy_all(userPassword_, userPassword, len + 1); |
| } |
| |
| inUseConns_ = new (h) NAList<IpcConnection *>(h); |
| freeConns_ = new (h) NAList<IpcConnection *>(h); |
| } |
| |
| ExUdrServer::~ExUdrServer() |
| { |
| UdrDebug1("[BEGIN ExUdrServer destructor] %p", this); |
| |
| stop(); |
| |
| CollHeap *h = myIpcHeap(); |
| NADELETEBASIC(options_, h); |
| NADELETEBASIC(optionDelimiters_, h); |
| NADELETEBASIC(userName_, h); |
| NADELETEBASIC(userPassword_, h); |
| |
| UdrDebug1("[END ExUdrServer destructor] %p", this); |
| } |
| |
| CollHeap *ExUdrServer::myIpcHeap() const |
| { |
| return myIpcEnv()->getHeap(); |
| } |
| |
| // |
| // Bring a UDR Server process to life if one hasn't been |
| // started already |
| // |
| ExUdrServer::ExUdrServerStatus ExUdrServer::start(ComDiagsArea **diags, |
| CollHeap *diagsHeap, |
| Int64 transId, |
| IpcProcessId &newId, |
| NABoolean usesTransactions) |
| { |
| #ifdef UDR_DEBUG |
| UdrDebug1("[BEGIN ExUdrServer::start()] %p", this); |
| UdrDebug1(" Startup options '%s'", options_); |
| UdrDebug1(" Startup option delimiters '%s'", optionDelimiters_); |
| if (diags && *diags) |
| { |
| Lng32 numDiags = (*diags)->getNumber(); |
| UdrDebug1(" The diagnostics area initially contains %d entries", |
| numDiags); |
| } |
| else |
| { |
| UdrDebug0(" No diagnostics area exists yet"); |
| } |
| #endif // UDR_DEBUG |
| |
| // The newId and result variables hold our return values. We will |
| // assume failure initially and set the values to something else |
| // once we are sure we have succeeded. |
| InvalidateProcessId(newId); |
| ExUdrServerStatus result = EX_UDR_ERROR; |
| |
| if (ready()) |
| { |
| #ifdef UDR_DEBUG |
| char buf[300]; |
| serverProcessId_.toAscii(buf, 300); |
| UdrDebug1(" A server is already running. Process ID %s", buf); |
| #endif // UDR_DEBUG |
| newId = serverProcessId_; |
| result = EX_UDR_SUCCESS; |
| } |
| else |
| { |
| stop(); |
| |
| UdrDebug0(" About to start the UDR server..."); |
| |
| // Notes on UDR Server startup |
| // - We are using a nowait depth of 2 for first connection. SPJs |
| // that cannot return RS use first connection for IPC. |
| // We use nowait depth of 3 for all other connections used by SPJs |
| // that can return RS. Look at getAnIpcConnection() for code details. |
| // |
| // - By specifying IPC_CPU_DONT_CARE as the CPU number in this |
| // call to allocateServerProcess() we get the default behavior |
| // from Guardian which is to start the process on the same CPU as |
| // the caller. |
| |
| Lng32 nowaitDepth = 2; |
| |
| #ifdef _DEBUG |
| char *e = getenv("UDR_NOWAIT_DEPTH"); |
| if (e && e[0]) |
| nowaitDepth = atol(e); |
| #endif |
| UdrDebug1(" Using a nowait depth of %d", nowaitDepth); |
| |
| NABoolean waitedCreation = TRUE; |
| // co-locate the tdm_udrserv with the executor process (master or ESP) |
| // This is done for a couple of reasons: One is that since the ESPs |
| // are evenly balanced across the CPUs, this ensures an even distribution |
| // of the tdm_udrservs as well, probably better than a random distribution. |
| // The second reason is that for certain maintenance UDFs (only example |
| // so far is udf(event_log_reader())), we must ensure that we run one |
| // tdm_udrserv on each node of the cluster, and we do that by starting |
| // one ESP per node. |
| IpcCpuNum collocatedCPU = |
| myIpcEnv()->getMyOwnProcessId(IPC_DOM_GUA_PHANDLE).getCpuNum(); |
| |
| ipcServer_ = |
| udrServerClass_->allocateServerProcess(diags, |
| diagsHeap, |
| NULL, |
| collocatedCPU, |
| 1, // espLevel (not relevant for UDR servers) |
| usesTransactions, |
| waitedCreation, |
| nowaitDepth); |
| |
| #ifdef UDR_DEBUG |
| UdrDebug1(" allocateServerProcess() returned %p", ipcServer_); |
| if (diags && *diags) |
| { |
| Lng32 numDiags = (*diags)->getNumber(); |
| UdrDebug1(" The diagnostics area contains %d entries", |
| numDiags); |
| } |
| else |
| { |
| UdrDebug0(" No diagnostics area exists"); |
| } |
| #endif // UDR_DEBUG |
| |
| startAttempts_++; |
| |
| if (diags && *diags) |
| { |
| Lng32 sqlcode = ((*diags)->mainSQLCODE()); |
| if (sqlcode < 0) |
| { |
| // |
| // An error occurrred |
| // |
| // $$$$ |
| // Need to verify whether we can release the server class |
| // instance Looks like allocateServerProcess() can return |
| // non-NULL and also generate diagnostics. If that happens and |
| // we call ipcServer_->release() an assertion fails because |
| // ipcServer_ does not yet have a valid process handle. The |
| // assertion is in IpcProcessId::getPhandle(). |
| // |
| UdrDebug0(" ***"); |
| UdrDebug1(" *** WARNING: Errors occurred. Main SQLCODE is %d", |
| sqlcode); |
| UdrDebug0(" ***"); |
| ipcServer_ = NULL; |
| } |
| } |
| |
| if (ipcServer_ && ipcServer_->getControlConnection()) |
| { |
| // We enter this block once a server process has successfully |
| // been started. |
| |
| setState(EX_UDR_READY); |
| |
| // Record the process ID |
| serverProcessId_ = ipcServer_->getControlConnection()->getOtherEnd(); |
| |
| #ifdef UDR_DEBUG |
| char buf[300]; |
| serverProcessId_.toAscii(buf, 300); |
| UdrDebug1(" A new server was started. Process ID %s", buf); |
| #endif // UDR_DEBUG |
| |
| // Set a flag in the control connection indicating whether or |
| // not to perform integrity checks on incoming buffers. |
| NABoolean trust = FALSE; |
| #ifdef UDR_DEBUG |
| if (getenv("UDR_TRUST_REPLIES")) |
| { |
| trust = TRUE; |
| } |
| #endif |
| ipcServer_->getControlConnection()->setTrustIncomingBuffers(trust); |
| |
| |
| // Send down any requested runtime options. Right now the only |
| // options we support are JVM startup options. |
| sendStartupOptions(diags, diagsHeap, transId); |
| |
| } // if (ipcServer_ && ipcServer_->getControlConnection()) |
| |
| if (ready()) |
| { |
| newId = serverProcessId_; |
| result = EX_UDR_SUCCESS; |
| } |
| else |
| { |
| UdrDebug0(" Unable to start the UDR server"); |
| |
| // |
| // If ready() is not TRUE and no diagnostics have been created yet, |
| // create the generic "Unable to receive reply from MXUDR" diagnostic |
| // here. |
| // |
| Lng32 numDiags = 0; |
| if (diags && *diags) |
| { |
| numDiags = (*diags)->getNumber(); |
| } |
| if (numDiags == 0) |
| { |
| UdrDebug0(" ***"); |
| UdrDebug0(" *** WARNING: Errors occurred but no diagnostics created"); |
| UdrDebug0(" ***"); |
| if (diags) |
| { |
| if (!(*diags)) |
| { |
| *diags = ComDiagsArea::allocate(diagsHeap); |
| } |
| **diags << DgSqlCode(-EXE_UDR_REPLY_ERROR); |
| } |
| } |
| |
| stop(); |
| result = EX_UDR_ERROR; |
| |
| } // if (!ready()) |
| |
| } // if (ready()) ... else ... |
| |
| UdrDebug1("[END ExUdrServer::start()] %p", this); |
| return result; |
| } |
| |
| void ExUdrServer::sendStartupOptions(ComDiagsArea **diags, |
| CollHeap *diagsHeap, |
| Int64 transId) |
| { |
| // Send down any requested runtime options. Right now the only |
| // options we support are JVM startup options. To do the work we |
| // will allocate a stream and a message on the IPC heap. The |
| // message gets cleaned up eventually when its reference count |
| // reaches zero. The stream takes care of its own cleanup by |
| // putting itself on the IPC environment's list of "completed" |
| // streams, and cleanup of that list is always guaranteed to be |
| // done at a safe time. |
| |
| // This method should only be called after startup was successful |
| ex_assert(ipcServer_ && ipcServer_->getControlConnection(), |
| "Do not call this method without a running server"); |
| |
| UdrDebug0(" About to send startup options to the server"); |
| NAMemory *ipcHeap = myIpcHeap(); |
| |
| // If the options_ is set to OFF or ANYTHING and there is |
| // no userName_ then we don't have anything to do here |
| if (userName_ == NULL && |
| (str_cmp_ne(options_, "OFF") == 0 || str_cmp_ne(options_, "ANYTHING") == 0)) |
| { |
| return ; |
| } |
| |
| // Send the user name also in startup options as |
| // "-Dsqlmx.udr.username=userName_" |
| char *userNameOption = NULL; |
| Int32 userNameOptionLen = 0; |
| |
| if (userName_) |
| { |
| const char *userNamePrefix = "-Dsqlmx.udr.username="; |
| Int32 userNamePrefixLen = str_len(userNamePrefix); |
| Int32 userNameLen = str_len(userName_); |
| userNameOption = new (ipcHeap) char[userNamePrefixLen + userNameLen + 1]; |
| str_sprintf(userNameOption, "%s%s", userNamePrefix, userName_); |
| userNameOptionLen = str_len(userNameOption); |
| } |
| |
| // Send the user password also in startup options as |
| // "-Dsqlmx.udr.password=userPassword_" |
| char *passwordOption = NULL; |
| Int32 passwordOptionLen = 0; |
| |
| if (userPassword_) |
| { |
| const char *passwordPrefix = "-Dsqlmx.udr.password="; |
| Int32 passwordPrefixLen = str_len(passwordPrefix); |
| Int32 passwordLen = str_len(userPassword_); |
| passwordOption = new (ipcHeap) char[passwordPrefixLen + passwordLen + 1]; |
| str_sprintf(passwordOption, "%s%s", passwordPrefix, userPassword_); |
| passwordOptionLen = str_len(passwordOption); |
| } |
| |
| char *optionsToSend = NULL; |
| Int32 delimiterLen = 1; |
| Int32 tmpLen = 0; |
| if (str_cmp_ne(options_, "OFF") != 0 && |
| str_cmp_ne(options_, "ANYTHING") != 0) |
| { |
| if (userName_) |
| { |
| Int32 optionsLen = str_len(options_); |
| Int32 len = optionsLen + delimiterLen + userNameOptionLen |
| + delimiterLen + passwordOptionLen; |
| |
| optionsToSend = new (ipcHeap) char[len + 1]; |
| str_cpy_all(optionsToSend, options_, optionsLen); |
| optionsToSend[optionsLen] = optionDelimiters_[0]; |
| |
| // Copy the user name |
| str_cpy_all(optionsToSend + optionsLen + delimiterLen, |
| userNameOption, |
| userNameOptionLen); |
| tmpLen = optionsLen + delimiterLen + userNameOptionLen; |
| optionsToSend[tmpLen] = optionDelimiters_[0]; |
| |
| // Copy the user password |
| str_cpy_all(optionsToSend + tmpLen + delimiterLen, |
| passwordOption, |
| passwordOptionLen); |
| optionsToSend[len] = '\0'; |
| } |
| else |
| { |
| Int32 optionsLen = str_len(options_); |
| optionsToSend = new (ipcHeap) char[optionsLen + 1]; |
| str_cpy_all(optionsToSend, options_, optionsLen + 1); |
| } |
| } |
| else |
| { |
| if (userName_) |
| { |
| if (userPassword_) |
| { |
| Int32 len = userNameOptionLen + delimiterLen + passwordOptionLen; |
| |
| optionsToSend = new (ipcHeap) char[len + 1]; |
| |
| // Copy the user name |
| str_cpy_all(optionsToSend, userNameOption, userNameOptionLen); |
| optionsToSend[userNameOptionLen] = ' '; |
| |
| // Copy the user password |
| str_cpy_all(optionsToSend + userNameOptionLen + delimiterLen, |
| passwordOption, passwordOptionLen); |
| optionsToSend[len] = '\0'; |
| } |
| else |
| { |
| optionsToSend = new (ipcHeap) char[userNameOptionLen + 1]; |
| |
| // Copy the user name |
| str_cpy_all(optionsToSend, userNameOption, userNameOptionLen + 1); |
| } |
| } |
| else |
| { |
| // No need to send any options. We never come here because this |
| // case is already checked above. |
| return; |
| } |
| } |
| |
| NABoolean isTransactional = (transId == -1 ? FALSE : TRUE); |
| |
| UdrClientControlStream *stream = new (ipcHeap) |
| UdrClientControlStream(myIpcEnv(), |
| NULL, // tcb |
| NULL, // stmt globals |
| TRUE, // keep diags for caller |
| isTransactional); |
| |
| #ifdef UDR_DEBUG |
| if (traceFile_) |
| { |
| stream->setTraceFile(traceFile_); |
| } |
| #endif |
| |
| UdrSessionMsg *msg = new (ipcHeap) |
| UdrSessionMsg(UdrSessionMsg::UDR_SESSION_TYPE_JAVA_OPTIONS, |
| 0, ipcHeap); |
| msg->addString(optionsToSend); |
| msg->addString(optionDelimiters_); |
| |
| IpcConnection *conn = ipcServer_->getControlConnection(); |
| stream->addRecipient(conn); |
| *stream << *msg; |
| stream->send(TRUE, // TRUE indicates a waited send |
| transId); |
| |
| msg->decrRefCount(); |
| NADELETEBASIC(userNameOption, ipcHeap); |
| NADELETEBASIC(passwordOption, ipcHeap); |
| NADELETEBASIC(optionsToSend, ipcHeap); |
| |
| //-------------------------------------------------------------------- |
| // We just completed a waited send of the startup options. There |
| // are a couple of error checks we need to perform now. |
| // a) The stream may have encountered IPC errors and put itself into |
| // an error state. |
| // b) The IPC could have been successful but the server may have |
| // returned SQL diags in its reply. Our UDR-specific message |
| // stream subclass caches these diags for us. |
| //-------------------------------------------------------------------- |
| |
| // This boolean will track whether we need to add a generic |
| // "Unable to receive reply from MXUDR" condition to the diags |
| // area. |
| NABoolean addUdrCondition = FALSE; |
| |
| // a) Did the stream put itself into an error state? |
| if (stream->getErrorInfo() != 0) |
| { |
| UdrDebug0(" The message stream encountered errors"); |
| addUdrCondition = TRUE; |
| if (diags) |
| { |
| if (*diags == NULL) |
| { |
| *diags = ComDiagsArea::allocate(diagsHeap); |
| } |
| conn->populateDiagsArea(*diags, diagsHeap); |
| } |
| stop(); |
| } |
| else |
| { |
| // b) Did the server return diags? |
| ComDiagsArea *diagsFromServer = stream->extractUdrDiags(); |
| if (diagsFromServer) |
| { |
| UdrDebug0(" The server returned diagnostics in its reply"); |
| if (diags) |
| { |
| addUdrCondition = TRUE; |
| if (*diags == NULL) |
| { |
| *diags = ComDiagsArea::allocate(diagsHeap); |
| } |
| (*diags)->mergeAfter(*diagsFromServer); |
| } |
| |
| diagsFromServer->decrRefCount(); |
| stop(); |
| } |
| } |
| |
| // If either a) or b) was true, we have diags to return but they |
| // may not be UDR-specific so we add a generic "Unable to |
| // receive reply from MXUDR" condition here. |
| if (diags && addUdrCondition) |
| { |
| if (*diags == NULL) |
| { |
| *diags = ComDiagsArea::allocate(diagsHeap); |
| } |
| **diags << DgSqlCode(-EXE_UDR_REPLY_ERROR); |
| } |
| |
| } |
| |
| // |
| // Bring down the server process. Note that this is not a forceful |
| // "kill" method. If the server is busy or hung then this call may not |
| // actually stop the process. Under normal circumstances the call to |
| // release() in this method will close the control connection to the |
| // server, the server will detect that its only client has gone away, |
| // and the server will exit. |
| // |
| ExUdrServer::ExUdrServerStatus ExUdrServer::stop() |
| { |
| UdrDebug1("[BEGIN ExUdrServer::stop()] %p", this); |
| |
| // Release all connections that were opened for this |
| // Server instance. We destruct the connections that are not being |
| // used. The in-use connections will be freed in releaseConnection() |
| // when they are tried to use next time. |
| for ( ; freeConns_->entries(); ) |
| { |
| IpcConnection *conn = freeConns_->at(0); |
| freeConns_->removeAt(0); |
| delete conn; |
| } |
| |
| if (ipcServer_) |
| { |
| UdrDebug0(" About to release the server class instance"); |
| // |
| // This call will remove the IpcServer instance from the server |
| // class and deallocate the IpcServer instance |
| // |
| ipcServer_->release(); |
| ipcServer_ = NULL; |
| InvalidateProcessId(serverProcessId_); |
| } |
| |
| setState(EX_UDR_NOT_STARTED); |
| |
| UdrDebug1("[END ExUdrServer::stop()] %p", this); |
| return EX_UDR_SUCCESS; |
| } |
| |
| ExUdrServer::ExUdrServerStatus ExUdrServer::kill(ComDiagsArea *diags) |
| { |
| UdrDebug1("[BEGIN ExUdrServer::kill()] %p", this); |
| short result = 0; |
| |
| char asciiPhandle[300]; |
| serverProcessId_.toAscii(asciiPhandle, 300); |
| UdrDebug1(" UDR Server process handle is %s", asciiPhandle); |
| |
| if (!ProcessIdIsNull(serverProcessId_)) |
| { |
| if (serverProcessId_.getDomain() == IPC_DOM_GUA_PHANDLE) |
| { |
| NAProcessHandle serverPhandle( |
| (SB_Phandle_Type *) &(serverProcessId_.getPhandle().phandle_)); |
| Int32 guaRetcode = serverPhandle.decompose(); |
| if (XZFIL_ERR_OK == guaRetcode) |
| msg_mon_stop_process_name(serverPhandle.getPhandleString()); |
| UdrDebug1(" PROCESS_STOP_ returned %d", (Int32) result); |
| if (diags != NULL) |
| { |
| *diags << DgSqlCode(EXE_UDR_ATTEMPT_TO_KILL) |
| << DgString0(asciiPhandle) |
| << DgInt0((Int32) result); |
| } |
| } |
| else |
| { |
| UdrDebug0(" *** ERROR: UDR Server is not a Guardian process"); |
| } |
| } |
| else |
| { |
| UdrDebug0(" Process handle is not valid"); |
| } |
| |
| UdrDebug1("[END ExUdrServer::kill()] %p", this); |
| return EX_UDR_SUCCESS; |
| } |
| |
| IpcConnection *ExUdrServer::getUdrControlConnection() const |
| { |
| IpcConnection *result = NULL; |
| if (ready() && ipcServer_) |
| { |
| result = ipcServer_->getControlConnection(); |
| |
| if (result && result->getState() == IpcConnection::ERROR_STATE) |
| result = NULL; |
| } |
| return result; |
| } |
| |
| // A free connection from freeConns_ will be moved into inUseConns_ list. |
| // If there is no available free connection, a new IPC Connection will |
| // be created. |
| IpcConnection *ExUdrServer::getAnIpcConnection() const |
| { |
| UdrDebug0("[BEGIN ExUdrServer::getAnIpcConnection()]"); |
| |
| IpcConnection *conn = NULL; |
| CollIndex numFreeConns = freeConns_->entries(); |
| |
| if (numFreeConns > 0) |
| { |
| // remove from freeList_ and add it in inUseList_ |
| conn = freeConns_->at(numFreeConns - 1); |
| freeConns_->removeAt(numFreeConns - 1); |
| inUseConns_->insert(conn); |
| |
| UdrDebug1(" An existing connection %p will be reused", conn); |
| } |
| else |
| { |
| Lng32 nowaitDepth = DEFAULT_NOWAIT_DEPTH; |
| |
| #ifdef _DEBUG |
| char *e = getenv("UDR_NOWAIT_DEPTH"); |
| if (e && e[0]) |
| nowaitDepth = atol(e); |
| #endif |
| UdrDebug1(" Using a nowait depth of %d", nowaitDepth); |
| |
| // create a new connection and add it in inUseList_ |
| conn = serverProcessId_.createConnectionToServer(myIpcEnv(), |
| TRUE, |
| nowaitDepth); |
| |
| // Set a flag in connection indicating whether or |
| // not to perform integrity checks on incoming buffers. |
| NABoolean trust = FALSE; |
| #ifdef UDR_DEBUG |
| if (getenv("UDR_TRUST_REPLIES")) |
| { |
| trust = TRUE; |
| } |
| #endif |
| |
| conn->setTrustIncomingBuffers(trust); |
| |
| |
| UdrDebug1(" A new connection %p is created", conn); |
| inUseConns_->insert(conn); |
| } |
| |
| UdrDebug0("[END ExUdrServer::getAnIpcConnection()]"); |
| |
| return conn; |
| } |
| |
| // Releases a connection back to freeConns_ for later use. |
| // |
| // Note: An opened connection will never be closed unless there is a |
| // problem accessing the server. This might be okay in most cases. |
| // But it's waste of resources in an app where many CALL stmts are opened |
| // at one point and does not do much with them later. |
| // Executor can have a model where the number of free conns are limited and |
| // a conn will be closed when it is being released if the free conn |
| // limit is reached. There are several ways to set this limit. One way |
| // is by way of session defaults. |
| void ExUdrServer::releaseConnection(IpcConnection *conn) |
| { |
| if (conn == NULL) |
| return; |
| |
| // Don't need to do anything for control connection since |
| // control connection is not added to these lists. |
| if (conn == getUdrControlConnection()) |
| return; |
| |
| if (! inUseConns_->remove(conn)) |
| { |
| // If UDR Server dies, we will have NULL control connection by |
| // the time we come here. In that case, we don't need to assert |
| // because 'conn' might be control connection |
| if (getUdrControlConnection()) |
| ex_assert(0, "A connection that is being released is not in use."); |
| } |
| |
| // Conn will be deleted in the following cases |
| // 1. conn got error |
| // 2. UDR Server is in error state because some other conn got error |
| // 3. UDR Server is restarted after an error |
| if (conn->getState() == IpcConnection::ERROR_STATE || |
| state_ == ExUdrServer::EX_UDR_BROKEN || |
| ! (conn->getOtherEnd() == serverProcessId_)) |
| delete conn; |
| else |
| freeConns_->insert(conn); |
| |
| return; |
| } |
| |
| NABoolean ExUdrServer::isIOPending(IpcConnection *conn) const |
| { |
| NABoolean result = FALSE; |
| if (ipcServer_ && conn) |
| { |
| NABoolean ioPendingOnConnection = |
| conn->sendIOPending() || conn->receiveIOPending(); |
| NABoolean anythingQueuedOnConnection = |
| (conn->numQueuedSendMessages() > 0) |
| || (conn->numQueuedReceiveMessages() > 0); |
| if (ioPendingOnConnection || anythingQueuedOnConnection || |
| conn->numReceiveCallbacksPending() > 0) |
| { |
| result = TRUE; |
| } |
| } |
| return result; |
| } |
| |
| void ExUdrServer::completeUdrRequests(IpcConnection *conn, |
| NABoolean waitForAllIO) const |
| { |
| #ifdef UDR_DEBUG |
| NABoolean firstTime = TRUE; |
| #endif // UDR_DEBUG |
| NABoolean done = FALSE; |
| |
| while (!done && isIOPending(conn)) |
| { |
| #ifdef UDR_DEBUG |
| if (firstTime) |
| { |
| firstTime = FALSE; |
| UdrDebug0("***"); |
| UdrDebug0("*** I/O is still pending on the UDR control connection"); |
| } |
| UdrDebug0("*** Waiting for one UDR I/O to complete..."); |
| #endif // UDR_DEBUG |
| |
| // Wait on 'conn' |
| conn->wait(IpcInfiniteTimeout); |
| UdrDebug0("*** A UDR I/O has completed."); |
| |
| if (!waitForAllIO) |
| { |
| done = TRUE; |
| } |
| } |
| |
| #ifdef UDR_DEBUG |
| if (!firstTime) |
| { |
| UdrDebug0("***"); |
| } |
| #endif // UDR_DEBUG |
| |
| } |
| |
| // Matchmaking logic to determine if this server has the requested |
| // attributes |
| NABoolean ExUdrServer::match(const Int32 &userId, |
| const char *options, |
| const char *optionDelimiters) const |
| { |
| // Two instances are considered a match if they have the same user |
| // identity and any of the following are true: |
| // - One or both instances have runtime options of "ANYTHING" |
| // - Both instances have runtime options of "OFF" |
| // - Both instances have matching runtime options and a matching |
| // option delimiter string |
| |
| if (userId_ != userId) |
| { |
| return FALSE; |
| } |
| |
| if (str_cmp_ne(options, "ANYTHING") == 0 || |
| str_cmp_ne(options_, "ANYTHING") == 0) |
| { |
| return TRUE; |
| } |
| |
| if (str_cmp_ne(options, "OFF") == 0 && |
| str_cmp_ne(options_, "OFF") == 0) |
| { |
| return TRUE; |
| } |
| |
| if (str_cmp_ne(options_, options) == 0 && |
| str_cmp_ne(optionDelimiters_, optionDelimiters) == 0) |
| { |
| return TRUE; |
| } |
| |
| return FALSE; |
| } |
| |
| // ----------------------------------------------------------------------- |
| // ExUdrServerManager |
| // ----------------------------------------------------------------------- |
| ExUdrServerManager::ExUdrServerManager(IpcEnvironment* env, |
| ComUInt32 maxServersPerGroup) |
| : ipcEnvironment_(env), |
| maxServersPerGroup_(maxServersPerGroup), |
| serverPool_(myIpcHeap()), |
| okToRetainOneServer_(FALSE) |
| #ifdef UDR_DEBUG |
| , traceFile_(NULL) |
| #endif |
| { |
| // |
| // Create the UDR server class. This does not actually |
| // start any processes. Use allocateServerProcess() to |
| // do that. |
| // |
| // This object is copied into each ExUdrServer object. |
| // ExUdrServer->start() method calls allocateServerProcess() |
| // on this object to create a process. |
| // |
| udrServerClass_ = new (myIpcHeap()) |
| IpcServerClass(ipcEnvironment_, IPC_SQLUDR_SERVER); |
| |
| #ifdef UDR_DEBUG |
| if ((getenv("UDR_""DEBUG") != NULL) || |
| (getenv("UDR_SERVER_MGR_DEBUG") != NULL)) |
| { |
| traceFile_ = stdout; |
| } |
| #endif |
| } |
| |
| ExUdrServerManager::~ExUdrServerManager() |
| { |
| // |
| // stop all the running ExUdrServer processes. |
| // |
| for (CollIndex i = 0; i < serverPool_.entries(); i++) |
| { |
| delete serverPool_[i]; |
| } |
| |
| if (udrServerClass_) |
| { |
| // |
| // class IpcServerClass does not have a destructor so to delete |
| // udrServerClass_ all we have to do is deallocate the memory from |
| // the IPC heap. |
| // |
| NADELETEBASIC(udrServerClass_, myIpcHeap()); |
| } |
| } |
| |
| // |
| // Returns an ExUdrServer with the requested attributes |
| // |
| // Creates a new ExUdrServer object if the number of existing servers |
| // with matching attributes is less than maximum number of servers |
| // allowed per group. Otherwise, we return one of the servers we |
| // already have with the lowest reference count. |
| // |
| // Successful completion of this method does not guarantee that the |
| // server process is actually started. |
| // |
| ExUdrServer* ExUdrServerManager::acquireUdrServer(const Int32 &userId, |
| const char *options, |
| const char *optionDelimiters, |
| const char *userName, |
| const char *userPassword, |
| NABoolean dedicated) |
| { |
| #ifdef UDR_DEBUG |
| if ((getenv("UDR_""DEBUG") != NULL) || |
| (getenv("UDR_SERVER_MGR_DEBUG") != NULL)) |
| { |
| traceFile_ = stdout; |
| } |
| else |
| { |
| traceFile_ = NULL; |
| } |
| #endif |
| |
| UdrDebug0("[BEGIN ExUdrServerManager::acquireUdrServer()]"); |
| UdrDebug1(" options: '%s'", options); |
| UdrDebug1(" delimiters: '%s'", optionDelimiters); |
| UdrDebug1(" Max servers per group: %u", getMaxServersPerGroup()); |
| |
| ExUdrServer *udrServer = NULL; |
| ComUInt32 lowestRefCnt = 0; |
| ComUInt32 numServersMatched = 0; |
| CollIndex entries = serverPool_.entries(); |
| |
| for (CollIndex i = 0; i < entries; i++) |
| { |
| ExUdrServer *s = serverPool_[i]; |
| if (s->match(userId, options, optionDelimiters) && |
| (!s->isDedicated())) |
| { |
| numServersMatched++; |
| if (udrServer == NULL || s->getRefCount() < lowestRefCnt) |
| { |
| udrServer = serverPool_[i]; |
| lowestRefCnt = udrServer->getRefCount(); |
| } |
| } |
| } |
| |
| UdrDebug2(" Found %u matching server%s", numServersMatched, |
| numServersMatched == 1 ? "" : "s"); |
| |
| if (entries > 0 && numServersMatched == 0) |
| { |
| UdrDebug0(" A new group is being encountered"); |
| UdrDebug0(" About to release idle servers..."); |
| |
| // We are seeing a server group for the first time, and it is not |
| // the only group we are currently managing. We will no longer |
| // treat this application as one that only requires a single UDR |
| // server. |
| okToRetainOneServer_ = FALSE; |
| |
| // Now release all idle servers (those with a ref count of |
| // zero). The loop here traverses the list in reverse order. |
| CollIndex idx = entries; |
| while (idx--) |
| { |
| ExUdrServer *curr = serverPool_[idx]; |
| if (curr->getRefCount() == 0) |
| { |
| delete curr; |
| serverPool_.removeAt(idx); |
| } |
| } |
| UdrDebug0(" Done releasing idle servers"); |
| } |
| |
| NABoolean reUseExistingServer = FALSE; |
| |
| if (numServersMatched >= getMaxServersPerGroup()) |
| { |
| // |
| // We hit our limit for servers in this group. We will return the |
| // one with the lowest reference count. |
| // |
| reUseExistingServer = TRUE; |
| } |
| |
| // If the request is for a dedicated server and the decision is to |
| // reuse an existing server, make sure the selected server's reference |
| // count is zero, else do not reuse this server. |
| if(dedicated && reUseExistingServer && |
| (udrServer->getRefCount() != 0)) |
| { |
| reUseExistingServer = FALSE; |
| } |
| |
| |
| if (reUseExistingServer) |
| { |
| // |
| // We hit our limit for servers in this group. We will return the |
| // one with the lowest reference count. |
| // |
| udrServer->incrRefCount(); |
| UdrDebug0(" No more servers can be started in this group"); |
| UdrDebug1(" Server %p will be reused", udrServer); |
| } |
| else |
| { |
| // If the requested options were "ANYTHING" then we are actually |
| // going to start a server as if the user specified "OFF". So |
| // that future requests for "OFF" will match this new server we |
| // are creating, we will change "ANYTHING" to "OFF" here. Note |
| // that the "ANYTHING" and "OFF" strings were uppercased by the |
| // SQL compiler when the plan was generated. |
| const char *newOptions = options; |
| if (str_cmp_ne(options, "ANYTHING") == 0) |
| { |
| newOptions = "OFF"; |
| } |
| |
| udrServer = new (myIpcHeap()) ExUdrServer(ipcEnvironment_, |
| userId, |
| newOptions, |
| optionDelimiters, |
| userName, |
| userPassword, |
| udrServerClass_); |
| #ifdef UDR_DEBUG |
| if (traceFile_) |
| { |
| udrServer->setTraceFile(traceFile_); |
| } |
| #endif |
| |
| UdrDebug1(" Created a new ExUdrServer instance %p", udrServer); |
| |
| udrServer->setRefCount(1); |
| serverPool_.insert(udrServer); |
| } |
| |
| UdrDebug2(" Returning ExUdrServer %p, ref count %u", |
| udrServer, udrServer->getRefCount()); |
| UdrDebug0("[END ExUdrServerManager::acquireUdrServer()]"); |
| |
| return udrServer; |
| } |
| |
| // |
| // A method to decrement the reference count on an ExUdrServer |
| // instance. Callers need to pass in a non-NULL argument because no |
| // checking is done here. When a reference count reaches zero we do |
| // one of two things: |
| // |
| // a) Retain the ExUdrServer instance as an idle server in our |
| // pool. This will be done as a performance optimization if the |
| // application has not yet made use of multiple UDR servers. |
| // |
| // b) Stop the server and remove it from the pool. We choose this |
| // option once we have detected that the application is using |
| // multiple UDR servers. |
| // |
| void ExUdrServerManager::releaseUdrServer(ExUdrServer *udrServer) |
| { |
| UdrDebug1("[BEGIN ExUdrServerManager::releaseUdrServer(%p)", udrServer); |
| |
| NABoolean found = FALSE; |
| for (CollIndex i = 0; !found && i < serverPool_.entries(); i++) |
| { |
| if (udrServer == serverPool_[i]) |
| { |
| found = TRUE; |
| udrServer->decrRefCount(); |
| udrServer->setDedicated(FALSE); |
| |
| UdrDebug2(" Found server %p. Ref count decremented to %u", |
| udrServer, udrServer->getRefCount()); |
| |
| if (udrServer->getRefCount() == 0) |
| { |
| if (okToRetainOneServer_ && serverPool_.entries() == 1) |
| { |
| // Do nothing. So far this application appears to only |
| // require a single UDR server so we will retain this one. |
| UdrDebug0(" There is one server in the pool, it will be retained"); |
| } |
| else |
| { |
| delete udrServer; |
| serverPool_.removeAt(i); |
| } |
| } |
| |
| } // if (udrServer == serverPool_[i]) |
| } // for each server in the pool |
| |
| UdrDebug1("[END ExUdrServerManager::releaseUdrServer(%p)]", udrServer); |
| } |
| |
