missing header

makefile

@@ -35,6 +35,7 @@ SOURCES += $(BUSSES:%=src/%Interface.cc)
 SOURCES += $(wildcard src/Stream*.cc)
 SOURCES += src/StreamVersion.c
 
+HEADERS += devStream.h
 HEADERS += StreamFormat.h
 HEADERS += StreamFormatConverter.h
 HEADERS += StreamBuffer.h

src/AsynDriverInterface.cc

@@ -22,6 +22,11 @@
 #include "StreamError.h"
 #include "StreamBuffer.h"
 
+#include <epicsVersion.h>
+#if (EPICS_VERSION == 3 && EPICS_REVISION == 14)
+#define EPICS_3_14
+#endif
+
 #ifdef EPICS_3_14
 #include <epicsAssert.h>
 #include <epicsTime.h>
@@ -126,7 +131,7 @@ static const char* ioActionStr[] = {
 };
 
 static const char* asynStatusStr[] = {
-    "asynSuccess", "asynTimeout", "asynOverflow", "asynError"
+    "asynSuccess", "asynTimeout", "asynOverflow", "asynError", "asynDisconnected", "asynDisabled"
 };
 
 static const char* eomReasonStr[] = {
@@ -212,7 +217,9 @@ class AsynDriverInterface : StreamBusInterface
     }
     void startTimer(double timeout) {
 #ifdef EPICS_3_14
-        timer->start(*this, timeout);
+        timer->start(*this, timeout
+            +epicsThreadSleepQuantum()*0.5
+        );
 #else
         callbackSetPriority(priority(), &timeoutCallback);
         wdStart(timer, (int)((timeout+1)*sysClkRateGet())-1,
@@ -516,7 +523,7 @@ connectToAsynPort()
         status = pasynOctet->read(pvtOctet, pasynUser,
             buffer, 0, &received, &eomReason);
         debug("AsynDriverInterface::connectToAsynPort(%s): "
-            "read(..., 0, ...) [timeout=%f seconds] = %s\n",
+            "read(..., 0, ...) [timeout=%g sec] = %s\n",
             clientName(), pasynUser->timeout,
             asynStatusStr[status]);
         pasynManager->isConnected(pasynUser, &connected);
@@ -648,7 +655,7 @@ writeHandler()
         outputBuffer, outputSize, &written);
     debug("AsynDriverInterface::writeHandler(%s): "
         "write(..., outputSize=%ld, written=%ld) "
-        "[timeout=%f seconds] = %s\n",
+        "[timeout=%g sec] = %s\n",
         clientName(), (long)outputSize,  (long)written,
         pasynUser->timeout, asynStatusStr[status]);
 
@@ -754,12 +761,22 @@ readRequest(unsigned long replyTimeout_ms, unsigned long readTimeout_ms,
         ioAction = Read;
         queueTimeout = replyTimeout;
     }
-    debug("AsynDriverInterface::readRequest %s:  queueRequest(..., priority=%d, queueTimeout=%f)\n",
-        clientName(), priority(), queueTimeout);
     status = pasynManager->queueRequest(pasynUser,
         priority(), queueTimeout);
-    if (status != asynSuccess && !async)
+    debug("AsynDriverInterface::readRequest %s: "
+        "queueRequest(..., priority=%d, queueTimeout=%g sec) = %s [async=%s] %s\n",
+        clientName(), priority(), queueTimeout,
+        asynStatusStr[status], async? "true" : "false",
+        status!=asynSuccess ? pasynUser->errorMessage : "");
+    if (status != asynSuccess)
     {
+        // Not queued for some reason (e.g. disconnected / already queued)
+        if (async)
+        {
+            // silently try again later
+            startTimer(replyTimeout);
+            return true;
+        }
         error("%s readRequest: pasynManager->queueRequest() failed: %s\n",
             clientName(), pasynUser->errorMessage);
         return false;
@@ -867,7 +884,7 @@ readHandler()
         
         debug("AsynDriverInterface::readHandler(%s): ioAction=%s "
             "read(..., bytesToRead=%ld, ...) "
-            "[timeout=%f seconds]\n",
+            "[timeout=%g sec]\n",
             clientName(), ioActionStr[ioAction],
             bytesToRead, pasynUser->timeout);
         status = pasynOctet->read(pvtOctet, pasynUser,
@@ -977,7 +994,7 @@ readHandler()
                 // read timeout
 #ifndef NO_TEMPORARY
                 debug("AsynDriverInterface::readHandler(%s): "
-                        "ioAction=%s, timeout [%f seconds] "
+                        "ioAction=%s, timeout [%g sec] "
                         "after %ld of %ld bytes \"%s\"\n",
                     clientName(), ioActionStr[ioAction], pasynUser->timeout,
                     (long)received, bytesToRead,
@@ -1256,6 +1273,8 @@ timerExpired()
             // queueRequest might fail if another request was just queued
             pasynManager->isAutoConnect(pasynUser, &autoconnect);
             pasynManager->isConnected(pasynUser, &connected);
+            debug("%s: polling for I/O Intr: autoconnected: %d, connect: %d\n",
+                clientName(), autoconnect, connected);
             if (autoconnect && !connected)
             {
                 // has explicitely been disconnected
@@ -1266,8 +1285,14 @@ timerExpired()
             else
             {
                 // queue for read poll (no timeout)
-                pasynManager->queueRequest(pasynUser,
+                asynStatus status = pasynManager->queueRequest(pasynUser,
                     asynQueuePriorityLow, -1.0);
+                // if this fails, we are already queued by another thread
+                debug("AsynDriverInterface::timerExpired %s: "
+                    "queueRequest(..., priority=Low, queueTimeout=-1) = %s %s\n",
+                    clientName(), asynStatusStr[status],
+                    status!=asynSuccess ? pasynUser->errorMessage : "");
+                if (status != asynSuccess) startTimer(replyTimeout);
                 // continues with:
                 //    handleRequest() -> readHandler() -> readCallback()
             }
@@ -1392,6 +1417,7 @@ void handleRequest(asynUser* pasynUser)
 {
     AsynDriverInterface* interface =
         static_cast<AsynDriverInterface*>(pasynUser->userPvt);
+    interface->cancelTimer();
     debug("AsynDriverInterface::handleRequest(%s) %s\n",
         interface->clientName(), ioActionStr[interface->ioAction]);
     switch (interface->ioAction)

src/StreamEpics.cc

@@ -18,9 +18,14 @@
 *                                                              *
 ***************************************************************/
 
+#include "devStream.h"
 #include "StreamCore.h"
 #include "StreamError.h"
-#include "devStream.h"
+
+#include <epicsVersion.h>
+#if (EPICS_VERSION == 3 && EPICS_REVISION == 14)
+#define EPICS_3_14
+#endif
 
 #ifndef EPICS_3_14
 extern "C" {

src/devStream.h

@@ -23,7 +23,7 @@
 
 #define STREAM_MAJOR 2
 #define STREAM_MINOR 5
-#define STREAM_PATCHLEVEL 10
+#define STREAM_PATCHLEVEL 11
 
 #if defined(__vxworks) || defined(vxWorks)
 #include <vxWorks.h>
@@ -70,23 +70,20 @@ typedef const struct format_s {
 extern "C" {
 #endif
 
-#ifdef _WIN32
-__declspec(dllimport)
-#endif
-extern FILE* StreamDebugFile;
+epicsShareExtern FILE* StreamDebugFile;
 
 extern const char StreamVersion [];
 
 typedef long (*streamIoFunction) (dbCommon*, format_t*);
 
-long streamInit(int after);
-long streamInitRecord(dbCommon *record, struct link *ioLink,
+epicsShareExtern long streamInit(int after);
+epicsShareExtern long streamInitRecord(dbCommon *record, struct link *ioLink,
     streamIoFunction readData, streamIoFunction writeData);
-long streamReport(int interest);
-long streamReadWrite(dbCommon *record);
-long streamGetIointInfo(int cmd, dbCommon *record, IOSCANPVT *ppvt);
-long streamPrintf(dbCommon *record, format_t *format, ...);
-long streamScanfN(dbCommon *record, format_t *format,
+epicsShareExtern long streamReport(int interest);
+epicsShareExtern long streamReadWrite(dbCommon *record);
+epicsShareExtern long streamGetIointInfo(int cmd, dbCommon *record, IOSCANPVT *ppvt);
+epicsShareExtern long streamPrintf(dbCommon *record, format_t *format, ...);
+epicsShareExtern long streamScanfN(dbCommon *record, format_t *format,
     void*, size_t maxStringSize);
 
 /* backward compatibility stuff */

streamApp/terminal.tcl

@@ -18,7 +18,7 @@ proc createTerm {sock} {
 }
 
 proc connect {sock addr port} {
-    fconfigure $sock -blocking 0 -buffering none
+    fconfigure $sock -blocking 0 -buffering none -translation binary
     createTerm $sock
     fileevent $sock readable "receiveHandler $sock"
 }
@@ -136,14 +136,6 @@ proc sendAsync {wait message} {
     after $wait sendAsync $wait [list $message]
 }
 
-proc sendAsyncX {wait} {
-    if {$::counter < 0} return
-    foreach term [array names ::socket] {
-        sendReply $::socket($term) "\u00101\u0004~\u0005~\u00100\u0002|0062|2|1|0|1216|0|0.1087E+0 \u0003\u0012"
-    }
-    after $wait sendAsyncX $wait
-}
-
 if {[info proc tkTextInsert] != ""} {
     set insert tkTextInsert
     set paste tkTextPaste
@@ -196,7 +188,7 @@ for {set ascii 0x61} {$ascii <= 0x7a} {incr ascii} {
     bind Text <Control-[format %c $ascii]> ""
 }
 #remove bindings on symbolic tags
-foreach tag {Clear Paste Copy Cut } {
+foreach tag {Clear Paste Copy Cut} {
     bind Text <<$tag>> ""
 }
 

streamApp/test.proto

@@ -1,5 +1,5 @@
-terminator = CR LF;
-readtimeout = 1000;
+terminator = LF;
+readtimeout = 10;
 pollperiod = 10;
 replytimeout = 1000;
 command {