on vxworks, return value of sscanf with %* formats is buggy

Makefile

@@ -24,13 +24,13 @@ RECORDTYPES += mbboDirect mbbiDirect
 RECORDTYPES += longout longin
 RECORDTYPES += stringout stringin
 RECORDTYPES += waveform
-RECORDTYPES += calcout
 
 SOURCES += $(RECORDTYPES:%=src/dev%Stream.c)
 SOURCES += $(FORMATS:%=src/%Converter.cc)
 SOURCES += $(BUSSES:%=src/%Interface.cc)
 SOURCES += $(wildcard src/Stream*.cc)
 SOURCES += src/StreamVersion.c
+SOURCES_3.14 += src/devcalcoutStream.c
 
 ifeq (${EPICS_BASETYPE},3.13)
 USR_INCLUDES += -include $(INSTALL_INCLUDE)/compat3_13.h
@@ -56,7 +56,7 @@ stream.dbd:
 	@echo "driver(stream)" >> $@
 else
 stream.dbd:
-	@for r in $(RECORDTYPES); \
+	@for r in $(RECORDTYPES) calcout; \
 	do echo "device($$r,INST_IO,dev$${r}Stream,\"stream\")"; \
 	done > $@
 	@echo "driver(stream)" >> $@

src/AsynDriverInterface.cc

@@ -765,7 +765,7 @@ readHandler()
     {
         // In AsyncRead mode just poll
         // and read as much as possible
-        pasynUser->timeout = readTimeout;
+        pasynUser->timeout = 0.0;
         bytesToRead = buffersize;
     }
     else
@@ -786,7 +786,7 @@ readHandler()
         
         status = pasynOctet->read(pvtOctet, pasynUser,
             buffer, bytesToRead, (size_t*)&received, &eomReason);
