Fix deadlock for pending write() calls on transport close. (#139)

Add tests for read() and write() deadlocks.

chronos/transports/common.nim

@@ -495,6 +495,9 @@ template getServerUseClosedError*(): ref TransportUseClosedError =
 template getTransportTooManyError*(): ref TransportTooManyError =
   newException(TransportTooManyError, "Too many open transports!")
 
+template getTransportUseClosedError*(): ref TransportUseClosedError =
+  newException(TransportUseClosedError, "Transport is already closed!")
+
 template getTransportOsError*(err: OSErrorCode): ref TransportOsError =
   var msg = "(" & $int(err) & ") " & osErrorMsg(err)
   var tre = newException(TransportOsError, msg)

chronos/transports/stream.nim

@@ -322,6 +322,12 @@ when defined(windows):
     var ovl = cast[PtrCustomOverlapped](udata)
     var transp = cast[StreamTransport](ovl.data.udata)
 
+    if WriteClosed in transp.state:
+      transp.state.excl(WritePending)
+      transp.state.incl({WritePaused})
+      let error = getTransportUseClosedError()
+      failPendingWriteQueue(transp.queue, error)
+    else:
       while len(transp.queue) > 0:
         if WritePending in transp.state:
           ## Continuation
@@ -1177,6 +1183,11 @@ else:
       ## after transport was closed.
       return
 
+    if WriteClosed in transp.state:
+      transp.state.incl({WritePaused})
+      let error = getTransportUseClosedError()
+      failPendingWriteQueue(transp.queue, error)
+    else:
       if len(transp.queue) > 0:
         var vector = transp.queue.popFirst()
         while true:

tests/teststream.nim

@@ -1152,6 +1152,54 @@ suite "Stream Transport test suite":
         await server.closeWait()
         setMaxOpenFiles(maxFiles)
 
+  proc testWriteOnClose(address: TransportAddress): Future[bool] {.async.} =
+    var server = createStreamServer(address, flags = {ReuseAddr, NoPipeFlash})
+    var res = 0
+
+    proc acceptTask(server: StreamServer) {.async.} =
+      let transp = await server.accept()
+      var futs = newSeq[Future[int]](TestsCount)
+      var msg = createBigMessage(1024)
+      for i in 0 ..< len(futs):
+        futs[i] = transp.write(msg)
+
+      await transp.closeWait()
+      await sleepAsync(100.milliseconds)
+
+      for i in 0 ..< len(futs):
+        if futs[i].failed() and (futs[i].error of TransportUseClosedError):
+          inc(res)
+
+      await server.closeWait()
+
+    var acceptFut = acceptTask(server)
+    var transp = await connect(address)
+    await server.join()
+    await transp.closeWait()
+    await acceptFut
+    return (res == TestsCount)
+
+  proc testReadOnClose(address: TransportAddress): Future[bool] {.async.} =
+    var server = createStreamServer(address, flags = {ReuseAddr, NoPipeFlash})
+    var res = false
+
+    proc acceptTask(server: StreamServer) {.async.} =
+      let transp = await server.accept()
+      var buffer = newSeq[byte](1024)
+      var fut = transp.readOnce(addr buffer[0], len(buffer))
+      await transp.closeWait()
+      await sleepAsync(100.milliseconds)
+      if fut.failed() and (fut.error of TransportUseClosedError):
+        res = true
+      await server.closeWait()
+
+    var acceptFut = acceptTask(server)
+    var transp = await connect(address)
+    await server.join()
+    await transp.closeWait()
+    await acceptFut
+    return res
+
   markFD = getCurrentFD()
 
   for i in 0..<len(addresses):
@@ -1226,6 +1274,10 @@ suite "Stream Transport test suite":
         skip()
       else:
         check waitFor(testAcceptTooMany(addresses[i])) == true
+    test prefixes[i] & "write() queue notification on close() test":
+      check waitFor(testWriteOnClose(addresses[i])) == true
+    test prefixes[i] & "read() notification on close() test":
+      check waitFor(testReadOnClose(addresses[i])) == true
   test "Servers leak test":
     check getTracker("stream.server").isLeaked() == false
   test "Transports leak test":