HPCC-31474 Refactor socket readtms/writtms code to handle nonblocking

Cleanup and refactor the jsocket and securesocket readtms/writetms
and derivatives to cope with nonblocking sockets.

Symantically they remain the same.
e.g. readtms will not return until it has written at least min_size.
Duplicate/very similar code in dedupped/consolidated into readtms
and writetms

1) in non-blocking mode, readtms would fire an exception if socket
not ready (EWOULDBLOCK, EAGAIN).
2) jsocket writetms would temporarily place a non-blocking socket
into nonblocking mode.
Now writetms takes a minSize (akin to readtms minSize). It will
return if at least this much is written (but normal usage is with
min=max).
NB: previous writetms in both implementations were
not called anywhere.
3) in non-blocking mode securesocket readtms and writetms would
previously fail if blocked

Signed-off-by: Jake Smith <[email protected]>

fs/dafsclient/rmtclient.cpp

@@ -733,7 +733,7 @@ void CRemoteBase::connectSocket(SocketEndpoint &ep, unsigned connectTimeoutMs, u
             {
                 unsigned remaining;
                 if (tm.timemon->timedout(&remaining))
-                    throwJSocketException(JSOCKERR_connection_failed);
+                    THROWJSOCKEXCEPTION(JSOCKERR_connection_failed);
                 socket.setown(ISocket::connect_timeout(ep,remaining));
             }
             else

roxie/udplib/udpsha.hpp

@@ -344,7 +344,7 @@ class CSocketSimulator : public CInterfaceOf<ISocket>
                            unsigned timeout) override { UNIMPLEMENTED; }
     virtual void   read(void* buf, size32_t size) override { UNIMPLEMENTED; }
     virtual size32_t write(void const* buf, size32_t size) override { UNIMPLEMENTED; }
-    virtual size32_t writetms(void const* buf, size32_t size, unsigned timeoutms=WAIT_FOREVER) override { UNIMPLEMENTED; }
+    virtual size32_t writetms(void const* buf, size32_t minSize, size32_t size, unsigned timeoutms=WAIT_FOREVER) override { UNIMPLEMENTED; }
 
     virtual size32_t get_max_send_size() override { UNIMPLEMENTED; }
     virtual ISocket* accept(bool allowcancel=false, SocketEndpoint *peerEp = nullptr) override { UNIMPLEMENTED; }

system/jlib/jdebug.hpp

@@ -166,6 +166,15 @@ class CCycleTimer
     {
         return static_cast<unsigned>(cycle_to_millisec(elapsedCycles()));
     }
+    inline unsigned remainingMs(unsigned timeoutMs) const
+    {
+        if (INFINITE == timeoutMs)
+            return INFINITE;
+        unsigned eMs = elapsedMs();
+        if (eMs >= timeoutMs)
+            return 0;
+        return timeoutMs - eMs;
+    }
 };
 inline cycle_t queryOneSecCycles() { return oneSecInCycles; }