Merge lp:~epics-core/epics-base/cahuge into lp:~epics-core/epics-base/3.16

[mdavidsaver]

Relaxation of EPICS_CA_MAX_ARRAY_BYTES. Instead of failing, attempt to handle large messages with a single allocation, if realloc() errors then fail as before. These large allocations aren't pooled, and are free()'d when a client disconnects.

What I've seen (eg. @ nsls2 and frib) is that users always set EPICS_CA_MAX_ARRAY_BYTES to be larger than the largest possible array in a facility (typically a camera image). This leads to inefficient use of buffers as only a small number of PVs over 16KB are actually this large. (as CAJ does this as well, I've wondered how much this behavior contributes to cs-studio's reputation as a memory hog)

This change would allow EPICS_CA_MAX_ARRAY_BYTES to be used as an optimization based on the most common array size.

I've left a hook (the NO_HUGE macro) to disable this in libca and RSRV on based on some condition.

[Andrew Johnson (anj)]

F2F meeting 3/14/2017: Change to use a run-time switch instead of NO_HUGE, default setting configurable by target, new behaviour: Don’t use free-list, always allocate large buffers on demand and release when client disconnects.

[mdavidsaver (mdavidsaver)]

superseded by https://code.launchpad.net/~epics-core/epics-base/+git/cahuge/+merge/320105

review: Disapprove (superseded)

Unmerged revisions

12748. By mdavidsaver on 2017-03-13

rsrv: support larger than max. array bytes

12747. By mdavidsaver on 2017-03-13

pcas: support larger than max array bytes

clientBufMemoryManager already supports allocations
larger than max array bytes, adjust callers inBuf/outBuf
to actually request larger allocations.

12746. By mdavidsaver on 2017-03-13

ca: support alloc larger than max array bytes

automatically try to allocate a custom buffer
when a message larger than ca max array bytes
is encountered.

=== modified file 'src/ca/client/tcpiiu.cpp'
--- src/ca/client/tcpiiu.cpp	2016-10-20 20:32:06 +0000
+++ src/ca/client/tcpiiu.cpp	2017-03-13 23:51:22 +0000
@@ -26,6 +26,9 @@
 
 #include <stdexcept>
 #include <string>
+
+#include <stdlib.h>
+
 #include "errlog.h"
 
 #define epicsExportSharedSymbols
