From 6eb25148c580f2402c1c28262f6a5663ae2c4656 Mon Sep 17 00:00:00 2001 From: Jeff Hill Date: Sat, 25 Jul 2009 00:43:28 +0000 Subject: [PATCH] simplified tcp send interactions --- src/cas/generic/casDGClient.cc | 13 +++---- src/cas/generic/casDGClient.h | 4 +-- src/cas/generic/casStrmClient.cc | 60 ++++++-------------------------- src/cas/generic/casStrmClient.h | 6 ++-- src/cas/generic/outBuf.cc | 54 +++++++++++++--------------- src/cas/generic/outBuf.h | 22 ++++++------ 6 files changed, 55 insertions(+), 104 deletions(-) diff --git a/src/cas/generic/casDGClient.cc b/src/cas/generic/casDGClient.cc index 2375a8e4a..c2c993e6d 100644 --- a/src/cas/generic/casDGClient.cc +++ b/src/cas/generic/casDGClient.cc @@ -408,18 +408,15 @@ void casDGClient::sendBeacon ( ca_uint32_t beaconNumber ) // // casDGClient::xSend() // -outBufClient::flushCondition casDGClient::xSend ( char *pBufIn, // X aCC 361 - bufSizeT nBytesAvailableToSend, bufSizeT nBytesNeedToBeSent, - bufSizeT &nBytesSent ) +outBufClient::flushCondition casDGClient::xSend ( char *pBufIn, + bufSizeT nBytesToSend, bufSizeT & nBytesSent ) { - assert ( nBytesAvailableToSend >= nBytesNeedToBeSent ); - bufSizeT totalBytes = 0; - while ( totalBytes < nBytesNeedToBeSent ) { + while ( totalBytes < nBytesToSend ) { cadg *pHdr = reinterpret_cast < cadg * > ( & pBufIn[totalBytes] ); - assert ( totalBytes <= bufSizeT_MAX-pHdr->cadg_nBytes ); - assert ( totalBytes + pHdr->cadg_nBytes <= nBytesAvailableToSend ); + assert ( totalBytes <= bufSizeT_MAX - pHdr->cadg_nBytes ); + assert ( totalBytes + pHdr->cadg_nBytes <= nBytesToSend ); char * pDG = reinterpret_cast < char * > ( pHdr + 1 ); unsigned sizeDG = pHdr->cadg_nBytes - sizeof ( *pHdr ); diff --git a/src/cas/generic/casDGClient.h b/src/cas/generic/casDGClient.h index 952ac8c60..1f9087877 100644 --- a/src/cas/generic/casDGClient.h +++ b/src/cas/generic/casDGClient.h @@ -80,8 +80,8 @@ private: const caHdrLargeArray &, const pvExistReturn &, ca_uint16_t protocolRevision, ca_uint32_t sequenceNumber ); void sendVersion (); - outBufClient::flushCondition xSend ( char *pBufIn, bufSizeT nBytesAvailableToSend, - bufSizeT nBytesNeedToBeSent, bufSizeT &nBytesSent ); + outBufClient::flushCondition xSend ( char *pBufIn, bufSizeT nBytesToSend, + bufSizeT &nBytesSent ); inBufClient::fillCondition xRecv ( char * pBufIn, bufSizeT nBytesToRecv, fillParameter parm, bufSizeT & nByesRecv ); virtual outBufClient::flushCondition osdSend ( diff --git a/src/cas/generic/casStrmClient.cc b/src/cas/generic/casStrmClient.cc index 9100392ad..cde5a71d1 100644 --- a/src/cas/generic/casStrmClient.cc +++ b/src/cas/generic/casStrmClient.cc @@ -112,9 +112,9 @@ casStrmClient::~casStrmClient () } // -// casStrmClient::processMsg () +// casStrmClient :: processMsg () // -caStatus casStrmClient::processMsg () +caStatus casStrmClient :: processMsg () { epicsGuard < casClientMutex > guard ( this->mutex ); int status = S_cas_success; @@ -141,7 +141,7 @@ caStatus casStrmClient::processMsg () this->incommingBytesToDrain = 0u; } } - + // // process any messages in the in buffer // @@ -2330,60 +2330,20 @@ inline bool caServerI::roomForNewChannel() const // // casStrmClient::xSend() // -outBufClient::flushCondition casStrmClient::xSend ( char * pBufIn, - bufSizeT nBytesAvailableToSend, - bufSizeT nBytesNeedToBeSent, - bufSizeT & nActualBytes ) +outBufClient::flushCondition casStrmClient :: + xSend ( char * pBufIn, bufSizeT nBytesToSend, bufSizeT & nBytesSent ) { - outBufClient::flushCondition stat = outBufClient::flushDisconnect; - bufSizeT nActualBytesDelta; - bufSizeT totalBytes; - - assert ( nBytesAvailableToSend >= nBytesNeedToBeSent ); - - totalBytes = 0u; - while ( true ) { - stat = this->osdSend ( &pBufIn[totalBytes], - nBytesAvailableToSend-totalBytes, nActualBytesDelta ); - if ( stat != outBufClient::flushProgress ) { - if ( totalBytes > 0 ) { - nActualBytes = totalBytes; - // - // !! this time fetch may be slowing things down !! - // - //this->lastSendTS = epicsTime::getCurrent(); - stat = outBufClient::flushProgress; - break; - } - else { - break; - } - } - - totalBytes += nActualBytesDelta; - - if ( totalBytes >= nBytesNeedToBeSent ) { - // - // !! this time fetch may be slowing things down !! - // - //this->lastSendTS = epicsTime::getCurrent(); - nActualBytes = totalBytes; - stat = outBufClient::flushProgress; - break; - } - } - return stat; + return this->osdSend ( pBufIn, nBytesToSend, nBytesSent ); } // // casStrmClient::xRecv() // -inBufClient::fillCondition casStrmClient::xRecv ( char * pBufIn, bufSizeT nBytes, +inBufClient::fillCondition casStrmClient::xRecv ( char * pBufIn, bufSizeT nBytesToRecv, inBufClient::fillParameter, bufSizeT & nActualBytes ) { - inBufClient::fillCondition stat; - - stat = this->osdRecv ( pBufIn, nBytes, nActualBytes ); + inBufClient::fillCondition stat = + this->osdRecv ( pBufIn, nBytesToRecv, nActualBytes ); // // this is used to set the time stamp for write GDD's // @@ -2515,7 +2475,7 @@ bufSizeT casStrmClient::inBufBytesAvailable () const return this->in.bytesAvailable (); } -bufSizeT casStrmClient::outBufBytesPresent () const +bufSizeT casStrmClient :: outBytesPresent () const { epicsGuard < epicsMutex > guard ( this->mutex ); return this->out.bytesPresent (); diff --git a/src/cas/generic/casStrmClient.h b/src/cas/generic/casStrmClient.h index d0c987b4a..2596f7bb0 100644 --- a/src/cas/generic/casStrmClient.h +++ b/src/cas/generic/casStrmClient.h @@ -46,12 +46,12 @@ public: void userName ( char * pBuf, unsigned bufSize ) const; ca_uint16_t protocolRevision () const; void sendVersion (); + bufSizeT outBytesPresent () const; protected: caStatus processMsg (); bool inBufFull () const; bufSizeT inBufBytesAvailable () const; inBufClient::fillCondition inBufFill (); - bufSizeT outBufBytesPresent () const; private: char hostNameStr [32]; inBuf in; @@ -149,8 +149,8 @@ private: caStatus writeScalarData(); caStatus writeString(); - outBufClient::flushCondition xSend ( char * pBuf, bufSizeT nBytesAvailableToSend, - bufSizeT nBytesNeedToBeSent, bufSizeT & nBytesSent ); + outBufClient::flushCondition xSend ( char * pBuf, bufSizeT nBytesToSend, + bufSizeT & nBytesSent ); inBufClient::fillCondition xRecv ( char * pBuf, bufSizeT nBytesToRecv, inBufClient::fillParameter parm, bufSizeT & nByesRecv ); diff --git a/src/cas/generic/outBuf.cc b/src/cas/generic/outBuf.cc index d7eebad28..b97147253 100644 --- a/src/cas/generic/outBuf.cc +++ b/src/cas/generic/outBuf.cc @@ -15,11 +15,19 @@ */ #include "errlog.h" +#include "epicsTime.h" #define epicsExportSharedSymbols #include "outBuf.h" #include "osiWireFormat.h" +const char * outBufClient :: ppFlushCondText[3] = +{ + "flushNone", + "flushProgress", + "flushDisconnect" +}; + // // outBuf::outBuf() // @@ -62,11 +70,10 @@ caStatus outBuf::allocRawMsg ( bufSizeT msgsize, void **ppMsg ) stackNeeded = this->bufSize - msgsize; if ( this->stack > stackNeeded ) { - // // Try to flush the output queue // - this->flush ( this->stack - stackNeeded ); + this->flush (); // // If this failed then the fd is nonblocking @@ -216,45 +223,32 @@ void outBuf::commitMsg ( ca_uint32_t reducedPayloadSize ) // // outBuf::flush () // -outBufClient::flushCondition outBuf::flush ( bufSizeT spaceRequired ) +outBufClient::flushCondition outBuf :: flush () { - bufSizeT nBytes; - bufSizeT nBytesRequired; - outBufClient::flushCondition cond; - if ( this->ctxRecursCount > 0 ) { return outBufClient::flushNone; } - if ( spaceRequired > this->bufSize ) { - nBytesRequired = this->stack; - } - else { - bufSizeT stackPermitted; - - stackPermitted = this->bufSize - spaceRequired; - if ( this->stack > stackPermitted ) { - nBytesRequired = this->stack - stackPermitted; - } - else { - nBytesRequired = 0u; - } - } - - cond = this->client.xSend ( this->pBuf, this->stack, - nBytesRequired, nBytes ); + bufSizeT nBytesSent; + epicsTime beg = epicsTime::getCurrent (); + outBufClient :: flushCondition cond = + this->client.xSend ( this->pBuf, this->stack, nBytesSent ); + epicsTime end = epicsTime::getCurrent (); + printf ( "send of %u bytes, stat =%s, cost us %f u sec\n", + this->stack, this->client.ppFlushCondText[cond], ( end - beg ) * 1e6 ); if ( cond == outBufClient::flushProgress ) { - bufSizeT len; - - if ( nBytes >= this->stack ) { + if ( nBytesSent >= this->stack ) { this->stack = 0u; } else { - len = this->stack-nBytes; + bufSizeT len = this->stack - nBytesSent; // // memmove() is ok with overlapping buffers // - memmove ( this->pBuf, &this->pBuf[nBytes], len ); + epicsTime beg = epicsTime::getCurrent (); + memmove ( this->pBuf, &this->pBuf[nBytesSent], len ); + epicsTime end = epicsTime::getCurrent (); + printf ( "mem move cost us %f nano sec\n", ( end - beg ) * 1e9 ); this->stack = len; } @@ -262,7 +256,7 @@ outBufClient::flushCondition outBuf::flush ( bufSizeT spaceRequired ) char buf[64]; this->client.hostName ( buf, sizeof ( buf ) ); fprintf ( stderr, "CAS outgoing: %u byte reply to %s\n", - nBytes, buf ); + nBytesSent, buf ); } } return cond; diff --git a/src/cas/generic/outBuf.h b/src/cas/generic/outBuf.h index 69624aba9..dd1e62382 100644 --- a/src/cas/generic/outBuf.h +++ b/src/cas/generic/outBuf.h @@ -47,12 +47,18 @@ private: class outBufClient { // X aCC 655 public: - enum flushCondition { flushNone, flushProgress, flushDisconnect }; + enum flushCondition { + flushNone = 0, + flushProgress = 1, + flushDisconnect = 2 + }; + static const char * ppFlushCondText[3]; virtual unsigned getDebugLevel () const = 0; virtual void sendBlockSignal () = 0; - virtual flushCondition xSend ( char *pBuf, bufSizeT nBytesAvailableToSend, - bufSizeT nBytesNeedToBeSent, bufSizeT &nBytesSent ) = 0; + virtual flushCondition xSend ( char *pBuf, bufSizeT nBytesToSend, + bufSizeT & nBytesSent ) = 0; virtual void hostName ( char *pBuf, unsigned bufSize ) const = 0; + virtual bufSizeT osSendBufferSize () const = 0; protected: virtual ~outBufClient() {} }; @@ -72,7 +78,7 @@ public: // flush output queue // (returns the number of bytes sent) // - outBufClient::flushCondition flush ( bufSizeT spaceRequired=bufSizeT_MAX ); + outBufClient::flushCondition flush (); void show (unsigned level) const; @@ -128,13 +134,7 @@ private: // inline bufSizeT outBuf::bytesPresent () const { - // - // Note on use of lock here: - // This guarantees that any pushCtx() operation - // in progress completes before another thread checks. - // - bufSizeT result = this->stack; - return result; + return this->stack; } //