Merge lp:~matej-sekoranja/epics-base/pcas-variable-length-arrays into lp:~epics-core/epics-base/3.16

[Matej Sekoranja]

This branch adds variable length array support to pcas. Changes are mostly trivial - just replace msgHeader’s count with actual getElementCount(). Fine for monitors, however
read and readNotify (and first monitor event, that calls read) need additional code in the casPV::read(const casCtx &, gdd & protoIn) method, the method pCAS programmer needs to implement.

In order to do a “read” operation a DBR is created first (protoIn variable) and then the value is then copied into the “protoIn”. “protoIn” GDD structure contains “bounds” structure that specifies array size. If “bounds” are not specified, then scalar value is assumed.
If “0” count is given by the CA client then the GDD created has no “bounds” specified and this implies a scalar GDD. Consequently “read” will always return only one element.

So, the implementation of casPV::read(const casCtx &, gdd & protoIn) needs to check if “bounds” exists and create one if necessary (getBounds() == 0 && element count > 1):

if (!protoIn.getBounds() && this->info.getElementCount() > 1)
{
    // convert to atomic
    gddBounds bds;
    bds.setSize ( this->info.getElementCount() );
    bds.setFirst ( 0u );
    protoIn.setDimension ( 1u, & bds );
}

This check cannot be done in casStrmClient.cc since only the casPV implementation knows that the value actually is.

I think this is acceptable design: if “bounds” is not provided then casPV::read implementation needs to set one (to reflect current value “bounds”) - as shown above.

[Andrew Johnson (anj)]

What will happen if an unmodified application gets built against this version of the pcas code and a client subscribes (or does a ca_get_callback() ) to it with zero size?

This change will need documenting in the documentation/RELEASE_NOTES.html file. Since there isn't a document that describes the pcas API, all the detail of the necessary application changes should go there.

[Matej Sekoranja (matej-sekoranja)]

> What will happen if an unmodified application gets built against this version
> of the pcas code and a client subscribes (or does a ca_get_callback() ) to it
> with zero size?

get: only one element (scalar) will be returned
monitors: first event will return only one element (scalar), subsequent events will work as expected for "variable length array feature"

[Andrew Johnson (anj)]

Previously the CA client library converted monitor requests with zero size to mean all elements; the client knows the maximum array size since that information is sent out as part of the initial channel connection process. Would it be hard to change the pcas behavior so that an unmodified server application would send the max number of elements for the channel instead of just one, in both the get-callback and first monitor cases? I think that would be safer.

[Matej Sekoranja (matej-sekoranja)]

I need to get an indication whether casPV implementation support variable arrays or not.

I can create a method "virtual bool casPV::supports_variable_arrays() const { return false; }" with the default implementation as shown. If the method returns "false" then I will execute the following code before calling casPV::read :

if (!protoIn.getBounds() && this->info.getElementCount() > 1)
{
    // convert to atomic
    gddBounds bds;
    bds.setSize ( this->info.getElementCount() );
    bds.setFirst ( 0u );
    protoIn.setDimension ( 1u, & bds );
}

This will make code work as you've described above (using max number of elements for the PV when 0 count is requested).

OK to implement?

[Andrew Johnson (anj)]

Hi Matej,

I was hoping one of the other core developers with actual CAS experience would chime in on this question, but it doesn't look like anybody is going to, at least without some prompting. I've never written a full CAS application so I'm not ideally qualified to decide what the best adjustment of the API should be.

Here's my understanding of the issue:

Adding the variable-length-arrays feature to the CAS library should not change the behavior of an unmodified CAS application, although this branch currently does. To fix that, you are proposing adding a new virtual method to the casPV API that allows the library to ask each casPV object whether it supports variable length arrays. The default for an unmodified application would be no, in which case the CAS library will set up the protoIn gdd object that it passes into the application's PV::read() method for a zero-length read to request an array of the maximum size of the PV.

Would it be possible to always set the protoIn size to the max number of elements when handling a zero-length read? Presumably a casPV object's PV::read() method that supports variable length arrays can always make the array that it returns smaller than the request when there's less data available? If this works there should be no need for the additional method, although I admit it would reduce performance slightly (I don't think that bothers us though, we've been wanting to throw away GDD for a long time).

- Andrew

[Ralph Lange (ralph-lange)]

Why does the read() need to be changed in the first place?

If I remember correctly, when used with an rsrv (IOC) server, the variable length feature only applies to monitor updates. Reads and first replies always have the full size array.

Is there a requirement that the CAS implementation needs to support variable size arrays on read operations?

[Andrew Johnson (anj)]

No, ca_get_callback() also supports variable length arrays; ca_get() can't though because it has nowhere to return the data length. I suspect the read() operation gets called by the server library to fetch the data to be sent for monitor events as well as for get requests.

