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

feature: little endian support

Makefile

@@ -3,7 +3,7 @@ EXCLUDE_VERSIONS = 3.13.2
 PROJECT=stream2
 BUILDCLASSES += Linux
 
-DOCUDIR = doc
+#DOCUDIR = doc
 
 DBDS = stream.dbd
 
@@ -11,6 +11,7 @@ BUSSES  += AsynDriver
 FORMATS += Enum
 FORMATS += BCD
 FORMATS += Raw
+FORMATS += RawFloat
 FORMATS += Binary
 FORMATS += Checksum
 FORMATS += Regexp
@@ -23,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
@@ -55,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)" >> $@

doc/formats.html

@@ -240,6 +240,8 @@ The <code>%b</code> format uses the characters <code>0</code> and
 <code>1</code>.
 With the <code>%B</code> format, you can choose two other characters
 to represent zero and one.
+With the <code>#</code> flag, the bit order is changed to <em>little
+endian</em>, i.e. least significant bit first.
 </p>
 <p>
 Examples: <code>%B.!</code> or <code>%B\x00\xff</code>.
@@ -264,23 +266,28 @@ one character.
 <h2>9. Raw LONG Converter (<code>%r</code>)</h2>
 <p>
 The raw converter does not really "convert".
-A signed integer value is written or read in the internal
+A signed or unsigned integer value is written or read in the internal
 (usually two's complement) representation of the computer.
 The normal byte order is <em>big endian</em>, i.e. most significant byte
 first.
 With the <code>#</code> flag, the byte order is changed to <em>little
 endian</em>, i.e. least significant byte first.
+With the <code>0</code> flag, the value is unsigned, otherwise signed.
 </p>
 <p>
 In output, the <em>width</em> least significant bytes of the value
 are written.
 If <em>width</em> is larger than the size of a <code>long</code>,
-the value is sign extended.
+the value is sign extended or zero extended, depending on the
+<code>0</code> flag.
 </p>
 <p>
 In input, <em>width</em> bytes are read and put into the value.
-If <em>width</em> is longer than the size of a <code>long</code>, only
+If <em>width</em> is larger than the size of a <code>long</code>, only
 the least significant bytes are used.
+If <em>width</em> is smaller than the size of a <code>long</code>,
+the value is sign extended or zero extended, depending on the
+<code>0</code> flag.
 </p>
 
 <a name="bcd"></a>
@@ -451,6 +458,20 @@ the title of an HTML page, skipps anything before the
 <code>&lt;title&gt</code> tag and leaves anything after the
 <code>&lt;/title&gt;</code> tag in the input buffer.
 </p>
+<a name="exp"></a>
+<h2>13. Exponential DOUBLE converter (<code>%m</code>)</h2>
+<p>
+This experimental input-only format matches numbers in the format
+<i>[sign] mantissa sign exponent<i>, e.g <code>+123-4</code> meaning
+123e-4 or 0.0123. Mantissa and exponent are integers. The sign of the
+mantissa is optional. Compared to the standard <code>%e</code> format,
+this format does not contain the characters <code>.</code> and <code>e</code>.
+</p>
+<p>
+Output formatting is ambigous (e.g. <code>123-4</code> versus
+<code>1230-5</code>). Until it has been specified how to handle this,
+output is not supported.
+</p>
 <hr>
 <p align="right"><a href="processing.html">Next: Record Processing</a></p>
 <p><small>Dirk Zimoch, 2007</small></p>

doc/nav.html

@@ -121,6 +121,7 @@ h1        {font-size:120%;
    <li><a target="text" href="formats.html#bcd">%D</a></li>
    <li><a target="text" href="formats.html#chksum">%&lt;<em>checksum</em>&gt;</a></li>
    <li><a target="text" href="formats.html#regex">%/<em>regex</em>/</a></li>
+   <li><a target="text" href="formats.html#exp">%m</a></li>
   </ul>
  </div>
  <div class="top">

src/AsynDriverInterface.cc

@@ -458,12 +458,22 @@ connectToBus(const char* busname, int addr)
 bool AsynDriverInterface::
 lockRequest(unsigned long lockTimeout_ms)
 {
+    int connected;
+    asynStatus status;
+    
     debug("AsynDriverInterface::lockRequest(%s, %ld msec)\n",
         clientName(), lockTimeout_ms);
-    asynStatus status;
     lockTimeout = lockTimeout_ms ? lockTimeout_ms*0.001 : -1.0;
     ioAction = Lock;
-    status = pasynManager->queueRequest(pasynUser, priority(),
+    status = pasynManager->isConnected(pasynUser, &connected);
+    if (status != asynSuccess)
+    {
+        error("%s: pasynManager->isConnected() failed: %s\n",
+            clientName(), pasynUser->errorMessage);
+        return false;
+    }
+    status = pasynManager->queueRequest(pasynUser,
+        connected ? priority() : asynQueuePriorityConnect,
         lockTimeout);
     if (status != asynSuccess)
     {
@@ -755,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
@@ -776,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",
@@ -788,7 +798,7 @@ readHandler()
         switch (status)
         {
             case asynSuccess:
-                if (ioAction == AsyncRead)
+                if (ioAction != Read)
                 {
 #ifndef NO_TEMPORARY
                     debug("AsynDriverInterface::readHandler(%s): "
@@ -940,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)
     {
@@ -1277,6 +1289,8 @@ void handleRequest(asynUser* pasynUser)
 {
     AsynDriverInterface* interface =
         static_cast<AsynDriverInterface*>(pasynUser->userPvt);
+    debug("AsynDriverInterface::handleRequest(%s) %s\n",
+        interface->clientName(), ioActionStr[interface->ioAction]);
     switch (interface->ioAction)
     {
         case None:
@@ -1311,6 +1325,8 @@ void handleTimeout(asynUser* pasynUser)
 {
     AsynDriverInterface* interface =
         static_cast<AsynDriverInterface*>(pasynUser->userPvt);
+    debug("AsynDriverInterface::handleTimeout(%s)\n",
+        interface->clientName());
     switch (interface->ioAction)
     {
         case Lock:

src/BinaryConverter.cc

@@ -72,25 +72,53 @@ printLong(const StreamFormat& format, StreamBuffer& output, long value)
     if (format.width > width) width = format.width;
     char zero = format.info[0];
     char one = format.info[1];
-    if (!(format.flags & left_flag))
+    char fill = (format.flags & zero_flag) ? zero : ' ';
+    if (format.flags & alt_flag)
     {
-        // pad left
-        char fill = (format.flags & zero_flag) ? zero : ' ';
-        while (width > prec)
+        // little endian (least significan bit first)
+        if (!(format.flags & left_flag))
         {
-            output.append(fill);
+            // pad left
+            while (width > prec)
+            {
+                output.append(' ');
+                width--;
+            }
+        }
+        while (prec--)
+        {
+            output.append((value & 1) ? one : zero);
+            value >>= 1;
             width--;
         }
+        while (width--)
+        {
+            // pad right
+            output.append(fill);
+        }
     }
-    while (prec--)
+    else
     {
-        output.append((value & (1 << prec)) ? one : zero);
-        width--;
-    }
-    while (width--)
-    {
-        // pad right
-        output.append(' ');
+        // big endian (most significan bit first)
+        if (!(format.flags & left_flag))
+        {
+            // pad left
+            while (width > prec)
+            {
+                output.append(fill);
+                width--;
+            }
+        }
+        while (prec--)
+        {
+            output.append((value & (1 << prec)) ? one : zero);
+            width--;
+        }
+        while (width--)
+        {
+            // pad right
+            output.append(' ');
+        }
     }
     return true;
 }
@@ -102,14 +130,28 @@ scanLong(const StreamFormat& format, const char* input, long& value)
     int width = format.width;
     if (width == 0) width = -1;
     int length = 0;
-    while (isspace(input[++length])); // skip whitespaces
+    while (isspace(input[length])) length++; // skip whitespaces
     char zero = format.info[0];
     char one = format.info[1];
     if (input[length] != zero && input[length] != one) return -1;
-    while (width-- && (input[length] == zero || input[length] == one))
+    if (format.flags & alt_flag)
     {
-        val <<= 1;
-        if (input[length++] == one) val |= 1;
+        // little endian (least significan bit first)
+        long mask = 1;
+        while (width-- && (input[length] == zero || input[length] == one))
+        {
+            if (input[length++] == one) val |= mask;
+            mask <<= 1;
+        }
+    }
+    else
+    {
+        // big endian (most significan bit first)
+        while (width-- && (input[length] == zero || input[length] == one))
+        {
+            val <<= 1;
+            if (input[length++] == one) val |= 1;
+        }
     }
     value = val;
     return length;

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/ExponentialConverter.cc

@@ -25,23 +25,29 @@
 #include "StreamError.h"
 #include <math.h>
 
-// Exponential Converter %m: format +00351-02 means +351e-2
+// Exponential Converter %m
+// Eric Berryman requested a double format that reads
+// +00011-01 as 11e-01
+// I.e integer mantissa and exponent without 'e' or '.'
+// But why not +11000-04 ?
+// For writing, I chose the following convention:
+// Format precision defines number of digits in mantissa
+// Number of digits in exponent is at least 2
+// No leading '0' in mantissa (except for 0.0 of course)
+// Format flags +, -, and space are supported in the usual way
+// Flags #, 0 are not supported
 
 class ExponentialConverter : public StreamFormatConverter
 {
     virtual int parse(const StreamFormat&, StreamBuffer&, const char*&, bool);
     virtual int scanDouble(const StreamFormat&, const char*, double&);
+    virtual bool printDouble(const StreamFormat&, StreamBuffer&, double);
 };
 
 int ExponentialConverter::
 parse(const StreamFormat& fmt, StreamBuffer& info,
     const char*& source, bool scanFormat)
 {
-    if (!scanFormat)
-    {
-        error("At the moment for %%m format only input is implemented\n");
-        return false;
-    }
     return double_format;
 }
 
@@ -59,5 +65,35 @@ scanDouble(const StreamFormat& fmt, const char* input, double& value)
     return length;
 }
 
+bool ExponentialConverter::
+printDouble(const StreamFormat& fmt, StreamBuffer& output, double value)
+{
+    // Have to divide value into mantissa and exponent
+    // precision field is number of characters in mantissa
+    // number of characters in exponent is at least 2
+    int spaces;
+    StreamBuffer buf;
+    
+    buf.printf("%.*e", fmt.prec-1, fabs(value));
+    buf.remove(1,1);
+    buf.remove(buf.find('e'),1);
+    
+    spaces = fmt.width-buf.length();
+    if (fmt.flags & (space_flag|sign_flag) || value < 0) spaces--;
+    if (spaces < 0) spaces = 0;
+    if (!(fmt.flags & left_flag))
+        output.append(' ', spaces);
+    if (fmt.flags & (space_flag|sign_flag) == space_flag && value >= 0)
+        output.append(' ');
+    if (fmt.flags & sign_flag && value >= 0)
+        output.append('+');
+    if (value <= 0)
+        output.append('-');
+    output.append(buf);
+    if (fmt.flags & left_flag)
+        output.append(' ', spaces);
+    return true;
+}
+
 RegisterConverter (ExponentialConverter, "m");
 

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/RawConverter.cc

@@ -45,7 +45,7 @@ printLong(const StreamFormat& format, StreamBuffer& output, long value)
     int width = prec;         // number of bytes in output
     if (format.width > width) width = format.width;
     char byte = 0;
-    if (format.flags & alt_flag) // lsb first (little endian)
+    if (format.flags & alt_flag) // little endian (lsb first)
     {
         while (prec--)
         {
@@ -69,7 +69,7 @@ printLong(const StreamFormat& format, StreamBuffer& output, long value)
             output.append(byte);
         }
     }
-    else // msb first (big endian)
+    else // big endian (msb first)
     {
         if (format.flags & zero_flag)
         {
@@ -107,7 +107,7 @@ scanLong(const StreamFormat& format, const char* input, long& value)
     }
     if (format.flags & alt_flag)
     {
-        // little endian (sign extended)*/
+        // little endian (lsb first)
         unsigned int shift = 0;
         while (--width && shift < sizeof(long)*8)
         {
@@ -131,7 +131,7 @@ scanLong(const StreamFormat& format, const char* input, long& value)
     }
     else
     {
-        // big endian */
+        // big endian  (msb first)
         if (format.flags & zero_flag)
         {
             // fill with zero

src/RawFloatConverter.cc

@@ -0,0 +1,151 @@
+/***************************************************************
+* StreamDevice Support                                         *
+*                                                              *
+* (C) 1999 Dirk Zimoch (zimoch@delta.uni-dortmund.de)          *
+* (C) 2005 Dirk Zimoch (dirk.zimoch@psi.ch)                    *
+*                                                              *
+* This is the raw format converter of StreamDevice.            *
+* Please refer to the HTML files in ../doc/ for a detailed     *
+* documentation.                                               *
+*                                                              *
+* If you do any changes in this file, you are not allowed to   *
+* redistribute it any more. If there is a bug or a missing     *
+* feature, send me an email and/or your patch. If I accept     *
+* your changes, they will go to the next release.              *
+*                                                              *
+* DISCLAIMER: If this software breaks something or harms       *
+* someone, it's your problem.                                  *
+*                                                              *
+***************************************************************/
+
+#include "StreamFormatConverter.h"
+#include "StreamError.h"
+
+#ifdef vxWorks
+#include "vxWorks.h"
+#define __BYTE_ORDER _BYTE_ORDER 
+#define __LITTLE_ENDIAN _LITTLE_ENDIAN
+#define __BIG_ENDIAN _BIG_ENDIAN 
+#else
+// Let's hope all other architectures have endian.h
+#include "endian.h"
+#endif
+
+#ifndef __BYTE_ORDER
+#error define __BYTE_ORDER as __LITTLE_ENDIAN or __BIG_ENDIAN
+#endif
+
+#if (__BYTE_ORDER == __LITTLE_ENDIAN || __BYTE_ORDER == __BIG_ENDIAN)
+
+// Raw Float Converter %R
+
+class RawFloatConverter : public StreamFormatConverter
+{
+    int parse(const StreamFormat&, StreamBuffer&, const char*&, bool);
+    bool printDouble(const StreamFormat&, StreamBuffer&, double);
+    int scanDouble(const StreamFormat&, const char*, double&);
+};
+
+int RawFloatConverter::
+parse(const StreamFormat& format, StreamBuffer&,
+    const char*&, bool)
+{
+    // Assume IEEE formats with 4 or 8 bytes (default: 4)
+    if (format.width==0 || format.width==4 || format.width==8)
+        return double_format;
+    error ("Only width 4 or 8 allowed for %%R format.\n");
+    return false;
+}
+
+bool RawFloatConverter::
+printDouble(const StreamFormat& format, StreamBuffer& output, double value)
+{
+    int nbOfBytes;
+    int n;
+    union {
+        double dval;
+        float  fval;
+        char   bytes[8];
+    } buffer;
+    
+    nbOfBytes = format.width;
+    if (nbOfBytes == 0)
+        nbOfBytes = 4;
+
+    if (nbOfBytes == 4)
+        buffer.fval = value;
+    else 
+        buffer.dval = value;
+
+#if (__BYTE_ORDER == __BIG_ENDIAN)
+    bool swap = format.flags & alt_flag;
+#else
+    bool swap = !(format.flags & alt_flag);
+#endif
+    
+    if (swap)
+    {
+    	for (n = nbOfBytes-1; n >= 0; n--)
+    	{
+            output.append(buffer.bytes[n]);
+    	}
+    } else {
+    	for (n = 0; n < nbOfBytes; n++)
+    	{
+            output.append(buffer.bytes[n]);
+    	}
+    } 
+    return true;
+}
+
+int RawFloatConverter::
+scanDouble(const StreamFormat& format, const char* input, double& value)
+{
+    int nbOfBytes;
+    int i, n;
+
+    union {
+        double dval;
+        float  fval;
+        char   bytes[8];
+    } buffer;
+    
+    nbOfBytes = format.width;
+    if (nbOfBytes == 0)
+        nbOfBytes = 4;
+
+    if (format.flags & skip_flag)
+    {
+        return(nbOfBytes); // just skip input
+    }
+    
+#if (__BYTE_ORDER == __BIG_ENDIAN)
+    bool swap = format.flags & alt_flag;
+#else
+    bool swap = !(format.flags & alt_flag);
+#endif
+
+    if (swap)
+    {
+    	for (n = nbOfBytes-1, i = 0; n >= 0; n--, i++)
+    	{
+            buffer.bytes[n] = input[i];
+    	}
+    } else {
+    	for (n = 0; n < nbOfBytes; n++)
+    	{
+            buffer.bytes[n] = input[n];
+    	}
+    }
+
+    if (nbOfBytes == 4)
+        value = buffer.fval;
+    else 
+        value = buffer.dval;
+
+    return nbOfBytes;
+}
+
+RegisterConverter (RawFloatConverter, "R");
+
+#endif /* known byte order */

src/StreamBuffer.h

@@ -29,11 +29,11 @@
 
 class StreamBuffer
 {
+    char local[64];
     long len;
     long cap;
     long offs;
     char* buffer;
-    char local[64];
 
     void grow(long);
     void init(const void*, long);
@@ -103,11 +103,11 @@ public:
     // reserve: reserve size bytes of memory and return
     // pointer to that memory (for copying something to it)
     char* reserve(long size)
-        {check(size); char* p=buffer+offs+len; len+=size; return p; }
+        {grow(size); char* p=buffer+len; len+=size; return p;}
 
     // append: append data at the end of the buffer
-    StreamBuffer& append(char c)
-        {check(1); buffer[offs+len++]=c; return *this;}
+    StreamBuffer& append(char c, long count=1)
+        {check(count); while(count-->0) buffer[offs+len++]=c; return *this;}
 
     StreamBuffer& append(const void* s, long size);
 

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

@@ -23,12 +23,6 @@
 #include <ctype.h>
 #include <stdlib.h>
 
-#include <epicsExport.h>
-int showAsyncErrors = 0;
-extern "C" {
-epicsExportAddress(int, showAsyncErrors);
-}
-
 enum Commands { end_cmd, in_cmd, out_cmd, wait_cmd, event_cmd, exec_cmd,
     connect_cmd, disconnect_cmd };
 const char* commandStr[] = { "end", "in", "out", "wait", "event", "exec",
@@ -622,6 +616,7 @@ formatOutput()
     char command;
     const char* fieldName = NULL;
     const char* formatstring;
+    int formatstringlen;
     while ((command = *commandIndex++) != StreamProtocolParser::eos)
     {
         switch (command)
@@ -631,7 +626,7 @@ formatOutput()
                 debug("StreamCore::formatOutput(%s): StreamProtocolParser::format_field\n",
                     name());
                 // code layout:
-                // field <StreamProtocolParser::eos> addrlen AddressStructure formatstring <StreamProtocolParser::eos> StreamFormat [info]
+                // field <eos> addrlen AddressStructure formatstring <eos> StreamFormat [info]
                 fieldName = commandIndex;
                 commandIndex += strlen(commandIndex)+1;
                 unsigned short addrlen = extract<unsigned short>(commandIndex);
@@ -641,14 +636,23 @@ formatOutput()
             case StreamProtocolParser::format:
             {
                 // code layout:
-                // formatstring <StreamProtocolParser::eos> StreamFormat [info]
+                // formatstring <eos> StreamFormat [info]
                 formatstring = commandIndex;
-                while (*commandIndex++); // jump after <StreamProtocolParser::eos>
+                // jump after <eos>
+                while (*commandIndex)
+                {
+                    if (*commandIndex == esc) commandIndex++;
+                    commandIndex++;
+                }
+                formatstringlen = commandIndex-formatstring;
+                commandIndex++;
                 StreamFormat fmt = extract<StreamFormat>(commandIndex);
                 fmt.info = commandIndex; // point to info string
                 commandIndex += fmt.infolen;
+#ifndef NO_TEMPORARY
                 debug("StreamCore::formatOutput(%s): format = %%%s\n",
-                    name(), formatstring);
+                    name(), StreamBuffer(formatstring, formatstringlen).expand()());
+#endif
                 if (fmt.type == pseudo_format)
                 {
                     if (!StreamFormatConverter::find(fmt.conv)->
@@ -663,12 +667,13 @@ formatOutput()
                 flags &= ~Separator;
                 if (!formatValue(fmt, fieldAddress ? fieldAddress() : NULL))
                 {
+                    StreamBuffer formatstr(formatstring, formatstringlen);
                     if (fieldName)
                         error("%s: Cannot format field '%s' with '%%%s'\n",
-                            name(), fieldName, formatstring);
+                            name(), fieldName, formatstr.expand()());
                     else
                         error("%s: Cannot format value with '%%%s'\n",
-                            name(), formatstring);
+                            name(), formatstr.expand()());
                     return false;
                 }
                 fieldAddress.clear();
@@ -807,8 +812,6 @@ writeCallback(StreamIoStatus status)
     flags &= ~WritePending;
     if (status != StreamIoSuccess)
     {
-        error("%s: write failed: %s\n",
-            name(), StreamIoStatusStr[status]);
         finishProtocol(WriteTimeout);
         return;
     }
@@ -894,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;
@@ -997,7 +1006,7 @@ readCallback(StreamIoStatus status,
         }
         // try to parse what we got
         end = inputBuffer.length();
-        if (!(flags & AsyncMode)||showAsyncErrors)
+        if (!(flags & AsyncMode))
         {
             error("%s: Timeout after reading %ld byte%s \"%s%s\"\n",
                 name(), end, end==1 ? "" : "s", end > 20 ? "..." : "",
@@ -1127,7 +1136,7 @@ matchInput()
                     }
                     if (consumed < 0)
                     {
-                        if ((!(flags & AsyncMode)||showAsyncErrors) && onMismatch[0] != in_cmd)
+                        if (!(flags & AsyncMode) && onMismatch[0] != in_cmd)
                         {
                             error("%s: Input \"%s%s\" does not match format %%%s\n",
                                 name(), inputLine.expand(consumedInput, 20)(),
@@ -1142,7 +1151,7 @@ matchInput()
                 flags &= ~Separator;
                 if (!matchValue(fmt, fieldAddress ? fieldAddress() : NULL))
                 {
-                    if ((!(flags & AsyncMode)||showAsyncErrors) && onMismatch[0] != in_cmd)
+                    if (!(flags & AsyncMode) && onMismatch[0] != in_cmd)
                     {
                         if (flags & ScanTried)
                             error("%s: Input \"%s%s\" does not match format %%%s\n",
@@ -1172,7 +1181,7 @@ matchInput()
                 {
                     int i = 0;
                     while (commandIndex[i] >= ' ') i++;
-                    if ((!(flags & AsyncMode)||showAsyncErrors) && onMismatch[0] != in_cmd)
+                    if (!(flags & AsyncMode) && onMismatch[0] != in_cmd)
                     {
                         error("%s: Input \"%s%s\" too short."
                               " No match for \"%s\"\n",
@@ -1185,7 +1194,7 @@ matchInput()
                 }
                 if (command != inputLine[consumedInput])
                 {
-                    if ((!(flags & AsyncMode)||showAsyncErrors) && onMismatch[0] != in_cmd)
+                    if (!(flags & AsyncMode) && onMismatch[0] != in_cmd)
                     {
                         int i = 0;
                         while (commandIndex[i] >= ' ') i++;
@@ -1210,7 +1219,7 @@ matchInput()
     long surplus = inputLine.length()-consumedInput;
     if (surplus > 0 && !(flags & IgnoreExtraInput))
     {
-        if ((!(flags & AsyncMode)||showAsyncErrors) && onMismatch[0] != in_cmd)
+        if (!(flags & AsyncMode) && onMismatch[0] != in_cmd)
         {
             error("%s: %ld byte%s surplus input \"%s%s\"\n",
                 name(), surplus, surplus==1 ? "" : "s",

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
@@ -1157,11 +1195,11 @@ noMoreElements:
 }
 
 #ifdef EPICS_3_14
-
+// Pass command to iocsh
 void streamExecuteCommand(CALLBACK *pcallback)
 {
     Stream* pstream = static_cast<Stream*>(pcallback->user);
-    
+
     if (iocshCmd(pstream->outputLine()) != OK)
     {
         pstream->execCallback(StreamIoFault);
@@ -1172,12 +1210,13 @@ void streamExecuteCommand(CALLBACK *pcallback)
     }
 }
 #else
-extern "C" int execute (const char *cmd);
+// Pass command to vxWorks shell
+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
     {

src/StreamProtocol.cc

@@ -1255,7 +1255,7 @@ compileString(StreamBuffer& buffer, const char*& source,
                         // so that extra information is ready for format converter
                         if (numFormats+1 == sizeof(formatPos))
                         {
-                            errorMsg(line, "Too many formats in line");
+                            errorMsg(line, "Max 20 formats allowed in one protocol line");
                             return false;
                         }
                         formatPos[numFormats++]=buffer.length();