Simplify stream transport states (#630)

* `ReadClosed`/`WriteClosed` -> `Closed` (they're always set together)
* `atEof` doesn't need to check transport state
* remove redundant while looping in `readStreamLoop` (it's done by
`handleEintr`)
* move error handling to `resumeRead` / `resumeWrite` and make usre it
matches loop version
* clear reader future on read cancellation (no need to hold on to
memory)
* document closeFd

Author: arnetheduck

chronos/internal/asyncengine.nim

@@ -163,12 +163,6 @@ proc raiseOsDefect*(error: OSErrorCode, msg = "") {.noreturn, noinline.} =
   raise (ref Defect)(msg: msg & "\n[" & $int(error) & "] " & osErrorMsg(error) &
                           "\n" & getStackTrace())
 
-func toPointer(error: OSErrorCode): pointer =
-  when sizeof(int) == 8:
-    cast[pointer](uint64(uint32(error)))
-  else:
-    cast[pointer](uint32(error))
-
 func toException*(v: OSErrorCode): ref OSError = newOSError(v)
   # This helper will allow to use `tryGet()` and raise OSError for
   # Result[T, OSErrorCode] values.
@@ -672,30 +666,25 @@ elif defined(windows):
     ## Closes a socket and ensures that it is unregistered.
     let loop = getThreadDispatcher()
     loop.handles.excl(fd)
-    let
-      param = toPointer(
-        if closeFd(SocketHandle(fd)) == 0:
-          OSErrorCode(0)
-        else:
-          osLastError()
-      )
+    # Because we don't set SO_LINGER, `closeFd` should always succeed.
+    # Future API might be added to recover linger errors, but this is currently
+    # not supported so we discard the return value.
+    discard closeFd(SocketHandle(fd))
     if not(isNil(aftercb)):
-      loop.callbacks.addLast(AsyncCallback(function: aftercb, udata: param))
+      loop.callbacks.addLast(AsyncCallback(function: aftercb))
 
   proc closeHandle*(fd: AsyncFD, aftercb: CallbackFunc = nil) =
     ## Closes a (pipe/file) handle and ensures that it is unregistered.
     let loop = getThreadDispatcher()
     loop.handles.excl(fd)
-    let
-      param = toPointer(
-        if closeFd(HANDLE(fd)) == 0:
-          OSErrorCode(0)
-        else:
-          osLastError()
-      )
+    # Closing the handle should always succeed since async failures are reported
+    # via IOCP (pending operations cancelled on closed) and the documentation
+    # for CloseHandle does not specify and other specific conditions where it
+    # might fail.
+    discard closeFd(HANDLE(fd))
 
     if not(isNil(aftercb)):
-      loop.callbacks.addLast(AsyncCallback(function: aftercb, udata: param))
+      loop.callbacks.addLast(AsyncCallback(function: aftercb))
 
   proc unregisterAndCloseFd*(fd: AsyncFD): Result[void, OSErrorCode] =
     ## Unregister from system queue and close asynchronous socket.
@@ -890,22 +879,20 @@ elif defined(macosx) or defined(freebsd) or defined(netbsd) or
     let loop = getThreadDispatcher()
 
     proc continuation(udata: pointer) =
-      let
-        param = toPointer(
-          if SocketHandle(fd) in loop.selector:
-            let ures = unregister2(fd)
-            if ures.isErr():
-              discard closeFd(cint(fd))
-              ures.error()
-            else:
-              if closeFd(cint(fd)) != 0:
-                osLastError()
-              else:
-                OSErrorCode(0)
-          else:
-            osdefs.EBADF
-        )
-      if not(isNil(aftercb)): aftercb(param)
+      if SocketHandle(fd) in loop.selector:
+        discard unregister2(fd)
+      # `closeFd` might fail if an I/O error occurs during an async I/O
+      # operation, but on *most* posix systems this still results in the file
+      # descriptor being closed regardless of the error.
+      #
+      # For sockets in parituclar, we don't set SO_LINGER meaning that close
+      # happens immediately and does not wait for the socket to go through
+      # its graceful shutdown sequence.
+      #
+      # We currently don't have an API for returning this error to the caller
+      # so we discard it here - future work might expose it.
+      discard closeFd(cint(fd))
+      if not(isNil(aftercb)): aftercb(nil)
 
     withData(loop.selector, cint(fd), adata) do:
       # We are scheduling reader and writer callbacks to be called

chronos/transports/common.nim

@@ -121,19 +121,17 @@ type
   TransportAbortedError* = object of TransportError
     ## Remote client disconnected before server accepts connection
 
-  TransportState* = enum
+  TransportState* {.pure.} = enum
     ## Transport's state
     ReadPending,                  # Read operation pending (Windows)
     ReadPaused,                   # Read operations paused
-    ReadClosed,                   # Read operations closed
     ReadEof,                      # Read at EOF
     ReadError,                    # Read error
     WritePending,                 # Writer operation pending (Windows)
     WritePaused,                  # Writer operations paused
-    WriteClosed,                  # Writer operations closed
     WriteEof,                     # Remote peer disconnected
     WriteError                    # Write error
-
+    Closed                        # Transport was closed
 var
   AnyAddress* = TransportAddress(family: AddressFamily.IPv4, port: Port(0))
     ## Default INADDR_ANY address for IPv4
@@ -568,11 +566,11 @@ proc anyAddressFix*(a: TransportAddress): TransportAddress =
     a
 
 template checkClosed*(t: untyped) =
-  if (ReadClosed in (t).state) or (WriteClosed in (t).state):
+  if TransportState.Closed in (t).state:
     raise newException(TransportUseClosedError, "Transport is already closed!")
 
 template checkClosed*(t: untyped, future: untyped) =
-  if (ReadClosed in (t).state) or (WriteClosed in (t).state):
+  if TransportState.Closed in (t).state:
     future.fail(newException(TransportUseClosedError,
                              "Transport is already closed!"))
     return future

chronos/transports/datagram.nim

@@ -229,10 +229,10 @@ when defined(windows):
         of ERROR_OPERATION_ABORTED:
           # CancelIO() interrupt or closeSocket() call.
           transp.state.incl(ReadPaused)
-          if ReadClosed in transp.state and not(transp.future.finished()):
+          if TransportState.Closed in transp.state and not(transp.future.finished()):
             # Stop tracking transport
             untrackCounter(DgramTransportTrackerName)
-            # If `ReadClosed` present, then close(transport) was called.
+            # If `Closed` present, then close(transport) was called.
             transp.future.complete()
             GC_unref(transp)
           break
@@ -243,7 +243,7 @@ when defined(windows):
           asyncSpawn transp.function(transp, remoteAddress)
       else:
         ## Initiation
-        if transp.state * {ReadEof, ReadClosed, ReadError} == {}:
+        if transp.state * {ReadEof, ReadError, TransportState.Closed} == {}:
           transp.state.incl(ReadPending)
           let fd = SocketHandle(transp.fd)
           transp.rflag = 0
@@ -275,7 +275,7 @@ when defined(windows):
         else:
           # Transport closure happens in callback, and we not started new
           # WSARecvFrom session.
-          if ReadClosed in transp.state and not(transp.future.finished()):
+          if TransportState.Closed in transp.state and not(transp.future.finished()):
             # Stop tracking transport
             untrackCounter(DgramTransportTrackerName)
             transp.future.complete()
@@ -446,7 +446,7 @@ else:
       ## This situation can be happen, when there events present
       ## after transport was closed.
       return
-    if ReadClosed in transp.state:
+    if TransportState.Closed in transp.state:
       transp.state.incl({ReadPaused})
     else:
       while true:
@@ -479,7 +479,7 @@ else:
       ## This situation can be happen, when there events present
       ## after transport was closed.
       return
-    if WriteClosed in transp.state:
+    if TransportState.Closed in transp.state:
       transp.state.incl({WritePaused})
     else:
       if len(transp.queue) > 0:
@@ -655,9 +655,10 @@ proc close*(transp: DatagramTransport) =
       transp.future.complete()
       GC_unref(transp)
 
-  when defined(windows):
-    if {ReadClosed, WriteClosed} * transp.state == {}:
-      transp.state.incl({WriteClosed, ReadClosed})
+  if TransportState.Closed notin transp.state:
+    transp.state.incl(TransportState.Closed)
+
+    when defined(windows):
       if ReadPaused in transp.state:
         # If readDatagramLoop() is not running we need to finish in
         # continuation step.
@@ -666,10 +667,8 @@ proc close*(transp: DatagramTransport) =
         # If readDatagramLoop() is running, it will be properly finished inside
         # of readDatagramLoop().
         closeSocket(transp.fd)
-  else:
-    if {ReadClosed, WriteClosed} * transp.state == {}:
-      transp.state.incl({WriteClosed, ReadClosed})
-      closeSocket(transp.fd, continuation)
+    else:
+        closeSocket(transp.fd, continuation)
 
 proc getTransportAddresses(
     local, remote: Opt[IpAddress],
@@ -1001,7 +1000,7 @@ proc join*(transp: DatagramTransport): Future[void] {.
 
 proc closed*(transp: DatagramTransport): bool {.inline.} =
   ## Returns ``true`` if transport in closed state.
-  {ReadClosed, WriteClosed} * transp.state != {}
+  TransportState.Closed in transp.state
 
 proc closeWait*(transp: DatagramTransport): Future[void] {.
     async: (raises: []).} =