epics/pcas
0
0
Fork 0
Code Issues Pull Requests Projects Wiki Activity

simplified tcp send interactions

Browse Source
This commit is contained in:
Jeff Hill
2009-07-25 00:43:28 +00:00
parent 4ade695a60
commit 6eb25148c5
6 changed files with 55 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cas/generic/casDGClient.cc
+5 -8
View File
@@ -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 );
src/cas/generic/casDGClient.h
+2 -2
View File
@@ -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 (
src/cas/generic/casStrmClient.cc
+10 -50
View File
@@ -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 ();
src/cas/generic/casStrmClient.h
+3 -3
View File
@@ -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 );
src/cas/generic/outBuf.cc
+24 -30
View File
@@ -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;
src/cas/generic/outBuf.h
+11 -11
View File
@@ -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;
}
//