attempt to make asynchttpserver better; fixes #15925; [backport:1.0] (#15957)

* attempt to make asynchttpserver better; fixes #15925; [backport:1.0]
* better documentation
* fixes 'nim doc'
* makes test green again
* ported to FreeRTOS
* fixes the doc rendering

changelog.md

@@ -35,6 +35,12 @@
 
 - `doAssertRaises` now correctly handles foreign exceptions.
 
+- Added `asyncdispatch.activeDescriptors` that returns the number of currently
+  active async event handles/file descriptors
+- Added `asyncdispatch.maxDescriptors` that returns the maximum number of
+  active async event handles/file descriptors.
+
+
 ## Language changes
 
 - `nimscript` now handles `except Exception as e`

lib/pure/asyncdispatch.nim

@@ -1934,3 +1934,27 @@ proc waitFor*[T](fut: Future[T]): T =
     poll()
 
   fut.read
+
+proc activeDescriptors*(): int {.inline.} =
+  ## Returns the current number of active file descriptors for the current
+  ## event loop. This is a cheap operation that does not involve a system call.
+  when defined(windows):
+    result = getGlobalDispatcher().handles.len
+  elif not defined(nimdoc):
+    result = getGlobalDispatcher().selector.count
+
+when defined(posix):
+  import posix
+
+when defined(linux) or defined(windows) or defined(macosx) or defined(bsd):
+  proc maxDescriptors*(): int {.raises: OSError.} =
+    ## Returns the maximum number of active file descriptors for the current
+    ## process. This involves a system call. For now `maxDescriptors` is
+    ## supported on the following OSes: Windows, Linux, OSX, BSD.
+    when defined(windows):
+      result = 16_700_000
+    else:
+      var fdLim: RLimit
+      if getrlimit(RLIMIT_NOFILE, fdLim) < 0:
+        raiseOSError(osLastError())
+      result = int(fdLim.rlim_cur) - 1

lib/pure/asynchttpserver.nim

@@ -21,14 +21,27 @@
 ## respond to all requests with a ``200 OK`` response code and "Hello World"
 ## as the response body.
 ##
-## .. code-block::nim
+## .. code-block:: Nim
+##
 ##   import asynchttpserver, asyncdispatch
 ##
+##   proc main {.async.} =
 ##     var server = newAsyncHttpServer()
 ##     proc cb(req: Request) {.async.} =
-##      await req.respond(Http200, "Hello World")
+##       let headers = {"Date": "Tue, 29 Apr 2014 23:40:08 GMT",
+##           "Content-type": "text/plain; charset=utf-8"}
+##       await req.respond(Http200, "Hello World", headers.newHttpHeaders())
 ##
-##    waitFor server.serve(Port(8080), cb)
+##     server.listen Port(5555)
+##     while true:
+##       if server.shouldAcceptRequest(5):
+##         var (address, client) = await server.socket.acceptAddr()
+##         asyncCheck processClient(server, client, address, cb)
+##       else:
+##         poll()
+##
+##   asyncCheck main()
+##   runForever()
 
 import asyncnet, asyncdispatch, parseutils, uri, strutils
 import httpcore
@@ -58,14 +71,12 @@ type
     reuseAddr: bool
     reusePort: bool
     maxBody: int ## The maximum content-length that will be read for the body.
+    maxFDs: int
 
 proc newAsyncHttpServer*(reuseAddr = true, reusePort = false,
                          maxBody = 8388608): AsyncHttpServer =
   ## Creates a new ``AsyncHttpServer`` instance.
-  new result
+  result = AsyncHttpServer(reuseAddr: reuseAddr, reusePort: reusePort, maxBody: maxBody)
-  result.reuseAddr = reuseAddr
-  result.reusePort = reusePort
-  result.maxBody = maxBody
 
 proc addHeaders(msg: var string, headers: HttpHeaders) =
   for k, v in headers:
@@ -294,13 +305,18 @@ proc processClient(server: AsyncHttpServer, client: AsyncSocket, address: string
     )
     if not retry: break
 