-        if (ioAction != AsyncRead || status != asynTimeout)
+        if (ioAction == Read || status != asynTimeout)
         {
             debug("AsynDriverInterface::readHandler(%s): "
                 "read(..., bytesToRead=%d, ...) [timeout=%f seconds] = %s\n",
@@ -798,7 +798,7 @@ readHandler()
         switch (status)
         {
             case asynSuccess:
-                if (ioAction == AsyncRead)
+                if (ioAction != Read)
                 {
 #ifndef NO_TEMPORARY
                     debug("AsynDriverInterface::readHandler(%s): "
@@ -950,6 +950,8 @@ void intrCallbackOctet(void* /*pvt*/, asynUser *pasynUser,
 //    internal buffer of asynDriver.
 
     if (!interruptAccept) return; // too early to process records
+    debug("AsynDriverInterface::intrCallbackOctet(%s) ioAction = %s\n",
+        interface->clientName(), ioActionStr[interface->ioAction]);
     if (interface->ioAction == AsyncRead ||
         interface->ioAction == AsyncReadMore)
     {
@@ -1287,8 +1289,8 @@ void handleRequest(asynUser* pasynUser)
 {
     AsynDriverInterface* interface =
         static_cast<AsynDriverInterface*>(pasynUser->userPvt);
-    debug("AsynDriverInterface::handleRequest(%s)\n",
-        interface->clientName());
+    debug("AsynDriverInterface::handleRequest(%s) %s\n",
+        interface->clientName(), ioActionStr[interface->ioAction]);
     switch (interface->ioAction)
     {
         case None:

src/CONFIG_STREAM

@@ -41,6 +41,7 @@ FORMATS += Raw
 FORMATS += Binary
 FORMATS += Checksum
 FORMATS += Exponential
+FORMATS += RawFloat
 
 # Want Perl regular expression matching?
 # If PCRE is installed at the same location for all

src/Makefile

@@ -108,7 +108,6 @@ $(COMMON_DIR)/$(LIBRARY_DEFAULT).dbd: ../CONFIG_STREAM
 	done > $@
 	@echo "driver(stream)" >> $@
 	@echo "variable(streamDebug, int)" >> $@
-	@echo "variable(showAsyncErrors, int)" >> $@
 	@echo "registrar(streamRegistrar)" >> $@
 
 endif

src/StreamBusInterface.cc

@@ -20,7 +20,11 @@
 #include "StreamBusInterface.h"
 
 const char* StreamIoStatusStr[] = {
-    "StreamIoSuccess", "ioTimeout", "ioNoReply", "ioEnd", "ioFault"
+    "StreamIoSuccess",
+    "StreamIoTimeout",
+    "StreamIoNoReply",
+    "StreamIoEnd",
+    "StreamIoFault"
 };
 
 StreamBusInterfaceRegistrarBase* StreamBusInterfaceRegistrarBase::first;

src/StreamCore.cc

@@ -897,8 +897,14 @@ readCallback(StreamIoStatus status,
 
     if (!(flags & AcceptInput))
     {
-        error("StreamCore::readCallback(%s) called unexpectedly\n",
-            name());
+#ifdef NO_TEMPORARY
+        error("StreamCore::readCallback(%s, %s) called unexpectedly\n",
+            name(), StreamIoStatusStr[status]);
+#else
+        error("StreamCore::readCallback(%s, %s, \"%s\") called unexpectedly\n",
+            name(), StreamIoStatusStr[status],
+            StreamBuffer(input, size).expand()());
+#endif
         return 0;
     }
     flags &= ~AcceptInput;

src/StreamEpics.cc

@@ -41,6 +41,7 @@ extern "C" {
 
 #include <semLib.h>
 #include <wdLib.h>
+#include <taskLib.h>
 
 extern DBBASE *pdbbase;
 
@@ -52,6 +53,7 @@ extern DBBASE *pdbbase;
 #include <epicsMutex.h>
 #include <epicsEvent.h>
 #include <epicsTime.h>
+#include <epicsThread.h>
 #include <registryFunction.h>
 #include <iocsh.h>
 
@@ -76,10 +78,12 @@ epicsShareFunc int epicsShareAPI iocshCmd(const char *command);
 enum MoreFlags {
     // 0x00FFFFFF used by StreamCore
     InDestructor  = 0x0100000,
-    ValueReceived = 0x0200000
+    ValueReceived = 0x0200000,
+    Aborted       = 0x0400000
 };
 
 extern "C" void streamExecuteCommand(CALLBACK *pcallback);
+extern "C" void streamRecordProcessCallback(CALLBACK *pcallback);
 extern "C" long streamReload(char* recordname);
 
 class Stream : protected StreamCore
@@ -108,6 +112,7 @@ class Stream : protected StreamCore
     long currentValueLength;
     IOSCANPVT ioscanpvt;
     CALLBACK commandCallback;
+    CALLBACK processCallback;
 
 
 #ifdef EPICS_3_14
@@ -118,7 +123,7 @@ class Stream : protected StreamCore
 #endif
 
 // StreamCore methods
-    // void protocolStartHook(); // Nothing to do here?
+    void protocolStartHook();
     void protocolFinishHook(ProtocolResult);
     void startTimer(unsigned long timeout);
     bool getFieldAddress(const char* fieldname,
@@ -131,6 +136,7 @@ class Stream : protected StreamCore
     void releaseMutex();
     bool execute();
     friend void streamExecuteCommand(CALLBACK *pcallback);
+    friend void streamRecordProcessCallback(CALLBACK *pcallback);
 
 // Stream Epics methods
     long initRecord();
@@ -280,12 +286,15 @@ epicsExportAddress(drvet, stream);
 
 void streamEpicsPrintTimestamp(char* buffer, int size)
 {
+    int tlen;
     epicsTime tm = epicsTime::getCurrent();
-    tm.strftime(buffer, size, "%Y/%m/%d %H:%M:%S.%03f");
+    tlen = tm.strftime(buffer, size, "%Y/%m/%d %H:%M:%S.%03f");
+    sprintf(buffer+tlen, " %.*s", size-tlen-2, epicsThreadGetNameSelf());
 }
 #else
 void streamEpicsPrintTimestamp(char* buffer, int size)
 {
+    int tlen;
     char* c;
     TS_STAMP tm;
     tsLocalTime (&tm);
@@ -294,6 +303,8 @@ void streamEpicsPrintTimestamp(char* buffer, int size)
     if (c) {
         c[4] = 0;
     }
+    tlen = strlen(buffer);
+    sprintf(buffer+tlen, " %.*s", size-tlen-2, taskName(0));
 }
 #endif
 
@@ -310,7 +321,7 @@ report(int interest)
         printf("    %s\n", interface.name());
         ++interface;
     }
-    
+
     if (interest < 1) return OK;
     printf("  registered converters:\n");
     StreamFormatConverter* converter;
@@ -323,7 +334,7 @@ report(int interest)
             printf("    %%%c %s\n", c, converter->name());
         }
     }
-    
+
     Stream* pstream;
     printf("  connected records:\n");
     for (pstream = static_cast<Stream*>(first); pstream;
@@ -525,6 +536,8 @@ Stream(dbCommon* _record, struct link *ioLink,
 #endif
     callbackSetCallback(streamExecuteCommand, &commandCallback);
     callbackSetUser(this, &commandCallback);
+    callbackSetCallback(streamRecordProcessCallback, &processCallback);
+    callbackSetUser(this, &processCallback);
     status = ERROR;
     convert = DO_NOT_CONVERT;
     ioscanpvt = NULL;
@@ -561,14 +574,14 @@ initRecord()
     // scan link parameters: filename protocol busname addr busparam
     // It is safe to call this function again with different
     // link text or different protocol file.
-    
+
     char filename[80];
     char protocol[80];
     char busname[80];
     int addr = -1;
     char busparam[80];
     int n;
-    
+
     if (ioLink->type != INST_IO)
     {
         error("%s: Wrong link type %s\n", name(),
@@ -790,6 +803,12 @@ expire(CALLBACK *pcallback)
 
 // StreamCore virtual methods ////////////////////////////////////////////
 
+void Stream::
+protocolStartHook()
+{
+    flags &= ~Aborted;
+}
+
 void Stream::
 protocolFinishHook(ProtocolResult result)
 {
@@ -824,6 +843,7 @@ protocolFinishHook(ProtocolResult result)
             status = CALC_ALARM;
             break;
         case Abort:
+            flags |= Aborted;
         case Fault:
             status = UDF_ALARM;
             if (record->pact || record->scan == SCAN_IO_EVENT)
@@ -845,28 +865,46 @@ protocolFinishHook(ProtocolResult result)
 #endif
         return;
     }
-    if (record->pact || record->scan == SCAN_IO_EVENT)
-    {
-        debug("Stream::protocolFinishHook(stream=%s,result=%d) "
-            "processing record\n", name(), result);
-        // process record
-        // This will call streamReadWrite.
-        dbScanLock(record);
-        ((DEVSUPFUN)record->rset->process)(record);
-        dbScanUnlock(record);
 
-        debug("Stream::protocolFinishHook(stream=%s,result=%d) done\n",
-            name(), result);
-    }
     if (result != Abort && record->scan == SCAN_IO_EVENT)
+    {
+        // re-enable early input
+        flags |= AcceptInput;
+    }
+
+    if (record->pact || record->scan == SCAN_IO_EVENT)
+    {
+        // process record in callback thread to break possible recursion
+        callbackSetPriority(priority(), &processCallback);
+        callbackRequest(&processCallback);
+    }
+
+}
+
+void streamRecordProcessCallback(CALLBACK *pcallback)
+{
+    Stream* pstream = static_cast<Stream*>(pcallback->user);
+    dbCommon* record = pstream->record;
+
+    // process record
+    // This will call streamReadWrite.
+    debug("streamRecordProcessCallback(%s) processing record\n",
+            pstream->name());
+    dbScanLock(record);
+    ((DEVSUPFUN)record->rset->process)(record);
+    dbScanUnlock(record);
+    debug("streamRecordProcessCallback(%s) processing record done\n",
+            pstream->name());
+
+    if (record->scan == SCAN_IO_EVENT && !(pstream->flags & Aborted))
     {
         // restart protocol for next turn
-        debug("Stream::process(%s) restart async protocol\n",
-            name());
-        if (!startProtocol(StartAsync))
+        debug("streamRecordProcessCallback(%s) restart async protocol\n",
+            pstream->name());
+        if (!pstream->startProtocol(Stream::StartAsync))
         {
             error("%s: Can't restart \"I/O Intr\" protocol\n",
-                name());
+                pstream->name());
         }
     }
 }
@@ -1003,7 +1041,7 @@ matchValue(const StreamFormat& format, const void* fieldaddress)
     char* buffer;
     int status;
     const char* putfunc;
-    
+
     if (fieldaddress)
     {
         // Format like "%([record.]field)..." has requested to put value
@@ -1161,7 +1199,7 @@ noMoreElements:
 void streamExecuteCommand(CALLBACK *pcallback)
 {
     Stream* pstream = static_cast<Stream*>(pcallback->user);
-    
+
     if (iocshCmd(pstream->outputLine()) != OK)
     {
         pstream->execCallback(StreamIoFault);
@@ -1178,7 +1216,7 @@ extern "C" int execute(const char *cmd);
 void streamExecuteCommand(CALLBACK *pcallback)
 {
     Stream* pstream = static_cast<Stream*>(pcallback->user);
-    
+
     if (execute(pstream->outputLine()) != OK)
     {
         pstream->execCallback(StreamIoFault);

src/StreamFormatConverter.cc

@@ -160,7 +160,8 @@ scanLong(const StreamFormat& fmt, const char* input, long& value)
     int length = -1;
     if (fmt.flags & skip_flag)
     {
-        if (sscanf(input, fmt.info, &length) < 0) return -1;
+        /* on vxWorks, return value of sscanf with %*... is buggy */
+        sscanf(input, fmt.info, &length);
     }
     else
     {
@@ -216,7 +217,8 @@ scanDouble(const StreamFormat& fmt, const char* input, double& value)
     int length = -1;
     if (fmt.flags & skip_flag)
     {
-        if (sscanf(input, fmt.info, &length) < 0) return -1;
+        /* on vxWorks, return value of sscanf with %*... is buggy */
+        sscanf(input, fmt.info, &length);
     }
     else
     {
@@ -279,7 +281,8 @@ scanString(const StreamFormat& fmt, const char* input,
     }
     if (fmt.flags & skip_flag)
     {
-        if (sscanf(input, fmt.info, &length) < 0) return -1;
+        /* on vxWorks, return value of sscanf with %*... is buggy */
+        sscanf(input, fmt.info, &length);
     }
     else
     {
@@ -406,7 +409,8 @@ scanString(const StreamFormat& fmt, const char* input,
     int length = -1;
     if (fmt.flags & skip_flag)
     {
-        if (sscanf (input, fmt.info, &length) < 0) return -1;
+        /* on vxWorks, return value of sscanf with %*... is buggy */
+        sscanf(input, fmt.info, &length);
     }
     else
     {