| /********************************************************************** |
| // @@@ 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 <unistd.h> |
| #include <time.h> |
| #include "Platform.h" |
| #include "seabed/pctl.h" |
| #include "seabed/pevents.h" |
| #include "ExSMReader.h" |
| #include "ExSMCommon.h" |
| #include "ExSMTrace.h" |
| #include "ExSMGlobals.h" |
| #include "ExSMTask.h" |
| #include "ExSMTaskList.h" |
| #include "ExSMReadyList.h" |
| #include "ExSMQueue.h" |
| #include "ExSMEvent.h" |
| #include "ExSMShortMessage.h" |
| #include "NAAssert.h" |
| |
| void *ExSM_ReaderFunction(void *arg) |
| { |
| int32_t rc = NAAssertMutexCreate(); |
| if (rc < 2) // Main executor thread must precede this thread |
| abort(); |
| |
| ExSMGlobals *glob = (ExSMGlobals *) arg; |
| exsm_assert(glob, "Invalid SM globals pointer"); |
| |
| // Initialize the in-memory trace |
| ExSMEvent::initReaderThread(); |
| |
| // Settings for tracing to a file |
| UInt32 tLevel = glob->getTraceLevel(); |
| const char * tPrefix = glob->getTraceFilePrefix(); |
| ExSM_SetTraceLevel(tLevel); |
| ExSM_SetTraceEnabled(glob->getTraceEnabled(), glob); |
| |
| pid_t mainThreadPID = glob->getMainThreadPID(); |
| bool isMaster = glob->isMasterExecutor(); |
| |
| char readerPrefix[20]; |
| sprintf(readerPrefix, "[R%d]", (int) mainThreadPID); |
| |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_SM_CALLS, |
| "%s BEGIN READER (%s %d)", readerPrefix, |
| isMaster ? "MASTER" : "ESP", (int) mainThreadPID); |
| |
| ExSMTaskList *smTaskList = glob->getSMTaskList(); |
| ExSMReadyList *readyList = glob->getReadyList(); |
| int64_t loopCount = 0; |
| |
| // retryCount is the number of times SM_get has been called and |
| // returned zero chunks. retryCount is reset to 0 every time SM_get |
| // returns one or more chunks. |
| int32_t retryCount = 0; |
| |
| glob->setReaderThreadState(ExSMGlobals::STARTED, |
| true, // do nocking |
| true); // signal |
| |
| while (glob->getReaderThreadState() == ExSMGlobals::STARTED) |
| { |
| loopCount++; |
| |
| // If tracing got turned on via set session default, the main |
| // thread learns about it at fixup time and the reader thread will |
| // learn about it the second time through this loop after fixup, |
| // since the trace level changed. |
| if ((tLevel != glob->getTraceLevel()) || |
| (strcmp(tPrefix, glob->getTraceFilePrefix()))) |
| { |
| tLevel = glob->getTraceLevel(); |
| tPrefix = glob->getTraceFilePrefix(); |
| ExSM_SetTraceLevel(tLevel); |
| ExSM_SetTraceEnabled(glob->getTraceEnabled(), glob); |
| } |
| |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT, |
| "%s LOOP %ld", readerPrefix, (long) loopCount); |
| |
| sm_chunk_t *chunks = NULL; |
| int32_t nchunks = 0; |
| sm_handle_t dataHandle = 0; |
| |
| #ifndef NDEBUG |
| // In the debug build we can delay the first call to |
| // SM_get. This simulates "slow reap" of the server-side |
| // prepost. In response to the slow reap, the main thread should |
| // see ERR_PREPOST errors when it sends and retry the send. |
| if (loopCount == 1) |
| { |
| const char *envvar = getenv("EXSM_WAIT_DELAY"); |
| if (envvar && *envvar) |
| { |
| Int32 seconds = atoi(envvar); |
| if (seconds > 0) |
| { |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT, |
| "EXSM_WAIT_DELAY %d\n", (int) seconds); |
| sleep(seconds); |
| } |
| } |
| } |
| #endif |
| |
| // Wait for arrivals |
| rc = SM_get(0, &chunks, &nchunks, &dataHandle); |
| |
| // Add a RECEIVE completion event to the in-memory trace |
| ExSMEvent::add(ExSMEvent::Receive, NULL, rc); |
| |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT|EXSM_TRACE_SM_CALLS, |
| "%s wait rc %d n %d h %p", readerPrefix, |
| (int) rc, (int) nchunks, (void *) dataHandle); |
| |
| if (rc == SM_ERR_NOSERVICE) |
| { |
| glob->handleReaderThreadError(rc, "SM_get", dataHandle); |
| return 0; |
| } |
| |
| // We want to process the chunk array if rc is zero or rc |
| // indicates that error codes are in individual chunks. Otherwise |
| // assert. |
| if (rc != SM_ERR_INCHUNK) |
| exsm_assert_rc(rc, "SM_get"); |
| |
| // Go back to the top of the main loop if no chunks were returned |
| if (chunks == NULL || nchunks < 1) |
| { |
| retryCount++; |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT, |
| "%s wait will be retried, count %d", readerPrefix, |
| (int) retryCount); |
| continue; |
| } |
| |
| // If we reach here, SM_get returned a chunk array for |
| // processing. Reset the retry count and process the arrivals. |
| retryCount = 0; |
| |
| // Lock the task list |
| smTaskList->lock(); |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT, |
| "%s locked task list, tasks %d", readerPrefix, |
| (int) smTaskList->getNumTasks()); |
| |
| // A container for short messages |
| ExSMShortMessage shortMsg; |
| ExSMShortMessage::MsgType shortMsgType = ExSMShortMessage::UNKNOWN; |
| |
| // Process each arrival |
| for (int32_t i = 0; i < nchunks; i++) |
| { |
| sm_chunk_t &chunk = chunks[i]; |
| |
| // Add a CHUNK event to the in-memory trace to show that we |
| // are processing a new chunk. |
| // |
| // The fourth integer provided to ExSMEvent::add() will be one |
| // of the following: |
| // * Short message with length >= 4: first four data bytes |
| // * Otherwise: chunk.buff |
| // |
| // Short message data bytes are included in the trace event |
| // because that is where we store the message type. |
| // |
| // NOTE: We read the first four data bytes ONLY for short |
| // messages. For prepost messages we should not access |
| // chunk.buff until after we verify the task exists and the |
| // query ID is valid. If the query ID is not valid, that means |
| // the main thread already called SM cancel for and may have |
| // already deleted the buffer pointed to by chunk.buff. |
| int64_t fourthArg = (int64_t) chunk.buff; |
| |
| if ((chunk.flags & SM_FLAG_PREPOST) == 0) |
| { |
| int32_t firstFourBytes = 0; |
| if ((chunk.buff && chunk.size >= 4)) |
| memcpy(&firstFourBytes, chunk.buff, 4); |
| fourthArg = (int64_t) firstFourBytes; |
| } |
| |
| ExSMEvent *event = ExSMEvent::add(ExSMEvent::Chunk, &chunk.tgt, |
| chunk.errcode, |
| chunk.size, |
| chunk.flags, |
| fourthArg); |
| |
| // Is the chunk a request or reply |
| bool isRequest = !ExSMTag_GetReplyFlag(chunk.tgt.tag); |
| |
| // Retrieve the global request or reply counter from SM globals |
| uint32_t count = 0; |
| if ((chunk.flags & SM_FLAG_PREPOST) == 0) |
| count = glob->incrRecvShortCount(); |
| else if (isRequest) |
| count = glob->incrRecvRequestCount(); |
| else |
| count = glob->incrRecvReplyCount(); |
| |
| // Add chunk information to the trace file |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT|EXSM_TRACE_SM_CALLS, |
| "%s %s %d src %d:%d:%" PRId64 ":%d:0x%c fl 0x%x", |
| readerPrefix, |
| (isRequest ? "REQUEST" : "REPLY"), (int) count, |
| (int) chunk.tgt.node, (int) chunk.tgt.pid, chunk.tgt.id, |
| (int) ExSMTag_GetTagWithoutQualifier(chunk.tgt.tag), |
| (char) ExSMTag_GetQualifierDisplay(chunk.tgt.tag), |
| (int) chunk.flags); |
| |
| char buf[30]; |
| SM_strerror_r(chunk.errcode, buf, sizeof(buf)); |
| |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT|EXSM_TRACE_SM_CALLS, |
| "%s sz %d err %s h %p buf %p", readerPrefix, |
| (int) chunk.size, buf, |
| chunk.handle, chunk.buff); |
| |
| // Add chunk size to the trace file if this is one fragment of a |
| // SQL data buffer |
| if (ExSMTag_GetSQLChunkFlag(chunk.tgt.tag)) |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_BUFFER, |
| "%s RECV CHUNK %d", readerPrefix, (int) chunk.size); |
| |
| if (chunk.errcode != 0) |
| { |
| // Cases to consider |
| // |
| // If chunk.errcode is LOCAL_TARGET (meaning a chunk arrived |
| // but the ID was already cancelled): |
| // |
| // (a) The ID in the chunk is not in the global collection |
| // of active IDs. Treat this as a late arrival for a |
| // query that already ended. Discard the chunk. |
| // |
| // If a garbage ID value arrives we will most likely |
| // discard the chunk rather than reporting an |
| // error. Currently we are willing to live with this |
| // risk. |
| // |
| // (b) The ID is active. This is an internal error. |
| // |
| // Otherwise |
| // |
| // (c) This is an internal error |
| |
| if (chunk.errcode == SM_ERR_LOCAL_TARGET) |
| { |
| int64_t smQueryID = chunk.tgt.id; |
| bool found = ExSM_FindID(smQueryID); |
| if (!found) |
| { |
| // Case (a) |
| ExSMEvent::add(ExSMEvent::IDNotActive, &chunk.tgt, |
| chunk.errcode, chunk.flags); |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_SM_CALLS, |
| "ID is not active, LOCAL_TARGET error will be ignored"); |
| continue; |
| } |
| } |
| else if (chunk.errcode == SM_ERR_NODE_DOWN || |
| (chunk.errcode == SM_ERR_PREPOST) || |
| (chunk.errcode == SM_ERR_NOSERVICE)) |
| { |
| glob->handleReaderThreadError(chunk.errcode, "SM_get", dataHandle, smTaskList); |
| return 0; |
| } |
| |
| // Cases (b) and (c) |
| char buf[64]; |
| char buf2[30]; |
| SM_strerror_r(chunk.errcode, buf2, sizeof(buf)); |
| snprintf(buf, sizeof(buf), "chunk.errcode == %s", buf2); |
| exsm_assert(FALSE, buf); |
| } |
| |
| if (chunk.flags & SM_FLAG_SENDNOTIFICATION) |
| { |
| // SM is returning notification of a completed send. In |
| // response the reader thread places the send buffer on a |
| // global queue which the main thread cleans up periodically. |
| // |
| // The cleanup happens in |
| // IpcEnvironment::deleteCompletedMessages() |
| |
| exsm_assert(chunk.buff, "QUEUED event arrived but buffer is NULL"); |
| exsm_assert(chunk.handle, "QUEUED event arrived but handle is NULL"); |
| |
| if (event) |
| event->setOptional5((int64_t)chunk.handle); |
| |
| void *bufferAddr = ExSM_GetMessageBufferAddr(chunk.handle); |
| ExSM_AddCompletedSendBuffer(bufferAddr); |
| |
| |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_SM_CALLS, |
| "COMPLETED SEND data %p msg %p\n", |
| chunk.buff, chunk.handle); |
| |
| continue; |
| } |
| |
| // Verify that the chunk has a data region. Currently there is |
| // always a data region even for short messages, because the |
| // short message type travels as data and every short message |
| // must have a type. |
| exsm_assert(chunk.buff, "SM chunk arrived, invalid buff pointer"); |
| |
| // The target we use for task lookup is not always the target |
| // that just arrived. Today the only exception is: |
| // * For a prepost message that is a "large buffer" the tag for |
| // lookup is the sender's tag with the LARGE_MSG flag cleared |
| sm_target_t targetForTaskLookup = chunk.tgt; |
| |
| // Is the chunk a fragment of a SQL data buffer. This variable |
| // is initially false but can become true later when we look at |
| // flags inside the chunk. |
| bool receivedSQLBufferFragment = false; |
| |
| if ((chunk.flags & SM_FLAG_PREPOST) == 0) |
| { |
| // This is a short message |
| |
| // Initialize the short message container (a local variable) |
| // from the chunk that just arrived. |
| shortMsg.receive(chunk); |
| shortMsgType = (ExSMShortMessage::MsgType) shortMsg.getValue(0); |
| |
| if (EXSM_TRACE_ENABLED) |
| shortMsg.writeToTrace(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT| |
| EXSM_TRACE_SM_CALLS, |
| readerPrefix, " "); |
| |
| // Certain short messages require no further |
| // processing. Examples are the SHUTDOWN and FIXUP_REPLY |
| // messages. We do not need to locate a task object and |
| // execute logic specific to that task. All we need to do is |
| // change state variables and return to the top of the loop |
| // that processes each individual chunk. |
| |
| if (shortMsgType == ExSMShortMessage::SHUTDOWN) |
| { |
| // When a SHUTDOWN message is received: |
| // * return to the top of the chunk processing loop |
| // * following the chunk loop, return to the top of the main loop |
| // * the SHUTDOWN state will be noticed and the main loop |
| // will terminate |
| // * following the main loop, the reader thread will exit |
| |
| glob->setReaderThreadState(ExSMGlobals::PROCESSING_SHUTDOWN); |
| continue; |
| } |
| |
| else if (shortMsgType == ExSMShortMessage::FIXUP_REPLY) |
| { |
| // One FIXUP_REPLY is sent from ESP to master each time an |
| // ESP successfully fixes up a fragment. There is no |
| // interesting content in these messages. But they need to |
| // flow from ESP to master as an SM "go message". |
| // |
| // Without these replies flowing from ESP to master, if the |
| // master were to send a data request to the ESP, the |
| // receiving SM service might report that ESP preposts are |
| // not yet available. |
| // |
| // When a FIXUP_REPLY is received: |
| // * Increment a global counter. The master is watching this |
| // counter and when the counter reaches a certain value, |
| // the master assumes all ESPs are ready for SM data |
| // requests. |
| glob->incrFixupReplyCount(); |
| EXSM_TRACE(EXSM_TRACE_RDR_THR, |
| "%s Fixup reply count %d", readerPrefix, |
| (int) glob->getFixupReplyCount()); |
| continue; |
| } |
| } |
| else |
| { |
| // This is a prepost message |
| |
| // If this is one fragment of a SQL buffer, note that fact and |
| // adjust the target for task lookup |
| if (ExSMTag_GetSQLChunkFlag(chunk.tgt.tag)) |
| { |
| ExSMTag_ClearSQLChunkFlag(&targetForTaskLookup.tag); |
| receivedSQLBufferFragment = true; |
| } |
| } |
| |
| // Find the task |
| bool doLock = false; |
| bool doTrace = false; |
| ExSMTask *task = smTaskList->findTask(targetForTaskLookup, |
| doLock, |
| doTrace); |
| if (task == NULL) |
| { |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT, |
| "%s *** ERROR: SM TASK NOT FOUND", readerPrefix); |
| |
| // Perform the lookup again with tracing enabled. This lookup |
| // is expected to fail just like the first lookup failed. |
| doTrace = true; |
| task = smTaskList->findTask(targetForTaskLookup, |
| doLock, |
| doTrace); |
| |
| exsm_assert(task == NULL, |
| "Task lookup should not succeed after first lookup failed"); |
| |
| // If the ID is active, this is an unexpected arrival and we |
| // will assert. If the ID is not active, we assume this is a |
| // late arrival for a query that ended and silently ignore the |
| // chunk. |
| int64_t smQueryID = chunk.tgt.id; |
| bool found = ExSM_FindID(smQueryID); |
| if (!found) |
| { |
| ExSMEvent::add(ExSMEvent::IDNotActive, &chunk.tgt, |
| chunk.errcode, chunk.flags); |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_SM_CALLS, |
| "ID is not active, chunk will be ignored"); |
| continue; |
| } |
| |
| } // if (task == NULL) |
| |
| exsm_assert(task, "SM task not found"); |
| |
| if ((chunk.flags & SM_FLAG_PREPOST) != 0) |
| { |
| // This is a prepost message. Print the message type to the |
| // trace file. |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT| |
| EXSM_TRACE_SM_CALLS, |
| "%s msg type %s", readerPrefix, |
| ExSM_GetMessageBufferType(chunk.buff, chunk.size)); |
| } |
| |
| task->incrReceiveCount(); |
| |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT, |
| "%s tcb %p recv cnt %d", readerPrefix, |
| task->getTCB(), (int) task->getReceiveCount()); |
| |
| if ((chunk.flags & SM_FLAG_PREPOST) == 0) |
| { |
| // This is a short message. The short message contents have |
| // already been copied into the shortMsg variable. We now |
| // check the message type and process accordingly. |
| |
| // BUFFER SIZE |
| if (shortMsgType == ExSMShortMessage::SIZE) |
| { |
| // Message size and chunk size are the second and third |
| // values in the short message. The collection of values is |
| // 0-based so we need to extract slots 1 and 2. |
| uint32_t msgSize = (uint32_t) shortMsg.getValue(1); |
| uint32_t chunkSize = (uint32_t) shortMsg.getValue(2); |
| |
| void *largeBuffer = |
| ExSM_AllocateMessageBuffer(msgSize, glob->getThreadSafeHeap()); |
| exsm_assert(largeBuffer, "Allocation failed for largeBuffer"); |
| |
| task->recvChunk_Enter(largeBuffer, msgSize, chunkSize); |
| |
| // Next step is to post the receive buffer. The target for |
| // this post is the same as the sender's target with the |
| // LARGE MSG flag set in the tag. |
| |
| // Reasons for setting the LARGE MSG flag: |
| // * If we did not set the flag, the new post would be |
| // queued behind other buffers already posted for the |
| // same target |
| // * We don't want this post at the back of the queue |
| // * We set a flag in the tag so this post is seen at the |
| // front of a different queue |
| |
| sm_target_t target = chunk.tgt; |
| ExSMTag_SetSQLChunkFlag(&target.tag); |
| |
| char *chunkAddr = (char *) task->recvChunk_GetPrepostAddr(); |
| |
| int32_t rc = ExSM_Post(target, chunkSize, chunkAddr, task, 0); |
| |
| if (rc == SM_ERR_NOSERVICE || |
| rc == SM_ERR_NOPEER || |
| rc == SM_ERR_NODE_DOWN) |
| { |
| glob->handleReaderThreadError(rc, "ExSM_Post", dataHandle, smTaskList); |
| return 0; |
| } |
| |
| EXSM_TRACE(EXSM_TRACE_BUFFER|EXSM_TRACE_SM_CALLS, |
| "PREPOST %d:%d:%" PRId64 ":%d:0x%c %p sz %d", |
| (int) target.node, (int) target.pid, target.id, |
| (int) ExSMTag_GetTagWithoutQualifier(target.tag), |
| (char) ExSMTag_GetQualifierDisplay(target.tag), |
| chunkAddr, (int) chunkSize); |
| |
| exsm_assert_rc(rc, "ExSM_Post"); |
| |
| // Send an ACK. The target is the sender's target with the |
| // reply bit flipped. |
| target = chunk.tgt; |
| target.tag = ExSMTag_ToggleReplyFlag(chunk.tgt.tag); |
| |
| ExSMShortMessage m; |
| m.setTarget(target); |
| m.setNumValues(1); |
| m.setValue(0, (int32_t) ExSMShortMessage::ACK); |
| int32_t rc2 = m.send(); |
| if (rc2 == SM_ERR_NOSERVICE || |
| rc2 == SM_ERR_NOPEER || |
| rc2 == SM_ERR_NODE_DOWN) |
| { |
| glob->handleReaderThreadError(rc2, "ExSM_SendShortMessage", dataHandle, smTaskList); |
| return 0; |
| } |
| |
| } |
| |
| // ACK |
| else if (shortMsgType == ExSMShortMessage::ACK) |
| { |
| task->sendChunk_SetAckArrived(true); |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT, |
| "%s ACK arrived, task %p", readerPrefix, task); |
| } |
| |
| // UNKNOWN SHORT MESSAGE TYPE |
| else |
| { |
| char buf[64]; |
| snprintf(buf, sizeof(buf), "Invalid short message type %d", |
| (int) shortMsgType); |
| exsm_assert(FALSE, buf); |
| } |
| |
| // We are done processing short messages. Return to the top of |
| // the loop that processes each chunk. |
| continue; |
| |
| } // if (short message) |
| |
| // Now we process prepost messages. Cases to consider: |
| // (a) This is a chunk, but not the final chunk |
| // (b) This is the final chunk |
| // (c) This message is not chunked, the complete message arrived |
| |
| // In cases (b) and (c) we need to insert the receive buffer |
| // into the task output queue. The local variable |
| // bufferForOutputQueue will point to the buffer that gets |
| // inserted. |
| void *bufferForOutputQueue = NULL; |
| |
| // Before processing the data buffer or modifying any variables |
| // in the task or its queues, first make sure the address |
| // returned from SM matches one of the following: |
| // |
| // * For a SQL chunk: the task's most recent prepost address |
| // * For a complete message: first element of the task input |
| // queue |
| // |
| // An assertion will fail if there is not a match. |
| |
| ExSMQueue *outQ = task->getOutQueue(); |
| ExSMQueue *inQ = task->getInQueue(); |
| void *expectedAddr = NULL; |
| void *actualAddr = chunk.buff; |
| |
| // This block determines the correct value of expectedAddr. The |
| // comparison of expectedAddr versus actualAddr happens right |
| // after this block. |
| if (receivedSQLBufferFragment) |
| { |
| expectedAddr = task->recvChunk_GetPrepostAddr(); |
| } |
| else |
| { |
| exsm_assert(!inQ->isEmpty(), "SM task input queue is empty"); |
| ExSMQueue::Entry &entry = inQ->getHeadEntry(); |
| void *dataPointer = entry.getData(); |
| exsm_assert(dataPointer, |
| "Task queue entry contains a NULL buffer pointer"); |
| expectedAddr = ExSM_GetMessageBufferPrepostAddr(dataPointer); |
| |
| // We can also set bufferForOutputQueue when the arrival is a |
| // complete message |
| bufferForOutputQueue = dataPointer; |
| } |
| |
| // Make sure expectedAddr matches actualAddr |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT, |
| "%s prepost %p, arrival %p", readerPrefix, |
| expectedAddr, actualAddr); |
| exsm_assert(expectedAddr == actualAddr, |
| "Expected address does not match actual"); |
| |
| // Now we trust the chunk.buff address and can process data |
| |
| if (receivedSQLBufferFragment) |
| { |
| // Cases (a) and (b) |
| |
| // Tell the task that new bytes have arrived |
| task->recvChunk_Receive(chunk.size); |
| |
| // See if this is the final chunk or not and process |
| // accordingly |
| if (task->recvChunk_MoreExpected()) |
| { |
| // Case (a): Not the final chunk |
| |
| // Post the next chunk |
| sm_target_t target = chunk.tgt; |
| ExSMTag_SetSQLChunkFlag(&target.tag); |
| |
| uint32_t chunkSize = (uint32_t) task->recvChunk_GetChunkSize(); |
| char *chunkAddr = (char *) task->recvChunk_GetPrepostAddr(); |
| |
| int32_t rc = ExSM_Post(target, chunkSize, chunkAddr, task, 0); |
| |
| if (rc == SM_ERR_NOSERVICE || |
| rc == SM_ERR_NOPEER || |
| rc == SM_ERR_NODE_DOWN) |
| { |
| glob->handleReaderThreadError(rc, "ExSM_Post", dataHandle, smTaskList); |
| return 0; |
| } |
| |
| exsm_assert_rc(rc, "ExSM_Post"); |
| |
| EXSM_TRACE(EXSM_TRACE_BUFFER|EXSM_TRACE_SM_CALLS, |
| "PREPOST %d:%d:%" PRId64 ":%d:0x%c %p", |
| (int) target.node, (int) target.pid, target.id, |
| (int) ExSMTag_GetTagWithoutQualifier(target.tag), |
| (char) ExSMTag_GetQualifierDisplay(target.tag), |
| chunkAddr, (int) chunkSize); |
| |
| // Send an ACK. The target is the sender's target with the |
| // reply bit flipped and the chunk flag cleared. |
| target = chunk.tgt; |
| target.tag = ExSMTag_ToggleReplyFlag(chunk.tgt.tag); |
| ExSMTag_ClearSQLChunkFlag(&target.tag); |
| |
| ExSMShortMessage m; |
| m.setTarget(target); |
| m.setNumValues(1); |
| m.setValue(0, (int32_t) ExSMShortMessage::ACK); |
| int32_t rc2 = m.send(); |
| if (rc2 == SM_ERR_NOSERVICE || |
| rc2 == SM_ERR_NOPEER || |
| rc2 == SM_ERR_NODE_DOWN) |
| { |
| glob->handleReaderThreadError(rc2, "ExSM_SendShortMessage", dataHandle, smTaskList); |
| return 0; |
| } |
| } |
| else |
| { |
| // Case (b): Final chunk |
| |
| // Store a pointer to the receive buffer. It will be used |
| // later for an insert into the task output queue. |
| bufferForOutputQueue = task->recvChunk_GetBuffer(); |
| exsm_assert(bufferForOutputQueue, |
| "Task returned a NULL buffer pointer"); |
| |
| // Adjust the length of the receive buffer |
| ExSM_SetMessageLength(bufferForOutputQueue, chunk.size); |
| |
| // The task can now release its pointer to the receive |
| // buffer. The buffer will not be forgotten because it gets |
| // added to the task output queue a few lines below. |
| task->recvChunk_Exit(); |
| } |
| |
| } // received a chunk |
| |
| else |
| { |
| // Case (c): A complete message |
| |
| // Make sure bufferForOutputQueue was initialized |
| exsm_assert(bufferForOutputQueue, |
| "bufferForOutputQueue was not initialized"); |
| |
| // Set the message length in the preposted receive buffer |
| ExSM_SetMessageLength(bufferForOutputQueue, chunk.size); |
| |
| // Remove the input queue element |
| inQ->removeHead(); |
| |
| } // received a complete message (not a chunk) |
| |
| // If bufferForOutputQueue is not NULL, this is either a |
| // complete message or the final chunk. We need to place the |
| // receive buffer on the task output queue. We also add the task |
| // to the SM ready list if the task output queue was initially |
| // empty. |
| |
| if (bufferForOutputQueue) |
| { |
| exsm_assert(!outQ->isFull(), "SM task output queue is full"); |
| bool outQueueWasEmpty = outQ->isEmpty(); |
| |
| ExSMQueue::Entry &outEntry = outQ->getTailEntry(); |
| outEntry.setData(bufferForOutputQueue); |
| outQ->insert(); |
| |
| if (outQueueWasEmpty) |
| { |
| // 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). |
| readyList->add(task); |
| } |
| |
| // Schedule the recipient of this message (the recipient is |
| // currently a connection object whose wait() method will |
| // execute once the main thread receives the LRABBIT event) |
| int32_t *scheduledAddr = task->getScheduledAddr(); |
| exsm_assert(scheduledAddr, |
| "SM task scheduledAddr pointer is NULL"); |
| *scheduledAddr = 1; |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT, |
| "%s scheduled task %p", readerPrefix, task); |
| |
| } // if (bufferForOutputQueue) |
| } // for each chunk |
| |
| // Unlock the task list |
| smTaskList->unlock(); |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT, |
| "%s unlocked task list", readerPrefix); |
| |
| // Release SeaMonster buffer space |
| rc = SM_get_done(0, dataHandle); |
| |
| // Add a WAITDONE event to the in-memory trace |
| ExSMEvent::add(ExSMEvent::ReceiveDone); |
| |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT, |
| "%s SM_get_done rc %d", readerPrefix, (int) rc); |
| |
| // Generate a seabed event to wake the main thread. There is no |
| // need to wake the main thread if we already received the |
| // SHUTDOWN message. |
| if (glob->getReaderThreadState() != ExSMGlobals::PROCESSING_SHUTDOWN) |
| { |
| XAWAKE(EXSM_MAIN_THREAD_PIN, LRABBIT); |
| EXSM_TRACE(EXSM_TRACE_RDR_THR|EXSM_TRACE_WAIT, |
| "%s generated LRABBIT", readerPrefix); |
| } |
| |
| } // while (glob->getReaderThreadState() == ExSMGlobals::STARTED) |
| |
| // At this point the reader thread loop has exited and the thread is |
| // going to exit. |
| // |
| // Reader thread state is set to DONE so the main thread knows the |
| // thread is no longer active. |
| // |
| // We also signal the main thread via a condition variable so the |
| // main thread can wake up in case it was waiting for the reader |
| // thread to complete (this is the case where the main thread sends |
| // the reader thread a SHUTDOWN message). |
| |
| EXSM_TRACE(EXSM_TRACE_RDR_THR, "%s Reader thread shutting down", |
| readerPrefix); |
| EXSM_TRACE(EXSM_TRACE_RDR_THR, "%s Reader thread state %s", readerPrefix, |
| glob->getThreadStateString(glob->getReaderThreadState())); |
| |
| // Change the reader thread state to DONE. This will signal the main |
| // thread which could be waiting for the reader thread to finish its |
| // shutdown processing. |
| glob->setReaderThreadState(ExSMGlobals::DONE, |
| true, // do locking |
| true); // signal |
| |
| EXSM_TRACE(EXSM_TRACE_RDR_THR,"%s END READER", readerPrefix); |
| return 0; |
| } |