In any case, applying this change as written modifies the monitor behavior of an existing client. Start excas from a regular Base and do a camonitor on the alan PV. The first monitor event is the full length of the array (1000 elements IIRC). If you repeat that with excas built with this branch, the first monitor event is only a scalar.

[mdavidsaver (mdavidsaver)]

> Would it be possible to always set the protoIn size to the max number of elements when handling a zero-length read?

I think this is the appropriate strategy. This way existing code would see this as a request for max size, which it can handle w/o modification. So I would propose to do the merge once this point is reached w/o any API changes. This way we can get the protocol change in place.

[Bruce Hill (bhill)]

Sorry I'm so late to this party. Murali asked me to take a look at this last
summer and I didn't like the need to change the PV::read() to avoid getting just
one element on the first read. Then I got busy with other stuff and forgot
about this merge proposal altogether.

I've been wanting to find a fix for the var array issue w/ our gateways for some
time and finally dug into it Monday and found a fix. I was going to push it
to launchpad only to rediscover this merge from Matej.

My initial fix was to clip the number of elements to the current array size
in casStrmClient.cc, but I couldn't figure out why m_count was still coming in
w/ the full array size even when I specified zero. Turns out I'd neglected to
bump up EPICS_PROTOCOL_MINOR_VERSION in caHdrLargeArray.h.

After comparing our solutions, which were nearly the same, I boiled it down
to the following fix which creates the gdd w/ the max array size when zero is
passed in, thus avoiding the problem of no bounds.