@@ -1023,12 +1026,15 @@
 
     // free message body cache
     if ( this->pCurData ) {
-        if ( this->curDataMax == MAX_TCP ) {
+        if ( this->curDataMax <= MAX_TCP ) {
             this->cacRef.releaseSmallBufferTCP ( this->pCurData );
         }
-        else {
+        else if ( this->curDataMax <= cacRef.largeBufferSizeTCP()) {
             this->cacRef.releaseLargeBufferTCP ( this->pCurData );
         }
+        else {
+            free ( this->pCurData );
+        }
     }
 }
 
@@ -1197,18 +1203,44 @@
         // make sure we have a large enough message body cache
         //
         if ( this->curMsg.m_postsize > this->curDataMax ) {
-            if ( this->curDataMax == MAX_TCP && 
-                    this->cacRef.largeBufferSizeTCP() >= this->curMsg.m_postsize ) {
-                char * pBuf = this->cacRef.allocateLargeBufferTCP ();
-                if ( pBuf ) {
+            assert (this->curMsg.m_postsize > MAX_TCP);
+
+            char * newbuf = NULL;
+            arrayElementCount newsize;
+
+            if ( this->curMsg.m_postsize <= this->cacRef.largeBufferSizeTCP() ) {
+                newbuf = this->cacRef.allocateLargeBufferTCP ();
+                newsize = this->cacRef.largeBufferSizeTCP();
+            }
+#ifndef NO_HUGE
+            else {
+                // round size up to multiple of 4K
+                newsize = ((this->curMsg.m_postsize-1)|0xfff)+1;
+
+                if (this->curDataMax > this->cacRef.largeBufferSizeTCP()) {
+                    // expand existing huge buffer
+                    newbuf = (char*)realloc(this->pCurData, newsize);
+                } else {
+                    // trade up from small or large
+                    newbuf = (char*)malloc(newsize);
+                }
+            }
+#endif
+
+            if ( newbuf) {
+                if (this->curDataMax <= MAX_TCP) {
                     this->cacRef.releaseSmallBufferTCP ( this->pCurData );
-                    this->pCurData = pBuf;
-                    this->curDataMax = this->cacRef.largeBufferSizeTCP ();
-                }
-                else {
-                    this->printFormated ( mgr.cbGuard,
-                        "CAC: not enough memory for message body cache (ignoring response message)\n");
-                }
+                } else if (this->curDataMax <= this->cacRef.largeBufferSizeTCP()) {
+                    this->cacRef.releaseLargeBufferTCP ( this->pCurData );
+                } else {
+                    // called realloc()
+                }
+                this->pCurData = newbuf;
+                this->curDataMax = newsize;
+
+            } else {
+                this->printFormated ( mgr.cbGuard,
+                    "CAC: not enough memory for message body cache (ignoring response message)\n");
             }
         }
 
@@ -1426,7 +1458,7 @@
     }
     arrayElementCount maxBytes;
     if ( CA_V49 ( this->minorProtocolVersion ) ) {
-        maxBytes = this->cacRef.largeBufferSizeTCP ();
+        maxBytes = 0xfffffff0;
     }
     else {
         maxBytes = MAX_TCP;
@@ -1537,7 +1569,7 @@
         guard, CA_V413(this->minorProtocolVersion) );
     arrayElementCount maxBytes;
     if ( CA_V49 ( this->minorProtocolVersion ) ) {
-        maxBytes = this->cacRef.largeBufferSizeTCP ();
+        maxBytes = 0xfffffff0;
     }
     else {
         maxBytes = MAX_TCP;
@@ -1584,7 +1616,7 @@
         guard, CA_V413(this->minorProtocolVersion) );
     arrayElementCount maxBytes;
     if ( CA_V49 ( this->minorProtocolVersion ) ) {
-        maxBytes = this->cacRef.largeBufferSizeTCP ();
+        maxBytes = 0xfffffff0;
     }
     else {
         maxBytes = MAX_TCP;

=== modified file 'src/ca/legacy/pcas/generic/casStrmClient.cc'
--- src/ca/legacy/pcas/generic/casStrmClient.cc	2016-12-07 22:36:58 +0000
+++ src/ca/legacy/pcas/generic/casStrmClient.cc	2017-03-13 23:51:22 +0000
@@ -205,7 +205,7 @@
                 if ( bytesLeft < msgSize ) {
                     status = S_cas_success;
                     if ( msgSize > this->in.bufferSize() ) {
-                        this->in.expandBuffer ();
+                        this->in.expandBuffer (msgSize);
                         // msg to large - set up message drain
                         if ( msgSize > this->in.bufferSize() ) {
                             caServerI::dumpMsg ( this->pHostName, this->pUserName, & msgTmp, 0, 

=== modified file 'src/ca/legacy/pcas/generic/clientBufMemoryManager.cpp'
--- src/ca/legacy/pcas/generic/clientBufMemoryManager.cpp	2016-05-22 03:43:09 +0000
+++ src/ca/legacy/pcas/generic/clientBufMemoryManager.cpp	2017-03-13 23:51:22 +0000
@@ -16,11 +16,6 @@
 #define epicsExportSharedSymbols
 #include "clientBufMemoryManager.h"
 
-bufSizeT clientBufMemoryManager::maxSize () const
-{
-    return bufferFactory.largeBufferSize ();
-}
-
 casBufferParm clientBufMemoryManager::allocate ( bufSizeT newMinSize )
 {
     casBufferParm parm;

=== modified file 'src/ca/legacy/pcas/generic/clientBufMemoryManager.h'
--- src/ca/legacy/pcas/generic/clientBufMemoryManager.h	2003-02-12 19:06:15 +0000
+++ src/ca/legacy/pcas/generic/clientBufMemoryManager.h	2017-03-13 23:51:22 +0000
@@ -39,9 +39,9 @@
 
 class clientBufMemoryManager {
 public:
+    //! @throws std::bad_alloc on failure
     casBufferParm allocate ( bufSizeT newMinSize );
     void release ( char * pBuf, bufSizeT bufSize );
-    bufSizeT maxSize () const;
 private:
     casBufferFactory bufferFactory;
 };

=== modified file 'src/ca/legacy/pcas/generic/inBuf.cc'
--- src/ca/legacy/pcas/generic/inBuf.cc	2016-05-22 12:38:18 +0000
+++ src/ca/legacy/pcas/generic/inBuf.cc	2017-03-13 23:51:22 +0000
@@ -13,6 +13,8 @@
  *              505 665 1831
  */
 
+#include <stdexcept>
+
 #include <stdio.h>
 #include <string.h>
 
@@ -155,11 +157,17 @@
     }
 }
 
-void inBuf::expandBuffer ()
+void inBuf::expandBuffer (bufSizeT needed)
 {
-    bufSizeT max = this->memMgr.maxSize();
-    if ( this->bufSize <  max ) {
-        casBufferParm bufParm = this->memMgr.allocate ( max );
+    if (needed > bufSize) {
+        casBufferParm bufParm;
+        try {
+            bufParm = this->memMgr.allocate ( needed );
+        } catch (std::bad_alloc& e) {
+            // caller must check that buffer size has expended
+            return;
+        }
+
         bufSizeT unprocessedBytes = this->bytesPresent ();
         memcpy ( bufParm.pBuf, &this->pBuf[this->nextReadIndex], unprocessedBytes );
         this->bytesInBuffer = unprocessedBytes;
@@ -170,7 +178,7 @@
     }
 }
 
-unsigned inBuf::bufferSize () const
+bufSizeT inBuf::bufferSize() const
 {
     return this->bufSize;
 }

=== modified file 'src/ca/legacy/pcas/generic/inBuf.h'
--- src/ca/legacy/pcas/generic/inBuf.h	2012-04-12 16:13:50 +0000
+++ src/ca/legacy/pcas/generic/inBuf.h	2017-03-13 23:51:22 +0000
@@ -82,8 +82,8 @@
     //
     const inBufCtx pushCtx ( bufSizeT headerSize, bufSizeT bodySize );
 	bufSizeT popCtx ( const inBufCtx & ); // returns actual size
-    unsigned bufferSize () const;
-    void expandBuffer ();
+    bufSizeT bufferSize () const;
+    void expandBuffer (bufSizeT needed);
 private:
     class inBufClient & client;
     class clientBufMemoryManager & memMgr;

=== modified file 'src/ca/legacy/pcas/generic/outBuf.cc'
--- src/ca/legacy/pcas/generic/outBuf.cc	2016-05-22 12:38:18 +0000
+++ src/ca/legacy/pcas/generic/outBuf.cc	2017-03-13 23:51:22 +0000
@@ -59,7 +59,7 @@
     msgsize = CA_MESSAGE_ALIGN ( msgsize );
 
     if ( msgsize > this->bufSize ) {
-        this->expandBuffer ();
+        this->expandBuffer (msgsize);
         if ( msgsize > this->bufSize ) {
             return S_cas_hugeRequest;
         }
@@ -316,11 +316,17 @@
     }
 }
 
-void outBuf::expandBuffer ()
+void outBuf::expandBuffer (bufSizeT needed)
 {
-    bufSizeT max = this->memMgr.maxSize();
-    if ( this->bufSize < max  ) {
-        casBufferParm bufParm = this->memMgr.allocate ( max );
+    if (needed > bufSize) {
+        casBufferParm bufParm;
+        try {
+            bufParm = this->memMgr.allocate ( needed );
+        } catch (std::bad_alloc& e) {
+            // caller must check that buffer size has expended
+            return;
+        }
+
         memcpy ( bufParm.pBuf, this->pBuf, this->stack );
         this->memMgr.release ( this->pBuf, this->bufSize );
         this->pBuf = bufParm.pBuf;

=== modified file 'src/ca/legacy/pcas/generic/outBuf.h'
--- src/ca/legacy/pcas/generic/outBuf.h	2012-04-12 16:13:50 +0000
+++ src/ca/legacy/pcas/generic/outBuf.h	2017-03-13 23:51:22 +0000
@@ -122,7 +122,7 @@
 	bufSizeT stack;
     unsigned ctxRecursCount;
 
-    void expandBuffer ();
+    void expandBuffer (bufSizeT needed);
 
 	outBuf ( const outBuf & );
 	outBuf & operator = ( const outBuf & );

=== modified file 'src/ioc/rsrv/caservertask.c'
--- src/ioc/rsrv/caservertask.c	2017-01-23 23:20:51 +0000
+++ src/ioc/rsrv/caservertask.c	2017-03-13 23:51:22 +0000
@@ -1060,6 +1060,9 @@
             else if ( client->send.type == mbtLargeTCP ) {
                 freeListFree ( rsrvLargeBufFreeListTCP,  client->send.buf );
             }
+            else if (client->send.type == mbtHugeTCP ) {
+                free ( client->send.buf );
+            }
             else {
                 errlogPrintf ( "CAS: Corrupt send buffer free list type code=%u during client cleanup?\n",
                     client->send.type );
@@ -1072,6 +1075,9 @@
             else if ( client->recv.type == mbtLargeTCP ) {
                 freeListFree ( rsrvLargeBufFreeListTCP,  client->recv.buf );
             }
+            else if (client->recv.type == mbtHugeTCP ) {
+                free ( client->recv.buf );
+            }
             else {
                 errlogPrintf ( "CAS: Corrupt recv buffer free list type code=%u during client cleanup?\n",
                     client->send.type );
@@ -1301,43 +1307,84 @@
     taskwdInsert ( pClient->tid, NULL, NULL );
 }
 
+static
+void casExpandBuffer ( struct message_buffer *buf, ca_uint32_t size, int sendbuf )
+{
+    char *newbuf = NULL;
+    unsigned newsize;
+    enum messageBufferType newtype;
+
+    assert (size > MAX_TCP);
+
+    if ( size <= buf->maxstk || buf->type == mbtUDP ) return;
+
+    /* try to alloc new buffer */
+    if (size <= MAX_TCP) {
+        return; /* shouldn't happen */
+    } else if (size <= rsrvSizeofLargeBufTCP) {
+        newbuf = freeListCalloc ( rsrvLargeBufFreeListTCP );
+        newsize = rsrvSizeofLargeBufTCP;
+        newtype = mbtLargeTCP;
+    }
+#ifndef NO_HUGE
+    else {
+        size = ((size-1)|0xfff)+1;
+
+        if (buf->type==mbtHugeTCP)
+            newbuf = realloc (buf->buf, size);
+        else
+            newbuf = malloc (size);
+        newtype = mbtHugeTCP;
+        newsize = size;
+    }
+#endif
+
+    if (newbuf) {
+        /* copy existing buffer */
+        if (sendbuf) {
+            /* send buffer uses [0, stk) */
+            if (buf->type==mbtHugeTCP) {
+                /* realloc already copied */
+            } else {
+                memcpy ( newbuf, buf->buf, buf->stk );
+            }
+        } else {
+            /* recv buffer uses [stk, cnt) */
+            unsigned used;
+            assert ( buf->cnt >= buf->stk );
+            used = buf->cnt - buf->stk;
+
+            /* buf->buf may be the same as newbuf if realloc() used */
+            memmove ( newbuf, &buf->buf[buf->stk], used );
+
+            buf->cnt = used;
+            buf->stk = 0;
+
+        }
+
+        /* free existing buffer */
+        if(buf->type==mbtSmallTCP) {
+            freeListFree ( rsrvSmallBufFreeListTCP,  buf->buf );
+        } else if(buf->type==mbtLargeTCP) {
+            freeListFree ( rsrvLargeBufFreeListTCP,  buf->buf );
+        } else {
+            /* realloc() already free()'d if necessary */
+        }
+
+        buf->buf = newbuf;
+        buf->type = newtype;
+        buf->maxstk = newsize;
+    }
+}
+
 void casExpandSendBuffer ( struct client *pClient, ca_uint32_t size )
 {
-    if ( pClient->send.type == mbtSmallTCP && rsrvSizeofLargeBufTCP > MAX_TCP
-            && size <= rsrvSizeofLargeBufTCP ) {
-        int spaceAvailOnFreeList = freeListItemsAvail ( rsrvLargeBufFreeListTCP ) > 0;
-        if ( osiSufficentSpaceInPool(rsrvSizeofLargeBufTCP) || spaceAvailOnFreeList ) {
-            char *pNewBuf = ( char * ) freeListCalloc ( rsrvLargeBufFreeListTCP );
-            if ( pNewBuf ) {
-                memcpy ( pNewBuf, pClient->send.buf, pClient->send.stk );
-                freeListFree ( rsrvSmallBufFreeListTCP,  pClient->send.buf );
-                pClient->send.buf = pNewBuf;
-                pClient->send.maxstk = rsrvSizeofLargeBufTCP;
-                pClient->send.type = mbtLargeTCP;
-            }
-        }
-    }
+    casExpandBuffer (&pClient->send, size, 1);
 }
 
 void casExpandRecvBuffer ( struct client *pClient, ca_uint32_t size )
 {
-    if ( pClient->recv.type == mbtSmallTCP && rsrvSizeofLargeBufTCP > MAX_TCP
-            && size <= rsrvSizeofLargeBufTCP) {
-        int spaceAvailOnFreeList = freeListItemsAvail ( rsrvLargeBufFreeListTCP ) > 0;
-        if ( osiSufficentSpaceInPool(rsrvSizeofLargeBufTCP) || spaceAvailOnFreeList ) {
-            char *pNewBuf = ( char * ) freeListCalloc ( rsrvLargeBufFreeListTCP );
-            if ( pNewBuf ) {
-                assert ( pClient->recv.cnt >= pClient->recv.stk );
-                memcpy ( pNewBuf, &pClient->recv.buf[pClient->recv.stk], pClient->recv.cnt - pClient->recv.stk );
-                freeListFree ( rsrvSmallBufFreeListTCP,  pClient->recv.buf );
-                pClient->recv.buf = pNewBuf;
-                pClient->recv.cnt = pClient->recv.cnt - pClient->recv.stk;
-                pClient->recv.stk = 0u;
-                pClient->recv.maxstk = rsrvSizeofLargeBufTCP;
-                pClient->recv.type = mbtLargeTCP;
-            }
-        }
-    }
+    casExpandBuffer (&pClient->recv, size, 0);
 }
 
 /*

=== modified file 'src/ioc/rsrv/server.h'
--- src/ioc/rsrv/server.h	2016-05-13 18:12:08 +0000
+++ src/ioc/rsrv/server.h	2017-03-13 23:51:22 +0000
@@ -60,7 +60,7 @@
  * Eight-byte alignment is required by the Sparc 5 and other RISC
  * processors.
  */
-enum messageBufferType { mbtUDP, mbtSmallTCP, mbtLargeTCP };
+enum messageBufferType { mbtUDP, mbtSmallTCP, mbtLargeTCP, mbtHugeTCP };
 struct message_buffer {
   char                      *buf;
   unsigned                  stk;