| /********************************************************************** |
| // @@@ 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 "SMConnection.h" |
| #include "ExSMGlobals.h" |
| #include "ExSMTask.h" |
| #include "ExSMReadyList.h" |
| #include "ExSMQueue.h" |
| #include "ExSMEvent.h" |
| #include "ExSMShortMessage.h" |
| #include "ex_exe_stmt_globals.h" |
| #include "ExpErrorEnums.h" |
| #include "NAAssert.h" |
| #include "PortProcessCalls.h" |
| |
| // ----------------------------------------------------------------------- |
| // Methods for class SMConnection |
| // ----------------------------------------------------------------------- |
| SMConnection::SMConnection(IpcEnvironment *env, |
| sm_target_t &smTarget, |
| UInt32 numReceiveBuffers, |
| IpcMessageObjSize maxBufSize, |
| ex_tcb *tcb, |
| ExMasterStmtGlobals *masterGlobals, |
| NABoolean isServer) |
| : IpcConnection(env, |
| IpcProcessId(), |
| isServer ? |
| eye_SEAMONSTER_CONNECTION_TO_CLIENT : |
| eye_SEAMONSTER_CONNECTION_TO_SERVER) |
| , scheduled_(0) |
| , isServer_(isServer) |
| , outstandingSMRequests_(0) |
| , numReceiveBuffers_(numReceiveBuffers) |
| , maxBufSize_(maxBufSize) |
| , dataStream_(NULL) |
| , cancelStream_(NULL) |
| , smTarget_(smTarget) |
| , smTask_(NULL) |
| , tcb_(tcb) |
| , smErrorNumber_(0) |
| , ccErrorNumber_(0) |
| , masterGlobals_(masterGlobals) |
| , sendCount_(0) |
| , recvCount_(0) |
| , postCount_(0) |
| , chunk_buffer_(NULL) |
| , chunk_nextOffset_(0) |
| , chunk_size_(0) |
| , chunk_waitingForAck_(false) |
| , intranode_(false) |
| { |
| memset(smErrorFunction_, 0, sizeof(smErrorFunction_)); |
| |
| ExSMGlobals *smGlobals = ExSMGlobals::GetExSMGlobals(); |
| assert(smGlobals); |
| |
| IpcEnvironment *ipcEnv = getEnvironment(); |
| assert(ipcEnv); |
| |
| // Create the SM target that will be stored in the task |
| // |
| // * Server-side: connection and task store the same target |
| // |
| // * Client-side: the task stores a modified target. The only change |
| // is to target.tag. In the task's tag, we set a bit to indicate |
| // that this task receives replies (as opposed to receiving |
| // requests). This bit is needed for multi-fragment where both |
| // endpoints of a logical connection can be in the same process. |
| sm_target_t targetForTask = smTarget_; |
| if (!isServer_) |
| ExSMTag_SetReplyFlag(&targetForTask.tag); |
| |
| // Create the task. We allocate an additional slot in the |
| // receive queue for a "larger than expected" reply buffer |
| uint32_t queueSize = numReceiveBuffers_ + 1; |
| smTask_ = smGlobals->addTask(targetForTask, |
| queueSize, |
| &scheduled_, |
| ipcEnv->getHeap(), |
| tcb_, |
| this); |
| assert(smTask_); |
| |
| EXSM_TRACE(EXSM_TRACE_INIT, "CONN CTOR %s %p task %p", |
| isServer_ ? "S" : "C", this, smTask_); |
| EXSM_TRACE(EXSM_TRACE_INIT, "CONN CTOR %s %p max buf %d", |
| isServer_ ? "S" : "C", this, (int) maxBufSize_); |
| |
| if (masterGlobals) |
| masterGlobals->addSMConnection(this); |
| |
| ipcEnv->getAllConnections()->incrNumSMConnections(); |
| |
| // When intranode_ is true, both ends of the connection are on the |
| // same seaquest node and there is no limit on the chunk size that |
| // can be sent. |
| int32_t myNode = smGlobals->getSQNodeNum(); |
| if (myNode == smTarget_.node) |
| intranode_ = true; |
| } |
| |
| SMConnection::~SMConnection() |
| { |
| EXSM_TRACE(EXSM_TRACE_INIT,"CONN DTOR %s %p %s", |
| isServer_ ? "S" : "C", this, |
| getConnectionStateString(getState())); |
| |
| ExSMGlobals *smGlobals = ExSMGlobals::GetExSMGlobals(); |
| assert(smGlobals); |
| |
| handleIOErrorForSM(); |
| |
| ExSMEvent::add(ExSMEvent::SMError, &smTarget_, |
| smErrorNumber_, ccErrorNumber_, 0, (int64_t) this); |
| |
| if (smTask_) |
| { |
| smGlobals->removeTask(smTask_); |
| smTask_ = NULL; |
| } |
| |
| if (masterGlobals_) |
| masterGlobals_->removeSMConnection(this); |
| |
| IpcMessageBuffer *receiveBuf; |
| while (receiveBuf = getNextReceiveQueueEntry()) |
| receiveBuf->callReceiveCallback(this); |
| |
| // Right now we do not have error handling in the SM connection or |
| // streams for fatal error scenarios. For example the master might |
| // get an IPC error on an ESP control connection and decide to run |
| // TCB destructors while the SM connections are expecting |
| // arrivals. For now we will simply CLOSE this connection which |
| // removes it from the pendingIOs_ list. |
| setState(CLOSED); |
| |
| getEnvironment()->getAllConnections()->decrNumSMConnections(); |
| } |
| |
| Int32 SMConnection::numQueuedSendMessages() |
| { |
| return sendQueueEntries(); |
| } |
| |
| Int32 SMConnection::numQueuedReceiveMessages() |
| { |
| Int32 result = 0; |
| if (smTask_) |
| { |
| ExSMQueue *q = smTask_->getOutQueue(); |
| if (q) |
| result = (Int32) q->getLength(); |
| } |
| return result; |
| } |
| |
| int32_t SMConnection::allocateReceiveBuffers() |
| { |
| ExSMQueue *inQueue = smTask_->getInQueue(); |
| |
| int count = 0; |
| int32_t rc = 0; |
| |
| while (inQueue->getLength() < numReceiveBuffers_) |
| { |
| assert(!inQueue->isFull()); |
| |
| IpcEnvironment *env = getEnvironment(); |
| NAMemory *heap = (env ? env->getHeap() : NULL); |
| |
| IpcMessageBuffer *msgBuf = |
| IpcMessageBuffer::allocate(maxBufSize_, NULL, heap, 0); |
| assert(msgBuf); |
| |
| ExSMQueue::Entry &entry = inQueue->getTailEntry(); |
| entry.setData((void *) msgBuf); |
| inQueue->insert(); |
| count++; |
| |
| sm_target_t target = smTarget_; |
| if (!isServer_) |
| ExSMTag_SetReplyFlag(&target.tag); |
| |
| UInt32 maxSMChunkSize = ExSM_GetMaxChunkSize(intranode_); |
| UInt32 bufferSize = msgBuf->getBufferLength(); |
| |
| if (bufferSize > maxSMChunkSize) |
| bufferSize = maxSMChunkSize; |
| |
| rc = ExSM_Post(target, |
| bufferSize, |
| msgBuf->data(), |
| smTask_, |
| isServer_); |
| |
| if (rc == SM_ERR_NOPEER || rc == SM_ERR_NOSERVICE || rc == SM_ERR_NODE_DOWN) |
| return rc; |
| |
| exsm_assert_rc(rc, "ExSM_Post"); |
| postCount_++; |
| } |
| |
| EXSM_TRACE(EXSM_TRACE_BUFFER,"CONN %p %s alloc recv bufs %d", this, |
| isServer_ ? "server" : "client", count); |
| return rc; |
| } |
| |
| void SMConnection::send(IpcMessageBuffer *msgBuf) |
| { |
| assert(msgBuf); |
| |
| EXSM_TRACE(EXSM_TRACE_BUFFER, "CONN %p BEGIN SEND %s state %s", this, |
| (isServer_ ? "S" : "C"), getConnectionStateString(getState())); |
| EXSM_TRACE(EXSM_TRACE_BUFFER, "CONN %p buf %p len %d ref %d", this, |
| msgBuf, (int) msgBuf->getMessageLength(), |
| (int) msgBuf->getRefCount()); |
| |
| if (!isServer_ && getState() == ERROR_STATE) |
| { |
| // The connection is in an error state so we do not attempt the |
| // send. We issue a send callback followed by a receive callback |
| // because data streams require both callbacks, even when errors |
| // are encountered, to perform their bookkeeping. |
| |
| // This call will put the connection on pendingIOs_ even though no |
| // I/O is being attempted. The connection will be removed from |
| // pendingIOs_ when the receive callback is processed. |
| incrOutstandingSMRequests(); |
| |
| // Perform the send callback |
| msgBuf->callSendCallback(this); |
| |
| // Perform the recieve callback and release the reference to the |
| // message buffer |
| msgBuf->callReceiveCallback(this); |
| msgBuf->decrRefCount(); |
| } |
| else |
| { |
| queueSendMessage(msgBuf); |
| while (tryToSendOneChunk()) |
| ; |
| } |
| |
| EXSM_TRACE(EXSM_TRACE_BUFFER, "CONN %p END SEND state %s", this, |
| getConnectionStateString(getState())); |
| } |
| |
| // Try to send one buffer or one chunk if the buffer size is too |
| // large. Return true if a buffer or chunk is sent. |
| // |
| // Nothing is sent and false is returned if the connection is waiting |
| // for an ack. |
| // |
| // Buffers are too large if either of the following is true: |
| // * Buffer size exceeds the expected size for this connection |
| // * Buffer size exceeds the maximum SM chunk size |
| // |
| // When we send chunks the protocol is: |
| // |
| // * First send a short message containing the buffer size and |
| // the chunk size, and return false |
| // * For each chunk |
| // * The function will eventually be called again when an ack |
| // arrives. |
| // * Send a chunk and return true |
| // |
| // Any time the function is called while waiting for an ack, do |
| // nothing and return false. |
| // |
| // The function will return true if a send or prepost attempt returns |
| // the NOPEER error. |
| // |
| bool SMConnection::tryToSendOneChunk() |
| { |
| EXSM_TRACE(EXSM_TRACE_BUFFER, |
| "CONN %p TRY wait %d arr %d q_sz %d", |
| this, (int) getWaitingForAck(), (int) getAckArrived(), |
| (int) sendQueueEntries()); |
| |
| int32_t rc = 0; |
| |
| // These local variables will be used for the SM send |
| IpcMessageBuffer *sendBuffer = NULL; |
| UInt32 sendOffset = 0; |
| UInt32 sendBytes = 0; |
| |
| // Keep track of whether we are sending a message in chunks, and |
| // whether we have reached the final chunk |
| bool sqlChunkMode = false; |
| bool finalSQLChunk = false; |
| |
| // We want to send different things depending on whether we are in |
| // chunk mode. In chunk mode, we want to send the next |
| // chunk. Otherwise we want to send the next buffer from the send |
| // queue. |
| // |
| // In chunk mode the WAITING FOR ACK property of the connection will |
| // be TRUE. It gets set to TRUE by the enterChunkMode() method and |
| // stays TRUE until chunk transmissions are complete. |
| |
| if (getWaitingForAck()) |
| { |
| // Note the fact that we are in chunk mode |
| sqlChunkMode = true; |
| |
| // See if the ACK ARRIVED property is set. The property is stored |
| // in the task object. The property is set by the reader thread |
| // calling task->sendChunk_SetAckArrived(true). The property is |
| // cleared below, inside the getAckArrived() block. |
| if (getAckArrived()) |
| { |
| // The connection was waiting for an ack and the ack |
| // arrived. The next send can take place. |
| |
| // Adjust local variables to drive the send. Since we are |
| // already in chunk mode, the data to send comes from the buffer |
| // chunk_buffer_ at offset chunk_nextOffset_. |
| // |
| // The number of bytes to send will be chunk_size_ unless this |
| // is the last chunk and there are not that many bytes remaining |
| // in chunk_buffer_. |
| |
| exsm_assert(chunk_buffer_, "ACK arrived but no buffer to send"); |
| UInt32 msgLen = (UInt32) chunk_buffer_->getMessageLength(); |
| sendBuffer = chunk_buffer_; |
| sendOffset = chunk_nextOffset_; |
| if (chunk_nextOffset_ + chunk_size_ < msgLen) |
| { |
| sendBytes = chunk_size_; |
| } |
| else |
| { |
| exsm_assert(chunk_nextOffset_ < msgLen, |
| "Chunk offset should not exceed message length"); |
| |
| sendBytes = (msgLen - chunk_nextOffset_); |
| finalSQLChunk = true; |
| } |
| |
| // Adjust data members in response to the ack arriving |
| // |
| // * Clear the ACK ARRIVED flag |
| // |
| // * For a non-final chunk, set the WAITING FOR ACK flag because |
| // this connection will require an ack before sending the next |
| // chunk |
| // |
| // * For a final chunk, clear the WAITING FOR ACK flag |
| // |
| setAckArrived(false, !finalSQLChunk); |
| |
| EXSM_TRACE(EXSM_TRACE_BUFFER, |
| "CONN %p ACK ARRIVED buf %p offset %d bytes %d final %d", |
| this, sendBuffer, (int) sendBytes, (int) sendOffset, |
| (int) finalSQLChunk); |
| |
| } // ack arrived |
| |
| else |
| { |
| // No ack has arrived so there is nothing to do. The sendBuffer |
| // variable is NULL which will be seen by code below and trigger |
| // an early return. |
| } |
| |
| } // waiting for an ack |
| |
| else |
| { |
| // Remove a buffer from the send queue |
| sendBuffer = getNextSendQueueEntry(); |
| |
| if (sendBuffer) |
| { |
| // maxSMChunkSize will be the largest physical chunk we can |
| // transmit over SM. For intranode messages there is no limit so |
| // maxSMChunkSize will be the total size of sendBuffer. |
| UInt32 msgLen = (UInt32) sendBuffer->getMessageLength(); |
| UInt32 maxSMChunkSize = ExSM_GetMaxChunkSize(intranode_); |
| |
| EXSM_TRACE(EXSM_TRACE_BUFFER, |
| "CONN %p dequeue %p len %d", |
| this, sendBuffer, (int) msgLen); |
| |
| // Cases to consider after a buffer is removed from the send |
| // queue: |
| // |
| // (a) The buffer exceeds SM maximum chunk size. It will be |
| // sent in multiple chunks. |
| // |
| // (b) The buffer does not exceed SM maximum chunk size but is |
| // larger than the receiving connection's max buffer size. The |
| // buffer can be sent in one chunk but first requires a short |
| // message round-trip interaction so that the receiving end can |
| // allocate a receive buffer of sufficient size. |
| // |
| // (c) The buffer does not exceed either limit and can be sent |
| // right away |
| |
| if (msgLen <= maxBufSize_ && msgLen <= maxSMChunkSize) |
| { |
| // Case (c) |
| |
| // The buffer can be sent right away. There is nothing to do |
| // here except adjust local variables that will drive the |
| // send. The send takes place later in this method. |
| sendOffset = 0; |
| sendBytes = msgLen; |
| } |
| else |
| { |
| // Cases (a) and (b) are treated the same |
| |
| // The buffer exceeds either the SM maximum size or the |
| // receiving connection's maximum size. A short message will |
| // be sent containing the buffer size and the chunk size. Data |
| // from the buffer will not be sent until an ack arrives. |
| |
| // Adjust data members to ENTER chunk mode |
| UInt32 sqlChunkSize = |
| (msgLen > maxSMChunkSize ? maxSMChunkSize : msgLen); |
| enterChunkMode(sendBuffer, sqlChunkSize); |
| |
| // Setting sendBuffer to NULL will trigger an early return |
| // after the short message is sent |
| sendBuffer = NULL; |
| |
| // We are about to send a short message containing the buffer |
| // size and the chosen chunk size. The tag needs the REPLY bit |
| // set. The message data region contains: |
| // * Message type |
| // * Buffer size |
| // * SQL chunk size |
| |
| sm_target_t targetToSend = smTarget_; |
| if (isServer_) |
| ExSMTag_SetReplyFlag(&targetToSend.tag); |
| |
| ExSMShortMessage m; |
| m.setTarget(targetToSend); |
| m.setNumValues(3); |
| m.setValue(0, (int32_t) ExSMShortMessage::SIZE); |
| m.setValue(1, (int32_t) msgLen); |
| m.setValue(2, (int32_t) sqlChunkSize); |
| m.send(); |
| if (rc == SM_ERR_NOPEER || rc == SM_ERR_NOSERVICE || rc == SM_ERR_NODE_DOWN) |
| { |
| if (!isServer_) |
| { |
| setErrorInfo(-1); |
| setState(ERROR_STATE); |
| smErrorNumber_ = rc; |
| strcpy(smErrorFunction_, "ExSM_SendShortMessage"); |
| } |
| else |
| NAExit(0); |
| } |
| exsm_assert_rc(rc, "ExSM_SendShortMessage"); |
| } // buffer is too large |
| } // if (sendBuffer) |
| } // not waiting for an ack |
| |
| // There is nothing to do if the connection is waiting on an ack or |
| // if there was nothing in the send queue. In either case the |
| // sendBuffer variable is NULL. |
| if (sendBuffer == NULL) |
| return false; |
| |
| // On the client side this adds the SM connection to |
| // pendingIOs_. On the server side the SM connection is added to |
| // pendingIOs_ as soon as the connection is created. |
| if (!isServer_) |
| incrOutstandingSMRequests(); |
| |
| // Before sending an SM message we must have pre-allocated receive |
| // buffers for all possible replies. Pre-allocated buffers allow the |
| // reader thread to fill target buffers without the overhead of |
| // allocating space on the fly. |
| rc = allocateReceiveBuffers(); |
| if (rc == SM_ERR_NOPEER || rc == SM_ERR_NOSERVICE || rc == SM_ERR_NODE_DOWN) |
| { |
| if (!isServer_) |
| { |
| smErrorNumber_ = rc; |
| strcpy(smErrorFunction_, "ExSM_Post"); |
| setErrorInfo(-1); |
| setState(ERROR_STATE); |
| |
| // Issue the send callback with the connection in the ERROR |
| // state. Note that smErrorNumber_ and smErrorFunction_ were |
| // set by allocateReceiveBuffers(). |
| sendBuffer->callSendCallback(this); |
| |
| sendBuffer->decrRefCount(); |
| return true; |
| } |
| else |
| NAExit(0); |
| } |
| |
| // The next step is to send a message |
| |
| ExSM_MessageType msgType = |
| (isServer_ ? EXSM_MSG_REPLY : EXSM_MSG_REQUEST); |
| ExSMGlobals *smGlobals = ExSMGlobals::GetExSMGlobals(); |
| |
| sm_target_t targetToSend = smTarget_; |
| if (sqlChunkMode) |
| ExSMTag_SetSQLChunkFlag(&targetToSend.tag); |
| if (isServer_) |
| ExSMTag_SetReplyFlag(&targetToSend.tag); |
| |
| bool messageWasSent = false; |
| rc = ExSM_Send(smGlobals, |
| targetToSend, |
| sendBuffer->data(sendOffset), |
| sendBytes, |
| msgType, |
| true, // isPrepostRequired |
| messageWasSent, // OUT |
| -1, // retry forever |
| sendCount_, |
| sendBuffer->data(0)); |
| |
| if (rc == SM_ERR_NOPEER || rc == SM_ERR_NOSERVICE || rc == SM_ERR_NODE_DOWN) |
| { |
| if (!isServer_) |
| { |
| setErrorInfo(-1); |
| setState(ERROR_STATE); |
| smErrorNumber_ = rc; |
| strcpy(smErrorFunction_, "ExSM_Send"); |
| } |
| else |
| NAExit(0); |
| } |
| else if (rc == SM_ERR_QUEUED) |
| { |
| // Add a reference to the send buffer because SM is holding a |
| // pointer to the buffer. This new reference will be released |
| // after SM returns the buffer to the reader thread. The reader |
| // will place the buffer on a list that the main thread cleans up |
| // periodically. |
| sendBuffer->incrRefCount(); |
| } |
| else |
| { |
| assert(messageWasSent && rc == 0); |
| } |
| |
| sendCount_++; |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL,"CONN %p send count %lld", this, |
| sendCount_); |
| |
| if (sqlChunkMode) |
| { |
| // Do not attempt to send more chunks if an error was encountered |
| if (getState() == ERROR_STATE) |
| finalSQLChunk = true; |
| |
| if (finalSQLChunk) |
| { |
| // Adjust data members to EXIT chunk mode |
| exitChunkMode(); |
| } |
| else |
| { |
| // A chunk was sent and it was not the final chunk. Adjust data |
| // members to move to the NEXT chunk. |
| moveToNextChunk(); |
| } |
| } |
| |
| // Invoke the send callback if this was not a chunked message or if |
| // it was the final chunk |
| if (!sqlChunkMode || finalSQLChunk) |
| { |
| // The lastSentBuffer_ data member is used inside some callback |
| // methods, for example to record the size of the message that was |
| // just sent in runtime stats. |
| lastSentBuffer_ = sendBuffer; |
| sendBuffer->callSendCallback(this); |
| lastSentBuffer_ = NULL; |
| |
| // Release the message buffer. The buffer was passed in from a |
| // stream but the stream no longer holds a reference. If the |
| // buffer was not queued by SM, this connection holds the only |
| // reference. If the buffer was queued, an additional reference |
| // was added above that the buffer will be released later. |
| sendBuffer->decrRefCount(); |
| } |
| |
| return true; |
| } |
| |
| // SMConnection::receive() |
| // |
| // SM connections are only used by buffered streams (subclasses of |
| // IpcBufferedMsgStream). Buffered streams call connection.receive() |
| // in the following cases: |
| // |
| // Client |
| // * IpcClientMsgStream::internalActOnSend() calls |
| // connection->receive() BEFORE issuing the send callback on the |
| // stream |
| // |
| // Server |
| // * IpcServerMsgStream::internalActOnReceive() calls |
| // connection->receive() AFTER issuing the receive callback on the |
| // stream |
| // * During the send bottom TCB constructor, connection.receive() is |
| // called once to put the connection into an initial receiving state |
| // |
| void SMConnection::receive(IpcMessageStreamBase *msgStream) |
| { |
| // if we are on server side adds SM connection to pendingIOs_ |
| // on the client side SM connection is added to pendingIOs when |
| // the first message is sent and the connection state is set to SENDING |
| if (isServer_) |
| setState(RECEIVING); |
| |
| // The call to addReceiveCallback(stream) will add the stream to the |
| // connection's recvStreams_ list. It does nothing else. |
| // recvStreams_ is used as buffers arrive to match buffers to the |
| // correct stream. The matchmaking happens in |
| // IpcConnection::getNextReceiveQueueEntry(). |
| addReceiveCallback(msgStream); |
| |
| // Make sure the maximum number of receive buffers are now |
| // pre-allocated. This is needed only for server-side connections |
| // because the server-side has to be ready to receive messages |
| // before doing any sends. In contract, a client-side connection can |
| // allocate receive buffers when a new request is sent to the |
| // server. |
| if (isServer_) |
| { |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL, |
| "CONN %p receive, allocating buffers", this); |
| int32_t rc = allocateReceiveBuffers(); |
| if (rc == SM_ERR_NOPEER || rc == SM_ERR_NOSERVICE || rc == SM_ERR_NODE_DOWN) |
| NAExit(0); |
| } |
| } |
| |
| WaitReturnStatus SMConnection::wait(IpcTimeout timeout, |
| UInt32 *eventConsumed, |
| IpcAwaitiox *ipcAwaitiox) |
| { |
| WaitReturnStatus waitReturnStatus; |
| |
| // Temporary code till Gil's changes are merged to send a message to master |
| // through control connection |
| ExSMGlobals *smGlobals = ExSMGlobals::GetExSMGlobals(); |
| if (smGlobals->getReaderThreadState() == ExSMGlobals::TERMINATED_DUE_TO_ERROR) |
| { |
| if (!isServer_) |
| { |
| if (getState() != ERROR_STATE) |
| { |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL,"CONN %p Reader thread terminated", |
| this); |
| |
| smErrorNumber_ = smGlobals->getReaderThreadSmErrorNumber(); |
| strcpy(smErrorFunction_, smGlobals->getReaderThreadSmErrorFunction()); |
| setErrorInfo(-1); |
| setState(ERROR_STATE); |
| } |
| } |
| else |
| { |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL, |
| "CONN %p Reader thread terminated, exiting", this); |
| NAExit(0); |
| } |
| } |
| |
| // $$$$ SEAMONSTER PROJECT JUNE 2013 |
| // We should be able to remove the scheduled_ data member now that |
| // we have a ready list for tasks with arrivals, and only call |
| // connection.wait() for a task that is on the ready list. |
| if (scheduled_) |
| waitReturnStatus = workOnArrivals(); |
| else |
| waitReturnStatus = WAIT_OK; |
| return waitReturnStatus; |
| } |
| |
| WaitReturnStatus SMConnection::workOnArrivals() |
| { |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL,"CONN %p BEGIN WORKONARRIVALS", this); |
| |
| while (getWaitingForAck() && getAckArrived()) |
| while (tryToSendOneChunk()) |
| ; |
| |
| scheduled_ = 0; |
| ExSMQueue *outQueue = smTask_->getOutQueue(); |
| |
| // Each iteration of the while loop will do the following |
| // * Remove an incoming buffer from the SM task queue |
| // * On the client side, give the buffer a message stream pointer |
| // (data or cancel) |
| // * Add the buffer to the connection's receive queue |
| // * On the server side, set the connection state to ESTABLISHED |
| // |
| // Following the while loop, a message stream receive callback is |
| // issued for each buffer on the connection's receive queue |
| |
| ExSMTaskList *smTaskList = ExSMGlobals::GetExSMGlobals()->getSMTaskList(); |
| ExSMReadyList *readyList = ExSMGlobals::GetExSMGlobals()->getReadyList(); |
| |
| while (!outQueue->isEmpty()) |
| { |
| ExSMQueue::Entry &entry = outQueue->getHeadEntry(); |
| IpcMessageBuffer *msgBuf = (IpcMessageBuffer *) entry.getData(); |
| bool locked = false; |
| |
| // We are about to remove a buffer from the task output queue |
| // |
| // If the task output queue is about to become empty (current |
| // length is 1 and we are about to remove an entry), perform the |
| // following steps: |
| // |
| // (a) Before removing the buffer, acquire the task list lock. |
| // This prevents the reader thread from putting the task back |
| // on the completed task queue. |
| // |
| // (b) After removing the buffer, if the output queue is still |
| // empty, remove the task from the completed task queue and |
| // release the task list lock. |
| |
| // Acquire the task list lock |
| if (outQueue->getLength() == 1) |
| { |
| smTaskList->lock(); |
| locked = true; |
| } |
| |
| // Remove a buffer from the task output queue |
| outQueue->removeHead(); |
| recvCount_++; |
| |
| // Remove the SM task from the ready list and release the lock. A |
| // task is added to the ready list by the reader thread when a |
| // buffer arrives for a task and that task currently has an empty |
| // output queue. |
| // |
| // NOTE: The SM ready list is accessed by both threads (main and |
| // reader) but does not have its own lock. By convention, |
| // modifications to the ready list are always performed while |
| // holding a lock on the SM task list (a global collection of all |
| // SM tasks in this process). |
| if (outQueue->isEmpty()) |
| { |
| assert(locked); |
| readyList->remove(smTask_); |
| } |
| |
| if (locked) |
| smTaskList->unlock(); |
| |
| // The next few lines look into the message and extract the type |
| // of the sending message stream. The stream type is used for |
| // tracing and also used later to differentiate data versus cancel |
| // messages. |
| InternalMsgHdrInfoStruct *infoStruct = (InternalMsgHdrInfoStruct *) |
| msgBuf->data(0); |
| ESPMessageTypeEnum streamType = |
| (ESPMessageTypeEnum) infoStruct->getType(); |
| |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL, |
| "CONN %p recv %p len %d tag %d %s", this, msgBuf, |
| (int) msgBuf->getMessageLength(), |
| (int) msgBuf->getReplyTag(), |
| getESPStreamTypeString(streamType)); |
| if (!isServer_) |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL, |
| "CONN %p lbatch %d seq curr %d incoming %d", this, |
| (int) infoStruct->getSMLastInBatch(), |
| (int) getReplySeqNum(), (int) infoStruct->getSeqNum()); |
| |
| // After a control connection error was reported we discard any |
| // late-arriving data buffers. This is because the method |
| // reportControlConnectionError() already issued receive callbacks |
| // for all pending arrivals and we do not want callbacks issued |
| // twice for the same arrival. |
| if (ccErrorNumber_ != 0) |
| { |
| msgBuf->decrRefCount(); |
| |
| // Return to the top of the "WHILE OUTQ IS NOT EMPTY" loop |
| continue; |
| } |
| |
| if (!isServer_) |
| { |
| // For client-side only. Add a pointer to the appropriate stream |
| // to the message buffer object. |
| // |
| // Note: we do not call decrOutstandingSMRequests() for a DATA |
| // reply as we do for a CANCEL reply. This is because data |
| // streams use a batch protocol and we do not want the decrement |
| // for every buffer. Only the last buffer in a batch should |
| // trigger the decrement. The stream receive callback (issued |
| // later in this method) takes care of decrementing the counter |
| // once a complete batch is seen. This happens inside |
| // IpcClientMsgStream::internalActOnReceive(). |
| switch (streamType) |
| { |
| case IPC_MSG_SQLESP_DATA_REPLY: |
| { |
| msgBuf->addCallback(dataStream_); |
| break; |
| } |
| case IPC_MSG_SQLESP_CANCEL_REPLY: |
| { |
| msgBuf->addCallback(cancelStream_); |
| decrOutstandingSMRequests(); |
| break; |
| } |
| default: |
| assert(0); |
| |
| } // switch (streamType) |
| } // if (!isServer_) |
| |
| // To allow this buffer to pass through an IpcClientMsgStream or |
| // IpcServerMsgStream it needs a valid reply tag. The tag is only |
| // checked for being valid, the actual value currently has no |
| // meaning for SM. |
| short replyTag = 1; |
| msgBuf->setReplyTag(replyTag); |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL,"CONN %p set reply tag %d", |
| this, (int) replyTag); |
| |
| queueReceiveMessage(msgBuf); |
| |
| // right for server, as the server sets it to SENDING |
| // again before receive call is done. |
| // But for client the the conn should not removed |
| // from the pendingIOs_ after the first receive but |
| // should be set to ESTABLISHED after all the expected |
| // number of replies are recieved or EOD is received. |
| if (isServer_) |
| setState(ESTABLISHED); |
| |
| } // while out queue is not empty |
| |
| // NOTE: getNextReceiveQueueEntry() has side effects and is not |
| // guaranteed to return a buffer just because the queue is |
| // non-empty. To fully understand the de-queueing step, you should |
| // read getNextReceiveQueueEntry(). The logic there includes: |
| // * special cases for server-side and the ERROR state |
| // * matching of reply sequence numbers |
| // * if the incoming buffer has a stream pointer, matching of that |
| // pointer with an element of recvStreams_ |
| // * removal from receiveQueue_ and recvStreams_ |
| // * updating the buffer's stream pointer |
| // * increment of replySeqNum_ |
| |
| IpcMessageBuffer *receiveBuf; |
| while ((receiveBuf = getNextReceiveQueueEntry())) |
| receiveBuf->callReceiveCallback(this); |
| |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL,"CONN %p END WORKONARRIVALS", this); |
| return WAIT_OK; |
| } |
| |
| void SMConnection::setState(IpcConnectionState s) |
| { |
| IpcConnectionState oldState = getState(); |
| if (oldState != s) |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL,"CONN %p %s -> %s", this, |
| getConnectionStateString(oldState), |
| getConnectionStateString(s)); |
| IpcConnection::setState(s); |
| } |
| |
| void SMConnection::incrOutstandingSMRequests() |
| { |
| outstandingSMRequests_++; |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL,"CONN %p outstanding reqs %d", this, |
| (int) outstandingSMRequests_); |
| if (outstandingSMRequests_ == 1) |
| setState(SENDING); |
| } |
| |
| void SMConnection::decrOutstandingSMRequests() |
| { |
| assert(outstandingSMRequests_ > 0); |
| outstandingSMRequests_--; |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL,"CONN %p outstanding reqs %d", this, |
| (int) outstandingSMRequests_); |
| |
| if (outstandingSMRequests_ == 0) |
| setState(ESTABLISHED); |
| } |
| |
| // SMConnection::reportControlConnectionError() |
| // |
| // ExMasterEspMessage::actOnErrorConnection calls this function after |
| // an error is encountered on an ESP control connection and we want to |
| // propagate the error to SM connections. |
| // |
| // The method is only called in the master executor and is therefore a |
| // no-op for server-side connections. The method is also a no-op if |
| // the connection already encountered errors. |
| void SMConnection::reportControlConnectionError(GuaErrorNumber error) |
| { |
| if (isServer_) |
| return; |
| |
| // When this function is called for the first time, store the error |
| // number in a field specific to control connection errors |
| if (ccErrorNumber_ == 0) |
| { |
| ccErrorNumber_ = error; |
| |
| ExSMEvent::add(ExSMEvent::ControlConnectionError, &smTarget_, |
| ccErrorNumber_, smErrorNumber_, 0, (int64_t) this); |
| EXSM_TRACE(EXSM_TRACE_PROTOCOL,"CONN %p reportControlConnectionError %d", |
| this, (int) error); |
| |
| } |
| |
| // Also store the error in the general-purpose field smErrorNumber_ |
| if (smErrorNumber_ == 0) |
| smErrorNumber_ = error; |
| |
| setErrorInfo(-1); |
| setState(ERROR_STATE); |
| |
| handleIOErrorForSM(); |
| |
| // The code above that processes pending responses may have set the |
| // state of the connection to ESTABLISHED as a way to remove the |
| // connection from the global pendingIOs_ list. It would nice to |
| // avoid the transition to ESTABLISHED but for now that is not being |
| // attempted. Instead we simply make sure that the connection is in |
| // an ERROR state by the time this method returns. |
| setState(ERROR_STATE); |
| } |
| |
| void SMConnection::handleIOErrorForSM() |
| { |
| |
| IpcMessageBuffer *sendBuf; |
| // might need to decrement the ref count of the chunk mode buffer, |
| // otherwise it might be a memory leak |
| while (sendBuf = removeNextSendBuffer()) |
| sendBuf->decrRefCount(); |
| |
| |
| IpcMessageBuffer *receiveBuf; |
| while (receiveBuf = removeNextReceiveBuffer()) |
| getEnvironment()->getAllConnections()->bumpCompletionCount(); |
| |
| removeReceiveStreams(); |
| |
| if (dataStream_) |
| { |
| IpcClientMsgStream *dataStream = (IpcClientMsgStream *)dataStream_; |
| |
| for (Lng32 i=0; i < dataStream->numOfResponsesPending(); i++) |
| { |
| decrOutstandingSMRequests(); |
| dataStream->setSMBatchIsComplete(TRUE); |
| dataStream->internalActOnReceive(NULL, this); |
| } |
| } |
| |
| if (cancelStream_) |
| { |
| IpcClientMsgStream *cancelStream = (IpcClientMsgStream *)cancelStream_; |
| |
| for (Lng32 i=0; i < cancelStream->numOfResponsesPending(); i++) |
| { |
| decrOutstandingSMRequests(); |
| cancelStream->internalActOnReceive(NULL, this); |
| } |
| } |
| |
| } |
| |
| // SMConnection::populateDiagsArea() |
| // |
| // This method is a pure virtual function in the IpcConnection parent |
| // class. |
| // |
| // The method gets called by stream objects when a send or receive |
| // callback is issued and the connection has entered an ERROR |
| // state. The stream caller provides a diags area in which error |
| // conditions are written. |
| // |
| // The method is a no-op if the connection is not in an ERROR state. |
| // |
| void SMConnection::populateDiagsArea(ComDiagsArea *&diags, |
| CollHeap *diagsHeap) |
| { |
| if (smErrorNumber_ == GuaOK) |
| return; |
| |
| // Cases to consider |
| // |
| // a. An SM NOPEER error was encountered. This means the process at |
| // the other end of an SM connection went away unexpectedly. |
| // Generate a descriptive error. |
| // |
| // b. Otherwise all other errors are assumed to be Guardian errors |
| // on an ESP control connection. This assumption might also change |
| // in the future. For this case do not generate new diagnostics if |
| // the diags area already contains the error. |
| |
| // Case (a) |
| if (smErrorNumber_ == SM_ERR_NOPEER || |
| smErrorNumber_ == SM_ERR_NOSERVICE || |
| smErrorNumber_ == SM_ERR_NODE_DOWN) |
| { |
| IpcAllocateDiagsArea(diags, diagsHeap); |
| |
| char processName[12]; |
| sprintf(processName, "%d,%d", smTarget_.node, smTarget_.pid); |
| |
| *diags << DgSqlCode(-EXE_SM_FUNCTION_ERROR) |
| << DgString0(smErrorFunction_) |
| << DgInt0((Lng32) smErrorNumber_) |
| << DgInt1((Lng32) smTarget_.pid) |
| << DgString1(processName) |
| << DgNskCode((Lng32) 10000 + abs(smErrorNumber_)); |
| } |
| |
| // Case (b) |
| else |
| { |
| if (diags && diags->contains(-EXE_SM_CONTROL_CONN_ERROR)) |
| return; |
| |
| IpcAllocateDiagsArea(diags, diagsHeap); |
| |
| char targetBuf[128]; |
| sprintf(targetBuf, "%d:%d:%" PRId64 ":%d:0x%c", |
| (int) smTarget_.node, (int) smTarget_.pid, smTarget_.id, |
| (int) ExSMTag_GetTagWithoutQualifier(smTarget_.tag), |
| (char) ExSMTag_GetQualifierDisplay(smTarget_.tag)); |
| |
| *diags << DgSqlCode(-EXE_SM_CONTROL_CONN_ERROR) |
| << DgString1(targetBuf) |
| << DgInt0(smErrorNumber_); |
| |
| getEnvironment()->getMyOwnProcessId(IPC_DOM_GUA_PHANDLE). |
| addProcIdToDiagsArea(*diags,0); |
| } |
| } |
| |
| // Several data members must be maintained and kept in sync as the |
| // connection begins chunk mode, moves to the next chunk, and ends |
| // chunk mode. The following methods encapsuate the ENTER, NEXT, and |
| // EXIT operations. |
| void SMConnection::enterChunkMode(IpcMessageBuffer *sendBuffer, |
| UInt32 sqlChunkSize) |
| { |
| chunk_buffer_ = sendBuffer; |
| chunk_nextOffset_ = 0; |
| chunk_size_ = sqlChunkSize; |
| setAckArrived(false, // Clear the ACK ARRIVED flag |
| true); // Set the WAITING FOR ACK flag |
| |
| EXSM_TRACE(EXSM_TRACE_BUFFER, "CONN %p ENTER CHUNK buf %p sz %d", this, |
| sendBuffer, (int) sqlChunkSize); |
| } |
| |
| void SMConnection::moveToNextChunk() |
| { |
| chunk_nextOffset_ += chunk_size_; |
| |
| EXSM_TRACE(EXSM_TRACE_BUFFER, "CONN %p NEXT CHUNK offset %d", this, |
| (int) chunk_nextOffset_); |
| } |
| |
| void SMConnection::exitChunkMode() |
| { |
| chunk_buffer_ = NULL; |
| chunk_nextOffset_ = 0; |
| chunk_size_ = 0; |
| setAckArrived(false, // Clear the ACK ARRIVED flag |
| false); // Clear the WAITING FOR ACK flag |
| |
| EXSM_TRACE(EXSM_TRACE_BUFFER, "CONN %p EXIT CHUNK", this); |
| } |
| |
| // The following two methods manage the ACK ARRIVED attribute of this |
| // connection. The attribute is actually stored in the task object |
| // which makes the attribute visible and writeable by the reader |
| // thread. |
| void SMConnection::setAckArrived(bool arrived, bool waitForAnother) |
| { |
| // Set the ACK ARRIVED flag in the task object |
| if (smTask_) |
| smTask_->sendChunk_SetAckArrived(arrived); |
| |
| // Set the WAITING FOR ACK flag in the connection |
| chunk_waitingForAck_ = waitForAnother; |
| } |
| |
| bool SMConnection::getAckArrived() |
| { |
| bool result = false; |
| if (smTask_) |
| result = smTask_->sendChunk_GetAckArrived(); |
| return result; |
| } |
| |
| void SMConnection::dumpAndStopOtherEnd(bool doDump, bool doStop) const |
| { |
| char coreFile[1024]; |
| char processName[MS_MON_MAX_PROCESS_NAME+1]; |
| char seqName[PhandleStringLen]; |
| if (XZFIL_ERR_OK == msg_mon_get_process_name2(smTarget_.node, |
| smTarget_.pid, |
| smTarget_.verifier, |
| processName)) |
| { |
| |
| sprintf(seqName, "%s:%d", processName, smTarget_.verifier); |
| |
| if (doDump) |
| msg_mon_dump_process_name(NULL, seqName, coreFile); |
| |
| if (doStop) |
| msg_mon_stop_process_name(seqName); |
| } |
| } |