-proc serve*(server: AsyncHttpServer, port: Port,
+const
-            callback: proc (request: Request): Future[void] {.closure, gcsafe.},
+  nimMaxDescriptorsFallback* {.intdefine.} = 16_000 ## fallback value for \
-            address = "") {.async.} =
+    ## when `maxDescriptors` is not available.
-  ## Starts the process of listening for incoming HTTP connections on the
+    ## This can be set on the command line during compilation
-  ## specified address and port.
+    ## via `-d:nimMaxDescriptorsFallback=N`
-  ##
+
-  ## When a request is made by a client the specified callback will be called.
+proc listen*(server: AsyncHttpServer; port: Port; address = "") =
+  ## Listen to the given port and address.
+  when declared(maxDescriptors):
+    server.maxFDs = try: maxDescriptors() except: nimMaxDescriptorsFallback
+  else:
+    server.maxFDs = nimMaxDescriptorsFallback
   server.socket = newAsyncSocket()
   if server.reuseAddr:
     server.socket.setSockOpt(OptReuseAddr, true)
@@ -309,9 +325,44 @@ proc serve*(server: AsyncHttpServer, port: Port,
   server.socket.bindAddr(port, address)
   server.socket.listen()
 
-  while true:
+proc shouldAcceptRequest*(server: AsyncHttpServer;
+                          assumedDescriptorsPerRequest = 5): bool {.inline.} =
+  ## Returns true if the process's current number of opened file
+  ## descriptors is still within the maximum limit and so it's reasonable to
+  ## accept yet another request.
+  result = assumedDescriptorsPerRequest < 0 or
+    (activeDescriptors() + assumedDescriptorsPerRequest < server.maxFDs)
+
+proc acceptRequest*(server: AsyncHttpServer, port: Port,
+            callback: proc (request: Request): Future[void] {.closure, gcsafe.}) {.async.} =
+  ## Accepts a single request. Write an explicit loop around this proc so that
+  ## errors can be handled properly.
   var (address, client) = await server.socket.acceptAddr()
   asyncCheck processClient(server, client, address, callback)
+
+proc serve*(server: AsyncHttpServer, port: Port,
+            callback: proc (request: Request): Future[void] {.closure, gcsafe.},
+            address = "";
+            assumedDescriptorsPerRequest = 5) {.async.} =
+  ## Starts the process of listening for incoming HTTP connections on the
+  ## specified address and port.
+  ##
+  ## When a request is made by a client the specified callback will be called.
+  ##
+  ## If `assumedDescriptorsPerRequest` is 0 or greater the server cares about
+  ## the process's maximum file descriptor limit. It then ensures that the
+  ## process still has the resources for `assumedDescriptorsPerRequest`
+  ## file descriptors before accepting a connection.
+  ##
+  ## You should prefer to call `acceptRequest` instead with a custom server
+  ## loop so that you're in control over the error handling and logging.
+  listen server, port, address
+  while true:
+    if shouldAcceptRequest(server, assumedDescriptorsPerRequest):
+      var (address, client) = await server.socket.acceptAddr()
+      asyncCheck processClient(server, client, address, callback)
+    else:
+      poll()
     #echo(f.isNil)
     #echo(f.repr)
 
@@ -320,7 +371,7 @@ proc close*(server: AsyncHttpServer) =
   server.socket.close()
 
 when not defined(testing) and isMainModule:
-  proc main =
+  proc main {.async.} =
     var server = newAsyncHttpServer()
     proc cb(req: Request) {.async.} =
       #echo(req.reqMethod, " ", req.url)
@@ -329,6 +380,13 @@ when not defined(testing) and isMainModule:
           "Content-type": "text/plain; charset=utf-8"}
       await req.respond(Http200, "Hello World", headers.newHttpHeaders())
 
-    asyncCheck server.serve(Port(5555), cb)
+    server.listen Port(5555)
+    while true:
+      if server.shouldAcceptRequest(5):
+        var (address, client) = await server.socket.acceptAddr()
+        asyncCheck processClient(server, client, address, cb)
+      else:
+        poll()
+
+  asyncCheck main()
   runForever()
-  main()

lib/pure/ioselects/ioselectors_epoll.nim