=== modified file 'src/ca/legacy/pcas/generic/casStrmClient.cc'
--- src/ca/legacy/pcas/generic/casStrmClient.cc 2015-04-01 07:29:16 +0000
+++ src/ca/legacy/pcas/generic/casStrmClient.cc 2016-07-14 08:59:50 +0000
@@ -2612,10 +2612,16 @@
 {
     const caHdrLargeArray * pHdr = this->ctx.getMsg();

+ ca_uint32_t count = pHdr->m_count;
+ if ( count == 0 ) {
+ if ( this->ctx.getChannel() ) {
+ count = this->ctx.getChannel()->getPVI().nativeCount();
+ }
+ }
+
     {
         gdd * pDD = 0;
- caStatus status = createDBRDD ( pHdr->m_dataType,
- pHdr->m_count, pDD );
+ caStatus status = createDBRDD ( pHdr->m_dataType, count, pDD );
         if ( status != S_cas_success ) {
             return status;
     }

I've submitted this as a merge proposal to Matej's branch so it can be
brought in through his branch if everyone agrees.

I've tested this w/ a gateway app and it always returns the actual array size
when m_count is set to zero.

[Andrew Johnson (anj)]

Superseded by Micheal's pcas-fake-dynamic branch.

Unmerged revisions

12662. By Matej Sekoranja on 2015-04-01

added support for variable array length to read/readNotify and addEvent

=== modified file 'src/ca/legacy/pcas/generic/caHdrLargeArray.h'
--- src/ca/legacy/pcas/generic/caHdrLargeArray.h	2010-10-05 19:27:37 +0000
+++ src/ca/legacy/pcas/generic/caHdrLargeArray.h	2015-04-14 11:26:50 +0000
@@ -32,7 +32,7 @@
 #   include "shareLib.h"
 #endif
 
-static const unsigned char CA_MINOR_PROTOCOL_REVISION = 12;
+static const unsigned char CA_MINOR_PROTOCOL_REVISION = 13;
 
 typedef ca_uint32_t caResId;
 

=== modified file 'src/ca/legacy/pcas/generic/casStrmClient.cc'
--- src/ca/legacy/pcas/generic/casStrmClient.cc	2014-05-30 17:36:50 +0000
+++ src/ca/legacy/pcas/generic/casStrmClient.cc	2015-04-14 11:26:50 +0000
@@ -388,7 +388,7 @@
 //
 // casStrmClient::verifyRequest()
 //
-caStatus casStrmClient::verifyRequest ( casChannelI * & pChan )
+caStatus casStrmClient::verifyRequest ( casChannelI * & pChan, bool allowZeroCount )
 {
     const caHdrLargeArray * mp = this->ctx.getMsg();
 
@@ -411,7 +411,8 @@
     //
     // element count out of range ?
     //
-    if ( mp->m_count > pChan->getPVI().nativeCount() || mp->m_count == 0u ) {
+    if ( mp->m_count > pChan->getPVI().nativeCount() || 
+         ( !allowZeroCount && mp->m_count == 0u ) ) {
         return ECA_BADCOUNT;
     }
 
@@ -444,7 +445,7 @@
     casChannelI * pChan;
 
     {
-        caStatus status = this->verifyRequest ( pChan );
+        caStatus status = this->verifyRequest ( pChan, CA_V413 ( this->minor_version_number ) );
         if ( status != ECA_NORMAL ) {
             if ( pChan ) {
                 return this->sendErr ( guard, mp, pChan->getCID(), 
@@ -531,11 +532,15 @@
             pChan->getCID(), status, ECA_GETFAIL );
     }
 
+    ca_uint32_t count = (msg.m_count == 0) ?
+                            (ca_uint32_t)desc.getDataSizeElements() :
+                            msg.m_count;
+
     void * pPayload;
     {
-        unsigned payloadSize = dbr_size_n ( msg.m_dataType, msg.m_count );
+        unsigned payloadSize = dbr_size_n ( msg.m_dataType, count );
         caStatus localStatus = this->out.copyInHeader ( msg.m_cmmd, payloadSize,
-            msg.m_dataType, msg.m_count, pChan->getCID (), 
+            msg.m_dataType, count, pChan->getCID (), 
             msg.m_available, & pPayload );
         if ( localStatus ) {
             if ( localStatus==S_cas_hugeRequest ) {
@@ -551,21 +556,21 @@
     // (places the data in network format)
     //
     int mapDBRStatus = gddMapDbr[msg.m_dataType].conv_dbr(
-        pPayload, msg.m_count, desc, pChan->enumStringTable() );
+        pPayload, count, desc, pChan->enumStringTable() );
     if ( mapDBRStatus < 0 ) {
         desc.dump ();
         errPrintf ( S_cas_badBounds, __FILE__, __LINE__, "- get with PV=%s type=%u count=%u",
-                pChan->getPVI().getName(), msg.m_dataType, msg.m_count );
+                pChan->getPVI().getName(), msg.m_dataType, count );
         return this->sendErrWithEpicsStatus ( 
             guard, & msg, pChan->getCID(), S_cas_badBounds, ECA_GETFAIL );
     }
     int cacStatus = caNetConvert ( 
-        msg.m_dataType, pPayload, pPayload, true, msg.m_count );
+        msg.m_dataType, pPayload, pPayload, true, count );
     if ( cacStatus != ECA_NORMAL ) {
         return this->sendErrWithEpicsStatus ( 
             guard, & msg, pChan->getCID(), S_cas_internal, cacStatus );
     }
-    if ( msg.m_dataType == DBR_STRING && msg.m_count == 1u ) {
+    if ( msg.m_dataType == DBR_STRING && count == 1u ) {
         unsigned reducedPayloadSize = strlen ( static_cast < char * > ( pPayload ) ) + 1u;
         this->out.commitMsg ( reducedPayloadSize );
     }
@@ -585,7 +590,7 @@
     casChannelI * pChan;
 
     {
-        caStatus status = this->verifyRequest ( pChan );
+        caStatus status = this->verifyRequest ( pChan, CA_V413 ( this->minor_version_number ) );
         if ( status != ECA_NORMAL ) {
             return this->readNotifyFailureResponse ( guard, * mp, status );
         }
@@ -655,12 +660,16 @@
             guard, msg, ECA_GETFAIL );
         return ecaStatus;
     }
+    
+    ca_uint32_t count = (msg.m_count == 0) ?
+                            (ca_uint32_t)desc.getDataSizeElements() :
+                            msg.m_count;
 
     void *pPayload;
     {
-        unsigned size = dbr_size_n ( msg.m_dataType, msg.m_count );
+        unsigned size = dbr_size_n ( msg.m_dataType, count );
         caStatus status = this->out.copyInHeader ( msg.m_cmmd, size,
-                    msg.m_dataType, msg.m_count, ECA_NORMAL, 
+                    msg.m_dataType, count, ECA_NORMAL, 
                     msg.m_available, & pPayload );
         if ( status ) {
             if ( status == S_cas_hugeRequest ) {
@@ -675,23 +684,23 @@
     // convert gdd to db_access type
     //
     int mapDBRStatus = gddMapDbr[msg.m_dataType].conv_dbr ( pPayload, 
-        msg.m_count, desc, pChan->enumStringTable() );
+        count, desc, pChan->enumStringTable() );
     if ( mapDBRStatus < 0 ) {
         desc.dump();
         errPrintf ( S_cas_badBounds, __FILE__, __LINE__, 
             "- get notify with PV=%s type=%u count=%u",
-            pChan->getPVI().getName(), msg.m_dataType, msg.m_count );
+            pChan->getPVI().getName(), msg.m_dataType, count );
         return this->readNotifyFailureResponse ( guard, msg, ECA_NOCONVERT );
     }
 
     int cacStatus = caNetConvert ( 
-        msg.m_dataType, pPayload, pPayload, true, msg.m_count );
+        msg.m_dataType, pPayload, pPayload, true, count );
     if ( cacStatus != ECA_NORMAL ) {
         return this->sendErrWithEpicsStatus ( 
             guard, & msg, pChan->getCID(), S_cas_internal, cacStatus );
     }
 
-    if ( msg.m_dataType == DBR_STRING && msg.m_count == 1u ) {
+    if ( msg.m_dataType == DBR_STRING && count == 1u ) {
         unsigned reducedPayloadSize = strlen ( static_cast < char * > ( pPayload ) ) + 1u;
         this->out.commitMsg ( reducedPayloadSize );
     }
@@ -849,11 +858,15 @@
     casChannelI & chan, const caHdrLargeArray & msg, 
     const gdd & desc, const caStatus completionStatus )
 {
+    ca_uint32_t count = (msg.m_count == 0) ?
+                            (ca_uint32_t)desc.getDataSizeElements() :
+                            msg.m_count;
+
     void * pPayload = 0;
     {
-        ca_uint32_t size = dbr_size_n ( msg.m_dataType, msg.m_count );
+        ca_uint32_t size = dbr_size_n ( msg.m_dataType, count );
         caStatus status = out.copyInHeader ( msg.m_cmmd, size,
-            msg.m_dataType, msg.m_count, ECA_NORMAL, 
+            msg.m_dataType, count, ECA_NORMAL, 
             msg.m_available, & pPayload );
         if ( status ) {
             if ( status == S_cas_hugeRequest ) {
@@ -871,7 +884,7 @@
 
     gdd * pDBRDD = 0;
     if ( completionStatus == S_cas_success ) {
-        caStatus status = createDBRDD ( msg.m_dataType, msg.m_count, pDBRDD );
+        caStatus status = createDBRDD ( msg.m_dataType, count, pDBRDD );
         if ( status != S_cas_success ) {
             caStatus ecaStatus;
             if ( status == S_cas_badType ) {
@@ -892,7 +905,7 @@
                 pDBRDD->unreference ();
                 errPrintf ( S_cas_noConvert, __FILE__, __LINE__,
         "no conversion between event app type=%d and DBR type=%d Element count=%d",
-                    desc.applicationType (), msg.m_dataType, msg.m_count);
+                    desc.applicationType (), msg.m_dataType, count);
                 return monitorFailureResponse ( guard, msg, ECA_NOCONVERT );
             }
         }
@@ -915,14 +928,14 @@
     }
 
     int mapDBRStatus = gddMapDbr[msg.m_dataType].conv_dbr ( 
-        pPayload, msg.m_count, *pDBRDD, chan.enumStringTable() );
+        pPayload, count, *pDBRDD, chan.enumStringTable() );
     if ( mapDBRStatus < 0 ) {
         pDBRDD->unreference ();
         return monitorFailureResponse ( guard, msg, ECA_NOCONVERT );
     }
 
     int cacStatus = caNetConvert ( 
-        msg.m_dataType, pPayload, pPayload, true, msg.m_count );
+        msg.m_dataType, pPayload, pPayload, true, count );
     if ( cacStatus != ECA_NORMAL ) {
         pDBRDD->unreference ();
         return this->sendErrWithEpicsStatus ( 
@@ -932,7 +945,7 @@
     //
     // force string message size to be the true size 
     //
-    if ( msg.m_dataType == DBR_STRING && msg.m_count == 1u ) {
+    if ( msg.m_dataType == DBR_STRING && count == 1u ) {
         ca_uint32_t reducedPayloadSize = strlen ( static_cast < char * > ( pPayload ) ) + 1u;
         this->out.commitMsg ( reducedPayloadSize );
     }
@@ -1940,7 +1953,7 @@
 
     casChannelI *pciu;
     {
-        caStatus status = casStrmClient::verifyRequest ( pciu );
+        caStatus status = casStrmClient::verifyRequest ( pciu, CA_V413 ( this->minor_version_number ) );
         if ( status != ECA_NORMAL ) {
             if ( pciu ) {
                 return this->sendErr ( guard, mp, 

=== modified file 'src/ca/legacy/pcas/generic/casStrmClient.h'
--- src/ca/legacy/pcas/generic/casStrmClient.h	2010-08-10 21:05:46 +0000
+++ src/ca/legacy/pcas/generic/casStrmClient.h	2015-04-14 11:26:50 +0000
@@ -69,7 +69,7 @@
     bool responseIsPending;
 
     caStatus createChannel ( const char * pName );
-    caStatus verifyRequest ( casChannelI * & pChan );
+    caStatus verifyRequest ( casChannelI * & pChan, bool allowZeroCount = false );
     typedef caStatus ( casStrmClient :: * pCASMsgHandler ) 
         ( epicsGuard < casClientMutex > & );
     static pCASMsgHandler const msgHandlers[CA_PROTO_LAST_CMMD+1u];