Google Git
Sign in
apache / trafodion / refs/tags/2.3.0rc1 / . / core / sql / common / ipcmsg.cpp
blob: 1ee07201093a670bb9db447f803acc0364cbf92b [file] [log] [blame]
/**********************************************************************
// @@@ 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: IpcMsg.cpp
* Description: Implementation for the IPC classes using the NSK messaging
* API.
*
* Created: 6/25/99
* Language: C++
*
*
*
*
*****************************************************************************
*/
#define AEVENT 1
#define set_extern_data
#include "Platform.h"
#include "rosgen.h"
#include "fs_rosetta_dml.h"
#include "dpxnsdp2"
#include "yfsiopen"
//#include "dfsiopn.h"
enum {FS_SMS_VERSION_MAY94 = 1};
#include "wdialect"
#include "ppctlc(WAIT, SETSTOP)"
#include "dmsghi.h"
#include "psignalc.h(PK_SIG_SYSTEMCALL_ABORTINQUIRE_, PK_SUSPEND_DISALLOW_SET_)"
#include "pmallocc(ADDRESS_WIRE_, ADDRESS_UNWIRE_)"
#include "hpfs2f(fs2_transid_to_buffer)"
#include "ffilcpp(FS_SQL_SETUPREQUESTINFO, FS_SQL_PUTMSGIDINACB, \
FS_SQL_RESETAFTERREPLY)"
#include "Int64.h"
#define _resident
#define _priv
#include "ExCollections.h"
#include "Ipc.h"
#include "str.h"
#include "ComDiags.h"
#include "NAExit.h"
#include "ipcmsg.h"
#include <fcntl.h>
#include "logmxevent.h"
extern "C" {
//#include <cextdecs.h>
#include "cextdecs.h(PROCESSHANDLE_TO_FILENAME_,PROCESSHANDLE_DECOMPOSE_,FILE_OPEN_,SETMODE,FILE_GETINFO_,FILE_CLOSE_, AWAITIOX,PROCESS_DELAY_)"
#include <tal.h>
// should be #include <zsysc.h>
#include "zsysc.h"
}
//Function used to get a pointer to the pfs
_callable void fs2_get_pfsaddr(Long *);
//Function used to get transid from the filesystem
extern "C" _priv _resident int_16 FS_GETTRANSID_
(
extaddr tubaddr, // input
// the address of the Trans Usage Block
// (having the tcbref as the 1st
// field), or 0D or ptmfnocurtransid
phandle_template *destination, // input
// where it will be sent
int_16 *buffer); // output
// the tcbref is placed here
extern "C" _priv int_16 TMFLIBFS_ABORTTRANS_
(
int_16 *tcbref_ptr,
short disposition
);
// -----------------------------------------------------------------------
// Methods for class GuaMsgConnectionToServer
// -----------------------------------------------------------------------
GuaMsgConnectionToServer::GuaMsgConnectionToServer(
IpcEnvironment *env,
const IpcProcessId &procId,
NABoolean usesTransactions,
unsigned short nowaitDepth,
const char *eye) : IpcConnection(env,procId,eye)
{
openFile_ = InvalidGuaFileNumber;
nowaitDepth_ = nowaitDepth;
maxIOSize_ = env->getGuaMaxMsgIOSize();
activeIOs_ = new(env) ActiveIOQueueEntry[nowaitDepth_];
//get the length of the request/reply control structure
// int controllen = sizeof(fs_fs_template) + sizeof(fs_fs_template::__writeread);
Int32 controllen = sizeof(fs_fs_writeread);
//initialize each entry in the Active IOs queue
for (unsigned short i = 0; i < nowaitDepth_; i++)
{
activeIOs_[i].writeDataCBAPtr_ = (void*) new(env) char[sizeof(NSK_CBA)];
activeIOs_[i].readDataCBAPtr_ = (void*) new(env) char[sizeof(NSK_CBA)];
activeIOs_[i].controlCBAPtr_ = (void*) new(env) char[sizeof(NSK_CBA)];
activeIOs_[i].controlBuf_ = (void*) new(env) char[controllen];
activeIOs_[i].inUse_ = FALSE;
activeIOs_[i].expectReply_ = FALSE;
activeIOs_[i].msgid_ = 0;
activeIOs_[i].transid_ = -1;
activeIOs_[i].buffer_ = activeIOs_[i].readBuffer_ = NULL;
}
lastAllocatedEntry_ = nowaitDepth_-1;
numOutstandingIOs_ = 0;
partiallySentBuffer_ = NULL;
chunkBytesSent_ = 0;
partiallyReceivedBuffer_ = NULL;
chunkBytesRequested_ = 0;
chunkBytesReceived_ = 0;
usesTransactions_ = usesTransactions;
abortXnOnPathErrors_ = FALSE;
guaErrorInfo_ = GuaOK;
currentEntry_ = 0;
// We need a nowait dept of at least 2, one for a message and another
// one for out-of-band messages (not really implemented yet).
// assert(nowaitDepth_ >= 2);
sendCallbackBufferList_ = new(env) IpcMessageBuffer*[nowaitDepth_];
for (unsigned short i = 0; i < nowaitDepth_; i++)
sendCallbackBufferList_[i] = NULL;
// now open the server process
openPhandle(NULL);
}
GuaMsgConnectionToServer::~GuaMsgConnectionToServer()
{
closePhandle();
CollHeap *heap = getEnvironment()->getHeap();
for (Int32 i = 0; i < nowaitDepth_; i++)
{
ActiveIOQueueEntry &entry = activeIOs_[i];
heap->deallocateMemory(entry.writeDataCBAPtr_);
heap->deallocateMemory(entry.readDataCBAPtr_);
heap->deallocateMemory(entry.controlCBAPtr_);
heap->deallocateMemory(entry.controlBuf_);
}
heap->deallocateMemory(activeIOs_);
heap->deallocateMemory(sendCallbackBufferList_);
}
void GuaMsgConnectionToServer::send(IpcMessageBuffer *buffer)
{
// simply add the new buffer to the send queue and try to start
// as many new I/O operations as possible
queueSendMessage(buffer);
while (tryToStartNewIO())
;
}
void GuaMsgConnectionToServer::receive(IpcMessageStreamBase *msg)
{
// Receiving from a Guardian server is implicit, since the WRITEREADX
// call performs both a send and a receive operation together. However,
// we still need to add the callback and, if the I/O has already
// completed, call the callback.
addReceiveCallback(msg);
// maybe the Guardian I/O has already completed and the buffer is
// waiting in the base class' receive queue
IpcMessageBuffer *receiveBuf;
while (receiveBuf = getNextReceiveQueueEntry())
{
// yes, so just call its callback
receiveBuf->callReceiveCallback(this);
}
}
NABoolean GuaMsgConnectionToServer::moreWaitsAllowed()
{
return !stopWait_;
}
//
// Wait for an I/O reply. After receives a reply, the I/O entry looks like:
//
// - entry.buffer_=entry.readBuffer_=reply buffer
//
WaitReturnStatus GuaMsgConnectionToServer::wait(IpcTimeout timeout, UInt32 *eventConsumed = NULL, IpcAwaitiox *ipcAwaitiox)
{
NABoolean ipcAwaitioxCompleted = ipcAwaitiox != NULL;
if (ipcAwaitioxCompleted)
ipcAwaitioxCompleted = ipcAwaitiox->getCompleted();
assert(ipcAwaitioxCompleted == FALSE);
//Internal error: AWAITIOX should not have completed a message system based operation
short error = 0;
short waitField = LDONE | LSIG;
short status = 0;
direct_globals_template * pfsptr;
fs2_get_pfsaddr((Long*)&pfsptr);
if ((ULng32)(openFile_) >= (ULng32)(pfsptr->numftentries))
{
// If this connection is not open, there can be nothing to wait on.
// Return an error in this case to indicate that this is an inappropriate
// call. Do no wait on this connection again.
guaErrorInfo_ = FENOTOPEN;
setErrorInfo(-1);
handleIOError();
stopWait(TRUE);
return WAIT_OK;
}
// don't do anything if the connection is in an error state and
// there are no more pending requests to work on.
if (getState() == ERROR_STATE AND numOutstandingIOs_ <= 0
AND numQueuedSendMessages() <= 0)
{ // no more waits on this connection
stopWait(TRUE);
return WAIT_OK;
}
stopWait(FALSE);
setBreakReceived(FALSE);
// try to send or receive first if there is a timeout specified,
// or if we don't have any IOs outstanding. Also, try more
// I/O to cover the special case that we've posted only the
// first chunk of an multi-chunk msg.
// This latter scenario is possible because of the logic on
// tryToStartNewIO that gives up if the per-process limit on
// MQCs would be exceeded. It required this special handling
// for multichunk messages, as explained in the next paragraph.
// The connection may have to give up either 1) before the
// first chunk is posted, or 2) before the second chunk is
// posted or 3) after the second chunk is posted.
// The tryToStartNewIO method is called from this class's send
// method and then also from two places in this wait method.
// In case 1), the test of numOutStandingIOs_ will be
// sufficient to cause tryToStartNewIO to be called directly
// below this comment. For case 3), the other call to
// tryToStartNewIO will be taken, because the a_message_is_done
// will be set to true after the server replies to the chunk(s)
// already posted. But for case 2), we need special handling,
// because the server does not reply to the first chunk until
// all chunks are sent. And this class's send method will not
// be called again for the multichunk message. So the additional
// test below to detect that the second chunk has not yet been
// sent, is needed to force the call to tryToStartNewIO.
if (timeout != IpcImmediately ||
numOutstandingIOs_ == 0 ||
chunkBytesSent_ == maxIOSize_)
while (tryToStartNewIO())
;
// try to complete I/Os within the given time limit,
// if there are outstanding I/Os
if (numOutstandingIOs_ == 0)
return WAIT_OK;
//used to mark the point in the activeIOs queue from
//where we started checking for IO completion
unsigned short start = currentEntry_;
//indicates we found a completed message
Int32 a_message_is_done = 0;
//check if anyone of our messages has completed
do {
if ((activeIOs_[currentEntry_].inUse_) &&
(activeIOs_[currentEntry_].expectReply_))
{
//check if the message for this entry is done
if (MSG_ISDONE2_((NSK_msId2)activeIOs_[currentEntry_].msgid_))
{
getEnvironment()->setEvent(TRUE, AEVENT);
a_message_is_done = 1;
break;// found a complete message break
}
}
currentEntry_++;
/*
** Not that expensive since if-conversion reduces the control flow into
** predicated instructions
*/
if (currentEntry_ == nowaitDepth_)
currentEntry_ = 0;
} while(currentEntry_ != start );
//
// IMPORTANT: MUST NOT have any early return statement from here until
// the end of this function. So that the loop at the end of the function
// that issues receive callbacks will always be executed if any receive
// entries were queued by this function.
//
NABoolean interrupt = FALSE;
//if no message is done wait, then check again for message completion
//for nowaited mode (timeout == 0), do not call WAIT.
if ((!a_message_is_done && timeout != 0) || getEnvironment()->lsigConsumed())
{
// Only wait for LDONE if breakEnabled is not set.
IpcTimeout waitTimeout = timeout == 0 ? -2 : timeout;
if (getEnvironment()->lsigConsumed())
status = LSIG;
else
{
if (getEnvironment()->breakEnabled())
status = WAIT(waitField, waitTimeout);//wait for timeout
else
status = WAIT(LDONE, waitTimeout);//wait for timeout
}
if (!status)
{
a_message_is_done = FALSE; // timed out
}
else
{
if (status & LSIG)
{
getEnvironment()->setLsigConsumed(FALSE);
short oldsigmod = PK_SUSPEND_DISALLOW_SET_(1);
error = PK_SIG_SYSTEMCALL_ABORTINQUIRE_();
PK_SUSPEND_DISALLOW_SET_(oldsigmod);
if (error > 0)
{
// received a signal.
guaErrorInfo_ = FE_EINTR;
setBreakReceived(TRUE);
setErrorInfo(-1);
handleIOError();
interrupt = TRUE; // FE_EINTR
}
} // if (status & LSIG)
else
{
//woken up because some message completed on the LDONE queue
//somewhere. Check if the message completed was ours.
getEnvironment()->setLdoneConsumed(TRUE);
do {
//check if this activeIOs_ entry is in Use
if (activeIOs_[currentEntry_].inUse_ &&
activeIOs_[currentEntry_].expectReply_)
{
//check if the message for this entry is done
if (MSG_ISDONE2_((NSK_msId2)activeIOs_[currentEntry_].msgid_))
{
a_message_is_done = 1;
// The following line of code was added but is being
// removed because:
// a) LDONE consumed which servers the same purpose
// has already been set, and
// b) a compiler performance regression occurred and
// it's an unlikely but possible cause
//getEnvironment()->setEvent(TRUE, AEVENT);
break;
}
}
currentEntry_++;
if (currentEntry_ == nowaitDepth_) {
currentEntry_ = 0;
}
} while(currentEntry_ != start );
}
}
}
if (a_message_is_done)
{
ActiveIOQueueEntry &entry = activeIOs_[currentEntry_];
NSK_msResult2 results;
short oldstop = SETSTOP(2);//become unstoppable
//Pickup reply and terminate message
MSG_BREAK2_((NSK_msId2)entry.msgid_, &results,
(NSK_PHandle _ptr64 *)getOtherEnd().getPhandle().phandle_);
//Get the error from the reply control buffer
message_header_template * ReplyControlBuf = (message_header_template *)entry.controlBuf_;
error = ReplyControlBuf->error();
if (error == GuaTimeoutErr)
{
// timeout does not set the connection into an error state but it
// causes a return. later we shall wait on this connection again.
SETSTOP(oldstop);//become stoppable
guaErrorInfo_ = error;
return WAIT_OK;
}
//reset the filesystem data structures
Int64 localTransid = entry.transid_;
resetAfterReply(entry.msgid_, error, &localTransid);
SETSTOP(oldstop);//become stoppable
// we have got the reply for this I/O entry
entry.expectReply_ = FALSE;
if (error)
{
// Remember the Guardian error code
guaErrorInfo_ = error;
setErrorInfo(-1);
handleIOErrorForEntry(entry);
}
else
{
cleanUpActiveIOEntry(entry);
//get # of bytes written in reply
ULng32 countRead = (ULng32)results.rr_dataSize;
// Now try to figure out what the original operation was, so
// we know what to do with the IpcMessageBuffer:
//
// a) If this I/O was a write operation for part of a buffer,
// then another operation for the same buffer is following, so
// just remove the outstanding I/O entry.
// b) If this I/O returned part of a buffer, then we have to issue
// another I/O operation for the rest of the buffer.
// c) If we have received all of the data, the buffer is ready to
// be delivered to its destination. If the buffer has its callback
// assigned, then call the callback, otherwise add the buffer to
// the receive queue that is managed by the base class,
// IpcConnection. "completelyReadBuffer" is set for case c)
//
if (entry.receiveBufferSizeLeft_ == 0)
{
// case a)
assert(countRead == 0);
// Note that this does NOT count as a completion, we don't
// let the upper layers know that we are using multiple
// Guardian I/Os for this.
}
else
{
// we did expect data back, case b) or c)
if (entry.offset_ == 0)
{
if (numOutstandingIOs_ > 0)
{
// more pending I/Os on this connection. check if there
// are other chunks of entry.buffer_.
//
// - in multi-chunk mode, when server receives any
// of the subsequent chunk, it sends an empty reply
// right away. however, when server receives the first
// chunk, it does not reply right away. instead, server
// holds the first chunk until it has received all
// subsequent chunks and then server replies to the
// first chunk. on the client side, even though the
// first chunk receives its reply the last, but since
// we are looking at a randomly selected I/O entry,
// "entry" could be any one of the multi-chunk I/Os.
//
for (unsigned short i = 0; i < nowaitDepth_; i++)
{
ActiveIOQueueEntry &nextEntry = activeIOs_[i];
if (nextEntry.inUse_ &&
nextEntry.buffer_ == entry.buffer_)
{
// entry is the first chunk and nextEntry
// is a subsequent chunk of a multi-chunk
// send buffer.
// only the first chunk receives reply.
// there is no reply for subsequent chunks.
// we must clean up I/O on nextEntry now
// or otherwise the shared send buffer
// entry.buffer_ can get deallocated
// (see a few lines below). after that
// we cannot clean up I/O on nextEntry as
// entry.buffer_->chunkLockCount_ is not
// longer accessible.
cleanUpActiveIOEntry(nextEntry);
if (numOutstandingIOs_ == 0)
// no more pending I/Os on this connection
break;
}
}
}
if (entry.buffer_->isShared())
{
// no longer need the shared send buffer. release it.
entry.buffer_->decrRefCount(getEnvironment());
// now use only the reply buffer
entry.buffer_ = entry.readBuffer_;
}
// since this is the first (maybe only) chunk of the message
// buffer, we can get the length of the total message by
// looking into the message header.
// Get the size of the message sent (or the reply buffer if shared)
IpcMessageObjSize bytesSent = entry.buffer_->getMessageLength();
// unpack message header which contains reply message length
InternalMsgHdrInfoStruct *msgHdr =
new( (IpcMessageObj*)(entry.buffer_->data(0)) )
InternalMsgHdrInfoStruct(NULL);
IpcMessageObjSize msgLen = msgHdr->getMsgLengthFromData();
// remember the real length of the message coming back
entry.buffer_->setMessageLength(msgLen);
// check whether this is case b) or c)
if (msgLen == (IpcMessageObjSize) countRead)
{
// The "normal" case c). This is a single-chunk reply.
// If we were sending a large buffer(more than one chunk)
// and just received a small buffer (one chunk) then
// release the large buffer to conserve space on the IPC
// heap.
if (bytesSent > maxIOSize_)
{
entry.buffer_ = entry.buffer_->resize(getEnvironment(), msgLen);
}
queueReceiveMessage(entry.buffer_);
}
else
{
// Case b). This is a multi-chunk reply. Switch to the
// chunk protocol, countRead bytes are already received
// back from the server.
if (msgLen > entry.buffer_->getBufferLength() ||
bytesSent > maxIOSize_)
{
// We want to resize the reply buffer if either:
// - The server has a reply message that is larger
// than what our buffer can hold
// - The request buffer was large (more than one
// chunk) and may now be consuming space
// unnecessarily on the IPC heap
entry.buffer_->setMessageLength(countRead);
entry.buffer_ = entry.buffer_->resize(getEnvironment(), msgLen);
entry.buffer_->setMessageLength(msgLen);
}
// do some sanity checks, make sure we don't have two
// partial buffers at a time
assert(partiallyReceivedBuffer_ == NULL);
assert(msgLen > (IpcMessageObjSize) countRead);
// move some information from the entry to data members
// in the connection while the chunk protocol is going on
partiallyReceivedBuffer_ = entry.buffer_;
chunkBytesRequested_ = countRead;
chunkBytesReceived_ = countRead;
getEnvironment()->getAllConnections()->
setReceivedPartialMessage(TRUE);
}
} // first (maybe only) chunk
else
{
// case b), this is not the first chunk
assert (partiallyReceivedBuffer_ == entry.buffer_);
chunkBytesReceived_ += countRead;
if (chunkBytesReceived_ == entry.buffer_->getMessageLength())
{
// this was the last chunk
queueReceiveMessage(partiallyReceivedBuffer_);
partiallyReceivedBuffer_ = NULL;
chunkBytesRequested_ = 0;
chunkBytesReceived_ = 0;
}
else
{
getEnvironment()->getAllConnections()->
setReceivedPartialMessage(TRUE);
}
} // not the first chunk
} // case b) or c)
// this I/O completed
if (getState() != ERROR_STATE && numOutstandingIOs_ == 0)
setState(ESTABLISHED);
// after waiting, try (again) to start as many new I/O operations as
// possible
while (tryToStartNewIO())
;
} // if (error) else
} // if (a_message_is_done)
IpcMessageBuffer *receiveBuf;
NABoolean aCallbackIsCalled = FALSE;
while (receiveBuf = getNextReceiveQueueEntry())
{
// When the user of this connection sets trustIncomingBuffers_ to
// FALSE then we perform an integrity check on all incoming
// message buffers. A failure causes the connection to transition
// to the ERROR_STATE state.
if (!getTrustIncomingBuffers() && getState() != ERROR_STATE)
{
if (!receiveBuf->verifyBackbone())
{
setIpcMsgBufCheckFailed(TRUE);
guaErrorInfo_ = 0;
setErrorInfo(-1);
setState(ERROR_STATE);
}
}
receiveBuf->callReceiveCallback(this);
aCallbackIsCalled = TRUE;
}
// In the ERROR_STATE state we may have to announce I/O completion after
// callbacks are issued. The setState() method has the job of
// detecting when I/O is complete and informing the IpcEnviroment at
// the appropriate time. Even though it's not intuitive to call
// setState(ERROR_STATE) here (because we are already in the ERROR_STATE state),
// we make the call anyway to trigger any necessary bookkeeping.
if (aCallbackIsCalled && getState() == ERROR_STATE)
setState(ERROR_STATE);
return interrupt ? WAIT_INTERRUPT : WAIT_OK ;
}
GuaMsgConnectionToServer * GuaMsgConnectionToServer::castToGuaMsgConnectionToServer()
{
return this;
}
Int32 GuaMsgConnectionToServer::numQueuedSendMessages()
{
return sendQueueEntries();
}
Int32 GuaMsgConnectionToServer::numQueuedReceiveMessages()
{
return receiveQueueEntries();
}
void GuaMsgConnectionToServer::populateDiagsArea(ComDiagsArea *&diags,
CollHeap *diagsHeap)
{
if (guaErrorInfo_ != GuaOK)
{
IpcAllocateDiagsArea(diags,diagsHeap);
*diags << DgSqlCode(-2034) << DgInt0(guaErrorInfo_);
getEnvironment()->getMyOwnProcessId(IPC_DOM_GUA_PHANDLE).
addProcIdToDiagsArea(*diags,0);
getOtherEnd().addProcIdToDiagsArea(*diags,1);
}
if (getIpcMsgBufCheckFailed())
{
IpcAllocateDiagsArea(diags, diagsHeap);
*diags << DgSqlCode(-2037);
getEnvironment()->getMyOwnProcessId(IPC_DOM_GUA_PHANDLE).
addProcIdToDiagsArea(*diags, 0);
getOtherEnd().addProcIdToDiagsArea(*diags, 1);
}
}
//
// Send a new I/O request.
//
// In Guardian each msg I/O requires a write buffer to send the request, and
// a read buffer used to receive the reply. A same buffer can be used as both
// the write buffer and the read buffer. here we will refer the write buffer
// as the "send buffer", and the read buffer as the "reply buffer".
//
// In our implementation we use ActiveIOQueueEntry to represent a msg I/O.
// entry.buffer_ and entry.readBuffer_ can be used as the send buffer and the
// reply buffer. however, the values of entry.buffer_ and entry.readBuffer_
// depend on the type of the message buffer to be sent. There are four
// possible scenarios:
//
// 1. multi-chunk send buffer, shared by multiple connections:
//
// - first chunk: entry.buffer_=send buffer
// entry.readBuffer_=reply buffer
// - subsequent chunks: entry.buffer_=send buffer, entry.readBuffer_=NULL
//
// 2. multi-chunk send buffer, single connection (not shared):
//
// - first chunk: entry.buffer_=entry.readBuffer_=send buffer
// - subsequent chunks: entry.buffer_=send buffer, entry.readBuffer_=NULL
//
// 3. single-chunk send buffer, shared by multiple connections
//
// - entry.buffer_=send buffer
// - entry.readBuffer_=reply buffer
//
// 4. single-chunk send buffer, single connection (not shared):
//
// - entry.buffer_=entry.readBuffer_=send buffer
//
NABoolean GuaMsgConnectionToServer::tryToStartNewIO()
{
// Any more messages or parts of messages to send?
// There is nothing to do if there are neither new messages nor
// incompleted partial messages.
if (sendQueueEntries() == 0 && !partiallySentBuffer_ &&
!partiallyReceivedBuffer_)
return FALSE;
// do not allow new send if a partial message is being received and we
// already have requested all the reply data for it.
if (partiallyReceivedBuffer_ &&
partiallyReceivedBuffer_->getMessageLength() == chunkBytesRequested_)
return FALSE;
// Can't have more than nowaitDepth_ - 1 I/Os outstanding, except
// when there is an out-of-band message for which we make an exception.
// If there is an out-of-Band message then assume that it was placed
// in front of the send queue.
if (numOutstandingIOs_ >=
(IFX (partiallySentBuffer_ OR partiallyReceivedBuffer_)
THENX nowaitDepth_
ELSEX (nowaitDepth_-1)))
return FALSE;
// Check if the per-process limit on MQCs is exceeded.
// Note that this should be reconsidered when this code is
// multithreaded.
short numMsgsActual;
if (MESSAGESYSTEMINFO(5, &numMsgsActual))
assert(0);
if (numMsgsActual+1 >= getEnvironment()->getMaxPerProcessMQCs())
return FALSE;
// If we reach here this means we can start another nowait I/O;
// get to the outstanding I/O entry that is to be filled next.
#ifndef NDEBUG
NABoolean wrapAroundCheck = FALSE;
#endif
// We may have to return early from this method if we cannot find
// space on the IPC heap for an outgoing buffer. If that happens
// we'll want to restore the original value of lastAllocatedEntry_.
unsigned short originalLastAllocated = lastAllocatedEntry_;
// find an entry that is not in use
while (activeIOs_[lastAllocatedEntry_].inUse_)
{
// increment lastAllocatedEntry_ modulo nowaitDepth_
lastAllocatedEntry_++;
if (lastAllocatedEntry_ == nowaitDepth_)
{
lastAllocatedEntry_ = 0;
#ifndef NDEBUG
assert(!wrapAroundCheck); // to detect infinite loop (shouldn't happen)
wrapAroundCheck = TRUE;
#endif
}
}
// we have found an entry that is not in use
ActiveIOQueueEntry &entry = activeIOs_[lastAllocatedEntry_];
//assert(!entry.inUse_ && !entry.expectReply_);
// ---------------------------------------------------------------------
// set up a new outstanding IO entry, depending on what to do next
// but don't start the corresponding I/O quite yet
// ---------------------------------------------------------------------
// initialize all fields of the entry (there is no constructor)
entry.bytesSent_ = 0;
entry.receiveBufferSizeLeft_ = 0;
entry.offset_ = 0;
entry.msgid_ = 0;
// These help keep track of the need to callSendCallback.
NABoolean isFirstChunk = FALSE;
NABoolean isLastChunk = FALSE;
// ---------------------------------------------------------------------
// Decide what to do, depending on the currently pending buffers and
// IOs:
//
// a) send another chunk of a large message down to the server without
// asking for data back
// b) request some more data from the server, if the server replied
// with a partial message and we haven't asked for all of the
// rest of the data yet (never interleave this with I/Os of
// steps a) and b), so the server won't get confused
// c) get another message from the send queue and find out that it
// is too long for a single chunk, so send the first piece
// d) should be the normal case, get the next message from the send
// queue and send it in a single chunk
// ---------------------------------------------------------------------
if (partiallySentBuffer_)
{
// case a) continue sending more chunks for this buffer
// but don't ask for reply data, since we may want the reply to come
// back at entry.buffer_->data(0)
entry.buffer_ = partiallySentBuffer_;
// for multi-chunk buffer, whether shared or not, the read buffer for
// any chunk after first chunk is NULL.
entry.readBuffer_ = NULL;
assert(chunkBytesSent_ < entry.buffer_->getMessageLength());
entry.bytesSent_ = MINOF(maxIOSize_,
entry.buffer_->getMessageLength() -
chunkBytesSent_);
entry.offset_ = chunkBytesSent_;
chunkBytesSent_ += entry.bytesSent_;
// if this is the last chunk ...
if (chunkBytesSent_ >= entry.buffer_->getMessageLength())
{
// we're done sending chunks
partiallySentBuffer_ = NULL;
chunkBytesSent_ = 0;
// can call the send callback now
isLastChunk = TRUE;
}
lastSentBuffer_ = entry.buffer_;
}
else if (partiallyReceivedBuffer_)
{
// b) next thing to do is to receive another chunk from the server
entry.buffer_ = entry.readBuffer_ = partiallyReceivedBuffer_;
entry.offset_ = chunkBytesRequested_;
entry.receiveBufferSizeLeft_ =
MINOF(maxIOSize_,
entry.buffer_->getMessageLength() - chunkBytesRequested_);
chunkBytesRequested_ += entry.receiveBufferSizeLeft_;
lastSentBuffer_ = entry.buffer_;
}
else
{
// get the next buffer to send from the send queue and check
// whether it can be sent in one piece
assert(sendQueueEntries() > 0);
IpcMessageBuffer *nextToSend = sendQueue()[0];
assert(nextToSend);
// assume request and reply use same buffer
entry.buffer_ = entry.readBuffer_ = nextToSend;
isFirstChunk = TRUE;
if (entry.buffer_->getRefCount() > 1 || entry.buffer_->isShared())
{
if (!entry.buffer_->initLockCount(getEnvironment()->getHeap(),
maxIOSize_))
{
// We ran out of space on the IPC heap. This is OK and we
// can return early. Higher layers will redrive the I/O for
// this connection.
getEnvironment()->setHeapFullFlag(TRUE);
lastAllocatedEntry_ = originalLastAllocated;
return FALSE;
}
// The send buffer is shared by multiple connections. Therefore,
// allocate a different buffer for the reply. the reply buffer size
// is set as:
//
// - if user has explicitly specified the reply buffer length
// and it is less than maxIOSize_, then use it.
// - otherwise, use maxIOSize_.
IpcMessageStream *msgStream = entry.buffer_->getMessageStream()->castToIpcMessageStream();
//assert(msgStream);
IpcMessageObjSize maxReplyLen = msgStream->getMaxReplyLength();
if (maxReplyLen > 0 && maxReplyLen < maxIOSize_)
entry.readBuffer_ = entry.buffer_->createBuffer(getEnvironment(), maxReplyLen, FALSE);
else
entry.readBuffer_ = entry.buffer_->createBuffer(getEnvironment(), maxIOSize_, FALSE);
if (entry.readBuffer_ == NULL)
{
// We ran out of space on the IPC heap ...
getEnvironment()->setHeapFullFlag(TRUE);
lastAllocatedEntry_ = originalLastAllocated;
return FALSE;
}
}
if (entry.buffer_->getMessageLength() > maxIOSize_)
{
// case c), the message we just got from the send queue is too large
// to be sent in a single chunk :-(
assert(partiallySentBuffer_ == NULL);
// indicate multi-chunk protocol
partiallySentBuffer_ = entry.buffer_;
entry.bytesSent_ = maxIOSize_;
chunkBytesSent_ = entry.bytesSent_;
}
else
{
// case d) can send in single chunk
entry.bytesSent_ = entry.buffer_->getMessageLength();
// can call the send callback now
isLastChunk = TRUE;
}
prepareSendBuffer(entry.buffer_);
// got this far so de-queue buffer from this connection's queue
removeNextSendBuffer();
entry.receiveBufferSizeLeft_ =
MINOF(maxIOSize_, entry.readBuffer_->getBufferLength());
lastSentBuffer_ = entry.buffer_;
}
// ---------------------------------------------------------------------
// Next, start the I/O operation
// ---------------------------------------------------------------------
short wireOptions = 8;
short GuardianError = 0;
direct_globals_template * pfsptr;
acb_standard_template * acb;
// Get the pointer to the acb - needed in order to decrement the count
// of outstanding requests in the ACB on errors.
fs2_get_pfsaddr((Long*)&pfsptr);
// check that the file is actually open, i.e., that this is a valid
// filenum.
if ((ULng32)(openFile_) >= (ULng32)(pfsptr->numftentries))
{
GuardianError = FENOTOPEN;
}
else
acb = (acb_standard_template *) pfsptr->file_table[openFile_];
//become unstoppable
short oldstop = SETSTOP(2);
if (NOT GuardianError)
{
// lock memory used for the control info buffer
GuardianError = ADDRESS_WIRE_((unsigned char *)entry.controlBuf_,
sizeof(fs_fs_writeread), wireOptions);
}
if (NOT GuardianError)
{
entry.transid_ = (Int64)entry.buffer_->getTransid();
// setup request control buffer
GuardianError = setupRequestInfo((void*)entry.controlBuf_,
(Int64)entry.buffer_->getTransid());
if (GuardianError)
ADDRESS_UNWIRE_((unsigned char *)entry.controlBuf_,
(sizeof(fs_fs_writeread)), wireOptions);
}
if (NOT GuardianError)
{
GuardianError = addressWire(entry, wireOptions);
if (GuardianError)
{
// decrement the number of outstanding requests -
// this was incremented by the call to setupRequestInfo() above.
acb->acb_numreqs = acb->acb_numreqs - 1;
ADDRESS_UNWIRE_((unsigned char *)entry.controlBuf_,
(sizeof(fs_fs_writeread)), wireOptions);
}
}
if (NOT GuardianError)
{
// generate Context-Based Addresses (CBAs) for the data buffer
// and for the control buffer
NSK_CBA &writeDataCBA = *(NSK_CBA*) entry.writeDataCBAPtr_;
NSK_CBA &readDataCBA = *(NSK_CBA*) entry.readDataCBAPtr_;
NSK_CBA &controlCBA = *(NSK_CBA*) entry.controlCBAPtr_;
// for all other types of send use the send buffer
CBA_CREATE_((NSK_CBA *)entry.writeDataCBAPtr_, entry.buffer_->data(entry.offset_));
if (!entry.readBuffer_)
// this is a subsequent chunk of a multi-chunk send. no reply is
// needed.
CBA_CREATE_((NSK_CBA *)entry.readDataCBAPtr_, 0);
else
// all other types of reply
CBA_CREATE_((NSK_CBA *)entry.readDataCBAPtr_, entry.readBuffer_->data(entry.offset_));
CBA_CREATE_((NSK_CBA *)entry.controlCBAPtr_,entry.controlBuf_);
NSK_msLinkOpts2 options = MSG_LINK_CBA;
short retryCount = 0;
NABoolean needToRetry;
do {
// send the message to the server, using CBAs, so that hide/reveal
// operations later can't invalidate the addresses
NSK_msId2 localMsgId = 0;
GuardianError = MSG_LINK2_(
(NSK_PHandle _ptr64 *)getOtherEnd().getPhandle().phandle_,
(NSK_msId2 _ptr64 *)&localMsgId,
(int16 _ptr64 *)&controlCBA,
(NSK_msSize2)sizeof(fs_fs_writeread),
(int16 _ptr64 *)&controlCBA,
(NSK_msSize2)sizeof(fs_fs_writeread),
(char _ptr64 *)&writeDataCBA, //message to send to server
(NSK_msSize2)entry.bytesSent_, //# of bytes to send
(char _ptr64 *)&readDataCBA, //buffer to receive reply from server
(NSK_msSize2)entry.receiveBufferSizeLeft_, //reply bytes expected
0, 0, 0, options);
entry.msgid_ = (UInt32)localMsgId;
if (GuardianError == FENOLCB) // too many MQCs.
{
// Since the per-process limit was checked above,
// assume that it is the per-cpu limit, so let us
// retry.
retryCount++;
needToRetry = TRUE;
getEnvironment()->incrRetriedMessages();
PROCESS_DELAY_(10*1000); // 10,000 microseconds
}
else
needToRetry = FALSE;
} while (needToRetry &&
retryCount < 100*60 ); // after 60 seconds (and 6000 retries),
// just give up.
getEnvironment()->setEvent(TRUE, AEVENT);
if (GuardianError)
{
// failed message link
// decrement the number of outstanding requests -
// this was incremented by the call to setupRequestInfo() above.
acb->acb_numreqs = acb->acb_numreqs - 1;
addressUnwire(entry);
} // if (GuardianError)
else
{
// Put the message id into the acb.
putMsgIdinACB(entry.msgid_);
}
}
//go back to old stop mode
SETSTOP(oldstop);
if (isFirstChunk)
addSendCallbackBuffer(entry.buffer_);
if (GuardianError)
{
// an error happened somewhere along the way and we must
// a) record the Guardian error number,
guaErrorInfo_ = GuardianError;
// b) set the connection to the error state. If we have not invoked
// the send callback, then handleIOErrorForEntry() will invoke the
// send callback.
setErrorInfo(-1);
handleIOErrorForEntry(entry);
// if the design is to disallow any future i/o after connection become
// error state, should we return false to prevent from calling
// tryToStartNewIO() again?
//return FALSE;
}
else
{
// buffer has been sent successfully
if (numOutstandingIOs_ == 0)
setState(SENDING);
numOutstandingIOs_++;
entry.inUse_ = TRUE; // this entry now has an I/O in progress
entry.expectReply_ = TRUE;
// --------------------------------------------------------------
// If we started the I/O for all chunks of an IpcMessageBuffer
// (or if the IpcMessageBuffer was sent in a single message), then
// call its send callback.
// --------------------------------------------------------------
if (isLastChunk)
{
// removeSendCallbackBuffer() should always return TRUE here because
// send callback has not been invoked yet. If error occured prior to
// this method call then handleIOErrorForEntry() should have cleared
// all I/Os on the same message stream and we would not have come
// here.
NABoolean sendCallbackFlag = removeSendCallbackBuffer(entry.buffer_);
if (sendCallbackFlag)
// the send callback doesn't give away entry.buffer_, and this is
// good since the same buffer may be still used for the receive
// operation.
entry.buffer_ ->callSendCallback(this);
else // - for debugging only
assert(sendCallbackFlag);
}
}
return TRUE;
}
void GuaMsgConnectionToServer::openPhandle(char * processName)
{
char procFileName[IpcMaxGuardianPathNameLength];
short procFileNameLen;
short openFlags = nowaitDepth_ == 0 ? 0x0 : 0x4000;
if (! processName)
{
// convert the phandle to a string that can be passed to FILE_OPEN_
guaErrorInfo_ = PROCESSHANDLE_TO_FILENAME_(
(short *) getOtherEnd().getPhandle().phandle_,
procFileName,
IpcMaxGuardianPathNameLength,
&procFileNameLen,
0);
if (guaErrorInfo_ != GuaOK)
{
setErrorInfo(-1);
setState(ERROR_STATE);
return;
}
}
else
{
strcpy(procFileName, processName);
procFileNameLen = (short)strlen(processName);
}
getEnvironment()->setLdoneConsumed(TRUE);
// solution 10-081025-6810:
// wait for esp open reply indefinitely. time out every 10 minutes, write
// a warning msg to ems log including error code and esp pin, and then go
// back to waiting. we allow up to 10 timeouts, or 100 minutes.
//
// - when AWAITIOX times out, it completes/cancels the nowait i/o
// initiated by FILE_OPEN_. thus we cannot simply call AWAITIOX again since
// there is no more outstanding i/o. we must try FILE_OPEN_ again and then
// call AWAITIOX to wait another 10 minutes. also, we must save the file
// numbers returns from each timed out FILE_OPEN_ so we can close them later.
// and we cannot call FILE_CLOSE_ on any of the returned file numbers until
// AWAITIOX returns successfully with esp's open reply. otherwise the close
// msg from master will cause esp to exit.
//
GuaFileNumber oldOpens[10]; // allow up to 10 timeouts
short numOldOpens = 0;
_cc_status stat;
while (1)
{
guaErrorInfo_ = FILE_OPEN_(procFileName,
procFileNameLen,
&openFile_,
0, // open for read/write access
0, // shared access
nowaitDepth_,
0, // sync depth 0 (target proc is not NonStop)
openFlags); // options
if (guaErrorInfo_ != GuaOK)
break;
if (!(openFlags &= 0x4000))
// return if FILE_OPEN_ was waited
break;
// FILE_OPEN_ was no-waited
// wait indefinitely
stat = AWAITIOX(&openFile_);
if (_status_eq(stat))
// reply received with no error
break;
// we got error. get Guardian error code.
GuaErrorNumber getinfoError = FILE_GETINFO_(openFile_,&guaErrorInfo_);
if (getinfoError != 0)
guaErrorInfo_ = getinfoError; // not even FILE_GETINFO_ worked
// set guaErrorInfo_ to -1 if we did not get a valid error?
//if (guaErrorInfo_ == GuaOK)
// guaErrorInfo_ = -1;
// AWAITIOX returned error, or ran out allowed timeouts.
break;
} // while (1)
for (short i = 0; i < numOldOpens; i++)
FILE_CLOSE_(oldOpens[i]);
if (guaErrorInfo_ != GuaOK)
{
if (openFile_ != -1)
{
FILE_CLOSE_(openFile_); // Don't retain unopened ACB
openFile_ = -1; // Don't leave valid file number in object!
}
setErrorInfo(-1);
setState(ERROR_STATE);
return;
}
// some day we may want to perform nowaited FILE_OPEN_ calls and add
// a method to "work" on the open.
// use setmode 74 to turn off the automatic CANCEL upon AWAITIOX timeout
stat = SETMODE(openFile_,74,-1);
if (_status_ne(stat))
{
// get a Guardian error code
Int32 errcode2 = FILE_GETINFO_(openFile_,&guaErrorInfo_);
if (errcode2 != 0)
guaErrorInfo_ = errcode2; // not even FILE_GETINFO_ worked
setErrorInfo(-1);
setState(ERROR_STATE);
return;
}
// use setmode 30 to allow I/O operations to finish in any order
stat = SETMODE(openFile_,30,3);
if (_status_ne(stat))
{
// get a Guardian error code
Int32 errcode2 = FILE_GETINFO_(openFile_,&guaErrorInfo_);
if (errcode2 != 0)
guaErrorInfo_ = errcode2; // not even FILE_GETINFO_ worked
setErrorInfo(-1);
setState(ERROR_STATE);
return;
}
// use setmode 117 if no transactions should be propagated to the server
if (NOT usesTransactions_)
{
_cc_status stat = SETMODE(openFile_,117,1);
if (_status_ne(stat))
{
// get a Guardian error code
Int32 errcode2 = FILE_GETINFO_(openFile_,&guaErrorInfo_);
if (errcode2 != 0)
guaErrorInfo_ = errcode2; // not even FILE_GETINFO_ worked
setErrorInfo(-1);
setState(ERROR_STATE);
return;
}
}
// the connection is established now
setState(ESTABLISHED);
}
void GuaMsgConnectionToServer::closePhandle()
{
//
// it's possible that some pending I/Os still remain on this connection.
// if IPC error occurs during query execution, there are three possible
// scenarios for a connection when it comes here:
//
// 1. this connection received ipc error and handled the error on the spot
// by calling one of the handleIOError() methods.
// 2. this connection did not receive error, but some other connection(s)
// on the same message stream received error. in this case we should
// invoke IpcMessageStream::abandonPendingIOs() to abort pending I/Os
// on all of stream's connections and invoke all necessary callbacks.
// 3. this connection did not receive error, neither did any other
// connections on the same message stream. in this case it's possible
// that nothing happens on this connection until the destructor is
// invoked. thus we need to clean up any pending I/Os on the connection.
// we should invoke delinkConnection() on the stream to trigger any
// book keepings needed, as delinkConnection() and receive callback
// should have the same book keeping logic.
//
// example for #3: we have a test case that kills esps while executing a
// long running query. the master has multiple send tops, with each send top
// having a message stream that includes only one data connection to a top
// level esp. if IPC error occurs on other connections (but not on this
// connection) as the result of dead esps, all sql operations are aborted
// from higher level and the master may never get a chance to call
// handleIOError() to clean up any pending I/Os on this connection. thus we
// need to release any msg buffers from the pending I/Os. but note that
// in case of multi-chunk message buffer, there may be multiple I/O entries
// pointing to the same message buffer. in that case each buffer should be
// released only once.
//
handleIOError();
// receive queue may not be empty. if so invoke receive callback.
// example:
// GuaMsgConnectionToServer::setFatalError() invokes handleIOErrorForStream()
// that may have put message buffer on receive queue.
IpcMessageBuffer *receiveBuf;
while (receiveBuf = getNextReceiveQueueEntry())
receiveBuf->callReceiveCallback(this);
// Note that after closing, the connection is always in the initial
// state. This is the way to fix a connection in the error state.
// The close is always considered successful.
guaErrorInfo_ = GuaOK;
clearErrorInfo();
setState(INITIAL);
if (openFile_ != InvalidGuaFileNumber)
{
FILE_CLOSE_(openFile_);
openFile_ = InvalidGuaFileNumber;
}
}
//This function sets up the control information required by
//the file system on the server side
//control is a pointer to the buffer that is supposed to
//contain the control information
short GuaMsgConnectionToServer::setupRequestInfo(void * control, Int64 transid){
//Redirected to T9055 to insulate SQL/MX from changes in
//ACB_REQUEST_TEMPLATE
Int32 retcode = FS_SQL_SETUPREQUESTINFO(openFile_,
(fs_fs_template *) control,
transid);
return retcode;
#if 0
fs_fs_writeread* controlInfo =(fs_fs_writeread*) control;
direct_globals_template * pfsptr;//pointer to the PFS
acb_standard_template * acb;//pointer to the acb for this file
short error = FEOK; //variable to catch errors
//first get pointer to the PFS
fs2_get_pfsaddr((Long*)&pfsptr);
//then get pointer to the acb in the
//PFS Filetable at the index openFile_
acb = (acb_standard_template *) pfsptr->file_table[openFile_];
//set the dialect type to indicate this is a call from the file system
//since we are trying to emulate the WRITEREADX call
controlInfo->dialect_type = DIALECT_FS_FS;
controlInfo->request_type = FS_FS_WRITEREAD;
controlInfo->request_version = CURRENT_VERSION_FS_FS;
controlInfo->minimum_interpretation_version = MINIMUM_VERSION_FS_FS;
controlInfo->tcbref_valid = 0;
controlInfo->lid_valid = 0;
controlInfo->filler = 0;
memset((void *)&(controlInfo->sendflags), 0,
sizeof(linkmon_sendflags_template));
//get the number of outstanding requests
int_16 next = acb->acb_numreqs;
void ** acb_reqaddrs = (void **)((char *)&(acb->req.acb_requestbase_addr) +
sizeof(acb->req.acb_requestbase_addr) );
//get a pointer to the data for this request
acb_request_template * acb_reqptr =
(acb_request_template *)acb_reqaddrs[next];
//check if numreqs is less than the maximum number of reqs allowed
if (( acb->acb_numreqs) < (int_16) acb->acb_maxreqs)
{
//update the numreqs
acb->acb_numreqs = next + 1;
int_16 savenum = acb_reqptr->acb_reqnum;
//initialize the acb req to all zeros
_fill_32(acb_reqptr, (_len(acb_request_template)/ 4), 0 );
acb_reqptr->acb_reqnum = savenum;
} // if req avail
else
{
// we used to return FETOOMANY here, but the check in
// tryToStartNewIO of numOutStandingIOs and nowaitDepth_
// should have prevented the FETOOMANY.
assert(0);
}
// if transid argument is invalid set flag in controlInfo.
if (transid == (Int64)-1)
{
controlInfo->tcbref_valid = 0;
}
else
{
// Here, acb_reqptr->acb_tubaddr is not initialized with the
// context transid (as done in FS), to prevent ENDTRANSACTION
// from returning error 81 if there are outstanding ipc msgs
// (for read only query).
// For nowaited queries involves insert, update, delete,
// or nowaited prepares, CliGlobals::checkOperationsPending
// detects outstanding operations and returns error to
// FS2^FLUSH^ALL^VSBB.
// transid passed in is valid. Call FS2 to move transid into
// control buffer (may need to "massage" transid if the destination
// is a remote node.
error = Fs2_transid_to_buffer
(acb->posix.acb_procsection->tmfvirtualnode,
(unsigned char *)&transid,
(unsigned char *)&(controlInfo->tcbref),
0
);
if (error)
{
acb->acb_numreqs = acb->acb_numreqs - 1;
return error;
}
controlInfo->tcbref_valid = 1;
}
//hard coded since file open passes a zero
controlInfo->sender.syncid = 1;
//get the filenum for this connection
controlInfo->sender.first_word.filenum = openFile_;
//copy the phandle
memcpy(&controlInfo->sender.phandle,&pfsptr->my_phandle,(sizeof(short)*10));
controlInfo->sender.user_openid = acb->posix.acb_procsection->acb_useropenid;
controlInfo->sender.id.openid = acb->posix.acb_procsection->sender.acb_procopenid;
return FEOK;
#endif
}
//This function is used to put the msgid into the acb after a MSG_LINK_.
//This let's the filesystem cleanup outstanding IOs on PROCESS_STOP_
void GuaMsgConnectionToServer::putMsgIdinACB(UInt32 msgid){
//Redirected to T9055 to insulate SQL/MX from changes in
//ACB_REQUEST_TEMPLATE
Int32 retcode = FS_SQL_PUTMSGIDINACB(openFile_, msgid);
#if 0
direct_globals_template * pfsptr;//pointer to the PFS
acb_standard_template * acb;//pointer to the acb for openFile_
//first get pointer to the PFS
fs2_get_pfsaddr((Long*)&pfsptr);
//then get pointer to the acb in the PFS
//Filetable at the index openFile_
acb = (acb_standard_template *) pfsptr->file_table[openFile_];
//get the index into the array of acb requests
int_16 next = acb->acb_numreqs - 1;
void ** acb_reqaddrs = (void **)((char *)&(acb->req.acb_requestbase_addr) +
sizeof(acb->req.acb_requestbase_addr) );
//get a pointer to the acb data for this request
acb_request_template * acb_reqptr =
(acb_request_template *)acb_reqaddrs[next];
//put the msgid into the acb
acb_reqptr->mid.acb_mid = msgid;
if (acb_reqptr->tub.acb_tubaddr != 0)
acb_reqptr->tub.acb_tub->pending_count =
(ULng32)(acb_reqptr->tub.acb_tub->pending_count) + 1u;
#endif
}
void GuaMsgConnectionToServer::resetAfterReply(UInt32 msgid, short error,
Int64 *transid){
//Redirected to T9055 to insulate SQL/MX from changes in
//ACB_REQUEST_TEMPLATE
Int32 retcode = FS_SQL_RESETAFTERREPLY(openFile_, msgid, error, transid,
abortXnOnPathErrors_);
#if 0
direct_globals_template * pfsptr;//pointer to the PFS
acb_standard_template * acb;//pointer to the acb for openFile_
acb_request_template * acb_reqptr;
short reqIndex = -1;
short i;
//first get pointer to the PFS
fs2_get_pfsaddr((Long*)&pfsptr);
//then get pointer to the acb in the
//PFS Filetable at the index openFile_
acb = (acb_standard_template *) pfsptr->file_table[openFile_];
void ** acb_reqaddrs = (void **)((char *)&(acb->req.acb_requestbase_addr) +
sizeof(acb->req.acb_requestbase_addr) );
// get a pointer to the acb data for this request (search through array
// of outstanding requests,, to get the request id that matches this one,
// that was just completed). This is a fix for a bug - earlier, it was
// just assumed that the request completed was the last request in the
// array - this may not be the case, as requests may be completed out
// of order. This bug lead to %4100 halts (BR #85). The fix to search
// the array for the completed request, along with the fix to "shrink" the
// array (see code below) solves this bug by doing what the Enscribe
// Filesystem does in AWAITIOX.
for (i=0; i <= acb->acb_numreqs; i++)
{
acb_reqptr = (acb_request_template *)acb_reqaddrs[i];
if (acb_reqptr->mid.acb_mid == msgid)
{
reqIndex = i;
break;
}
}
//set msgid to 0
acb_reqptr->mid.acb_mid = 0;
acb_reqptr->acb_reqrdy = 1;
if (acb_reqptr->tub.acb_tubaddr != 0)
acb_reqptr->tub.acb_tub->pending_count =
(ULng32)(acb_reqptr->tub.acb_tub->pending_count) - 1u;
//decrement the numreqs
acb->acb_numreqs = acb->acb_numreqs - 1;
// Move all requests after this one to be before this one. This "shrinking"
// of the request array is done to make it efficient to find a slot while
// sending a new request (free slots are always at the end of the array)
if ((acb->acb_numreqs > 0) && (reqIndex != acb->acb_numreqs))
{
i = reqIndex;
while (i != acb->acb_numreqs)
{
acb_reqaddrs[i] = acb_reqaddrs[i + 1];
i = i + 1;
}
acb_reqaddrs[acb->acb_numreqs] = (void *)acb_reqptr;
}
if (error == FECPUFAIL ||
error == FEPATHDOWN ||
(error >= FENETERR && error <= 255)
)
{
// abort the transaction if a path error is received.
if ((transid != NULL) && (*transid != -1))
{
// Want to stop the transaction on path errors. This is
// to fix the bug reported in solution 10-030508-6267.
if (abortXnOnPathErrors_)
{
TMFLIBFS_ABORTTRANS_( (int_16 *)transid,
FETRANSABRTOWNERDIED );
}
}
}
#endif
}
void GuaMsgConnectionToServer::setFatalError(IpcMessageStreamBase *msgStream)
{
if (guaErrorInfo_ == GuaOK) // if error hasn't been set yet
guaErrorInfo_ = GuaIpcApplicationErr;
setState(ERROR_STATE);
// we must set error info to -1 so receive callback knows not to
// parse the potentially corrupted receive queue buffer.
setErrorInfo(-1);
// handleIOErrorForStream() may put the message buffer on receive queue.
// and that buffer may have the following content:
//
// - the actual full reply from server, or
// - the send buffer if send failed, or
// - partial send or reply if send/reply was multi-chunk and did not
// complete
//
handleIOErrorForStream(msgStream);
// This next call only works if GuaMsgConnectionToServer inherits
// directly from IpcConnection.
IpcConnection::setFatalError(msgStream);
}
void GuaMsgConnectionToServer::addSendCallbackBuffer(IpcMessageBuffer *buffer)
{
for (unsigned short i = 0; i < nowaitDepth_; i++)
{
if (!sendCallbackBufferList_[i])
{
sendCallbackBufferList_[i] = buffer;
return;
}
}
}
NABoolean GuaMsgConnectionToServer::removeSendCallbackBuffer(IpcMessageBuffer *buffer)
{
for (unsigned short i = 0; i < nowaitDepth_; i++)
{
if (sendCallbackBufferList_[i] == buffer)
{
sendCallbackBufferList_[i] = NULL;
return TRUE;
}
}
return FALSE;
}
void GuaMsgConnectionToServer::handleIOError()
{
// connection no longer usable due to I/O error. abort all existing I/Os
// on this connection.
for (unsigned short i = 0; i < nowaitDepth_; i++)
{
ActiveIOQueueEntry &entry = activeIOs_[i];
if (entry.inUse_)
handleIOErrorForEntry(entry);
}
assert(partiallySentBuffer_ == NULL);
assert(partiallyReceivedBuffer_ == NULL);
currentEntry_ = 0;
numOutstandingIOs_ = 0;
}
void GuaMsgConnectionToServer::handleIOErrorForStream(IpcMessageStreamBase *msgStream)
{
// abort all existing I/Os on this connection that are from the given stream
for (unsigned short i = 0; i < nowaitDepth_; i++)
{
ActiveIOQueueEntry &entry = activeIOs_[i];
if (entry.inUse_ && entry.buffer_->getMessageStream() == msgStream)
handleIOErrorForEntry(entry);
}
}
//
// The I/O entry has a write buffer (entry.buffer_) and a read buffer
// (entry.readBuffer_). For entry description after message send, see comments
// at the top of tryToStartNewIO(). For entry description after receive reply,
// see comments at the top of wait().
//
void GuaMsgConnectionToServer::handleIOErrorForEntry(ActiveIOQueueEntry &entry)
{
// I/O error occurred on given entry during send or receive
if (getState() != ERROR_STATE)
setState(ERROR_STATE);
if (getErrorInfo() == 0)
setErrorInfo(-1);
// clean up I/O entry
if (entry.inUse_)
cleanUpActiveIOEntry(entry);
// abort all existing I/Os on the same message stream
// - what about I/Os on other streams?
for (unsigned short i = 0; i < nowaitDepth_; i++)
{
ActiveIOQueueEntry &nextEntry = activeIOs_[i];
if (nextEntry.inUse_ &&
nextEntry.buffer_->getMessageStream() ==
entry.buffer_->getMessageStream())
{
cleanUpActiveIOEntry(nextEntry);
// special case: for multi-chunk shared buffer, if nextEntry is
// first chunk and entry is any chunk after first chunk, we need
// to free the read buffer of the first chunk since all other chunks
// have their read buffers set to NULL.
if (nextEntry.readBuffer_ &&
nextEntry.readBuffer_ != nextEntry.buffer_)
// nextEntry is the first chunk of a multi-chunk shared buffer
nextEntry.readBuffer_->decrRefCount(getEnvironment());
}
} // for i
// clear partial send/receive buffer on the same message stream so to
// prevent further I/Os.
if (partiallySentBuffer_ &&
partiallySentBuffer_->getMessageStream() ==
entry.buffer_->getMessageStream())
partiallySentBuffer_ = NULL;
else if (partiallyReceivedBuffer_ &&
partiallyReceivedBuffer_->getMessageStream() ==
entry.buffer_->getMessageStream())
partiallyReceivedBuffer_ = NULL;
// if entry is the first chunk of a multi-chunk shared buffer, we need to
// free its read buffer.
if (entry.readBuffer_ && entry.readBuffer_ != entry.buffer_)
entry.readBuffer_->decrRefCount(getEnvironment());
// put the buffer on receive queue, regardless of send succeeded or not.
queueReceiveMessage(entry.buffer_);
// if entry.buffer_ is still on the sendCallbackBufferList_, then send
// callback has not been invoked for this connection and thus send failed.
NABoolean sendSuccess = !removeSendCallbackBuffer(entry.buffer_);
// in case of send failure invoke send callback
if (!sendSuccess)
entry.buffer_->callSendCallback(this);
// the design is to disallow any future i/o after connection become error
// state. so we should cleanup send queue by invoking send callback for
// all send buffers.
IpcMessageBuffer *sendBuffer;
while (sendBuffer = removeNextSendBuffer())
{
queueReceiveMessage(sendBuffer);
sendBuffer->callSendCallback(this);
}
}
void GuaMsgConnectionToServer::cleanUpActiveIOEntry(ActiveIOQueueEntry &entry)
{
short oldstop = SETSTOP(2);//become unstoppable
// abort if we are still waiting for message reply
if (entry.expectReply_)
{
MSG_ABANDON2_((NSK_msId2)entry.msgid_);
resetAfterReply(entry.msgid_, 0, NULL);
}
addressUnwire(entry);
SETSTOP(oldstop);//become stoppable
entry.expectReply_ = FALSE;
entry.inUse_ = FALSE;
entry.msgid_ = 0;
entry.transid_ = -1;
numOutstandingIOs_--;
}
// lock memory used for the actual message send/reply
short GuaMsgConnectionToServer::addressWire(ActiveIOQueueEntry &entry,
short wireOptions)
{
//
// no need to lock control buffer in this method. it should have been locked
// already by the caller.
//
//
// lock memory for write/send buffer. write buffer may be shared.
//
short GuardianError = 0;
if (entry.buffer_->isShared())
{
// write buffer is shared by multiple connections
if (entry.buffer_->getLockCount(entry.offset_) == 0)
{
// if the write buffer chunk is not locked, lock it
GuardianError = ADDRESS_WIRE_
((unsigned char *)entry.buffer_->data(entry.offset_),
entry.bytesSent_, wireOptions);
if (GuardianError)
return GuardianError;
}
// increment lock count for shared write bufer
entry.buffer_->incrLockCount(entry.offset_);
}
else
{
// write buffer is only used by a single connection
// entry.buffer_ = entry.readBuffer_ = send buffer
ULng32 maxDataBufferLength =
MAXOF(entry.receiveBufferSizeLeft_, entry.bytesSent_);
GuardianError= ADDRESS_WIRE_
((unsigned char *)entry.buffer_->data(entry.offset_),
maxDataBufferLength, wireOptions);
if (GuardianError)
return GuardianError;
}
//
// lock memory for read/reply buffer. reply buffer is never shared.
//
if (entry.readBuffer_ && entry.readBuffer_ != entry.buffer_)
{
// entry.buffer_ = send buffer (shared)
// entry.readBuffer_ = reply buffer
assert(entry.buffer_->isShared());
GuardianError = ADDRESS_WIRE_
((unsigned char *)entry.readBuffer_->data(0),
entry.receiveBufferSizeLeft_, wireOptions);
if (GuardianError)
{
// if error, unlock write buffer that must be shared
if (entry.buffer_->decrLockCount(entry.offset_) == 0)
// no more I/O on the write buffer. unlock its memory.
ADDRESS_UNWIRE_((unsigned char *)entry.buffer_->data(entry.offset_),
entry.bytesSent_, wireOptions);
return GuardianError;
}
}
return GuardianError;
}
// unlock memory used for the actual message send/reply
void GuaMsgConnectionToServer::addressUnwire(ActiveIOQueueEntry &entry)
{
//
// unlock memory for control buffer
//
short wireOptions = 8;
ADDRESS_UNWIRE_((unsigned char *)entry.controlBuf_, sizeof(fs_fs_writeread), wireOptions);
//
// unlock memory for write/send buffer. write buffer may be shared.
//
if (entry.buffer_->isShared())
{
// write buffer is shared by multiple connections
if (entry.buffer_->decrLockCount(entry.offset_) == 0)
// no more I/O on the write buffer. unlock its memory.
ADDRESS_UNWIRE_((unsigned char *)entry.buffer_->data(entry.offset_),
entry.bytesSent_, wireOptions);
}
else
{
// write buffer is only used by a single connection
// entry.buffer_ = entry.readBuffer_ = send buffer
ULng32 maxDataBufferLength =
MAXOF(entry.receiveBufferSizeLeft_, entry.bytesSent_);
ADDRESS_UNWIRE_((unsigned char *)entry.buffer_->data(entry.offset_),
maxDataBufferLength, wireOptions);
}
//
// unlock memory for read/reply buffer. reply buffer is never shared.
//
if (entry.readBuffer_ && entry.readBuffer_ != entry.buffer_)
ADDRESS_UNWIRE_((unsigned char *)entry.readBuffer_->data(0),
entry.receiveBufferSizeLeft_, wireOptions);
}
#undef _resident
#undef _priv