@@ -55,7 +55,7 @@ when hasThreadSupport:
       maxFD: int
       numFD: int
       fds: ptr SharedArray[SelectorKey[T]]
-      count: int
+      count*: int
     Selector*[T] = ptr SelectorImpl[T]
 else:
   type
@@ -64,7 +64,7 @@ else:
       maxFD: int
       numFD: int
       fds: seq[SelectorKey[T]]
-      count: int
+      count*: int
     Selector*[T] = ref SelectorImpl[T]
 type
   SelectEventImpl = object

lib/pure/ioselects/ioselectors_kqueue.nim

@@ -30,7 +30,7 @@ when defined(macosx) or defined(freebsd) or defined(dragonfly):
   proc sysctl(name: ptr cint, namelen: cuint, oldp: pointer, oldplen: ptr csize_t,
               newp: pointer, newplen: csize_t): cint
        {.importc: "sysctl",header: """#include <sys/types.h>
-                                      #include <sys/sysctl.h>"""}
+                                      #include <sys/sysctl.h>""".}
 elif defined(netbsd) or defined(openbsd):
   # OpenBSD and NetBSD don't have KERN_MAXFILESPERPROC, so we are using
   # KERN_MAXFILES, because KERN_MAXFILES is always bigger,
@@ -39,7 +39,7 @@ elif defined(netbsd) or defined(openbsd):
   proc sysctl(name: ptr cint, namelen: cuint, oldp: pointer, oldplen: ptr csize_t,
               newp: pointer, newplen: csize_t): cint
        {.importc: "sysctl",header: """#include <sys/param.h>
-                                      #include <sys/sysctl.h>"""}
+                                      #include <sys/sysctl.h>""".}
 
 when hasThreadSupport:
   type
@@ -48,7 +48,7 @@ when hasThreadSupport:
       maxFD: int
       changes: ptr SharedArray[KEvent]
       fds: ptr SharedArray[SelectorKey[T]]
-      count: int
+      count*: int
       changesLock: Lock
       changesSize: int
       changesLength: int
@@ -61,7 +61,7 @@ else:
       maxFD: int
       changes: seq[KEvent]
       fds: seq[SelectorKey[T]]
-      count: int
+      count*: int
       sock: cint
     Selector*[T] = ref SelectorImpl[T]
 

lib/pure/ioselects/ioselectors_poll.nim

@@ -21,7 +21,7 @@ when hasThreadSupport:
       pollcnt: int
       fds: ptr SharedArray[SelectorKey[T]]
       pollfds: ptr SharedArray[TPollFd]
-      count: int
+      count*: int
       lock: Lock
     Selector*[T] = ptr SelectorImpl[T]
 else:
@@ -31,7 +31,7 @@ else:
       pollcnt: int
       fds: seq[SelectorKey[T]]
       pollfds: seq[TPollFd]
-      count: int
+      count*: int
     Selector*[T] = ref SelectorImpl[T]
 
 type

lib/pure/ioselects/ioselectors_select.nim

@@ -58,7 +58,7 @@ when hasThreadSupport:
       eSet: FdSet
       maxFD: int
       fds: ptr SharedArray[SelectorKey[T]]
-      count: int
+      count*: int
       lock: Lock
     Selector*[T] = ptr SelectorImpl[T]
 else:
@@ -69,7 +69,7 @@ else:
       eSet: FdSet
       maxFD: int
       fds: seq[SelectorKey[T]]
-      count: int
+      count*: int
     Selector*[T] = ref SelectorImpl[T]
 
 type

tests/errmsgs/tgcsafety.nim

@@ -6,7 +6,7 @@ type mismatch: got <AsyncHttpServer, Port, proc (req: Request): Future[system.vo
 but expected one of:
 proc serve(server: AsyncHttpServer; port: Port;
            callback: proc (request: Request): Future[void] {.closure, gcsafe.};
-          address = ""): owned(Future[void])
+           address = ""; assumedDescriptorsPerRequest = 5): owned(Future[void])
   first type mismatch at position: 3
   required type for callback: proc (request: Request): Future[system.void]{.closure, gcsafe.}
   but expression 'cb' is of type: proc (req: Request): Future[system.void]{.locks: <unknown>.}