Merge branch 'devel' of https://github.com/Araq/Nimrod into devel

lib/pure/asyncdispatch.nim

@@ -9,14 +9,14 @@
 
 import os, oids, tables, strutils, macros
 
-import sockets2, net
+import rawsockets
 
-## Asyncio2 
+## AsyncDispatch
 ## --------
 ##
-## This module implements a brand new asyncio module based on Futures.
+## This module implements a brand new dispatcher based on Futures.
-## IOCP is used under the hood on Windows and the selectors module is used for
+## On Windows IOCP is used and on other operating systems the selectors module
-## other operating systems.
+## is used instead.
 
 # -- Futures
 
@@ -27,7 +27,7 @@ type
 
   PFuture*[T] = ref object of PFutureBase
     value: T
-    error: ref EBase
+    error*: ref EBase # TODO: This shouldn't be necessary, generics bug?
 
 proc newFuture*[T](): PFuture[T] =
   ## Creates a new future.
@@ -90,21 +90,23 @@ proc read*[T](future: PFuture[T]): T =
     # TODO: Make a custom exception type for this?
     raise newException(EInvalidValue, "Future still in progress.")
 
+proc readError*[T](future: PFuture[T]): ref EBase =
+  if future.error != nil: return future.error
+  else:
+    raise newException(EInvalidValue, "No error in future.")
+
 proc finished*[T](future: PFuture[T]): bool =
   ## Determines whether ``future`` has completed.
   ##
-  ## ``True`` may indicate an error or a value. Use ``hasError`` to distinguish.
+  ## ``True`` may indicate an error or a value. Use ``failed`` to distinguish.
   future.finished
 
 proc failed*[T](future: PFuture[T]): bool =
   ## Determines whether ``future`` completed with an error.
   future.error != nil
 
-# TODO: Get rid of register. Do it implicitly.
-
 when defined(windows) or defined(nimdoc):
   import winlean, sets, hashes
-  #from hashes import THash
   type
     TCompletionKey = dword
 
@@ -481,6 +483,7 @@ when defined(windows) or defined(nimdoc):
                protocol: TProtocol = IPPROTO_TCP): TSocketHandle =
     ## Creates a new socket and registers it with the dispatcher implicitly.
     result = socket(domain, typ, protocol)
+    result.setBlocking(false)
     disp.register(result)
 
   proc close*(disp: PDispatcher, socket: TSocketHandle) =
@@ -519,6 +522,7 @@ else:
                typ: TType = SOCK_STREAM,
                protocol: TProtocol = IPPROTO_TCP): TSocketHandle =
     result = socket(domain, typ, protocol)
+    result.setBlocking(false)
     disp.register(result)
   
   proc close*(disp: PDispatcher, sock: TSocketHandle) =
@@ -775,7 +779,7 @@ proc processBody(node, retFutureSym: PNimrodNode): PNimrodNode {.compileTime.} =
     else: discard
   of nnkDiscardStmt:
     # discard await x
-    if node[0][0].ident == !"await":
+    if node[0][0].kind == nnkIdent and node[0][0].ident == !"await":
       var dummy = newNimNode(nnkStmtList)
       createVar("futureDiscard_" & $toStrLit(node[0][1]), node[0][1], dummy)
   else: discard
@@ -794,6 +798,9 @@ proc getName(node: PNimrodNode): string {.compileTime.} =
     assert false
 
 macro async*(prc: stmt): stmt {.immediate.} =
+  ## Macro which processes async procedures into the appropriate
+  ## iterators and yield statements.
+
   expectKind(prc, nnkProcDef)
 
   hint("Processing " & prc[0].getName & " as an async proc.")
@@ -893,6 +900,10 @@ proc recvLine*(p: PDispatcher, socket: TSocketHandle): PFuture[string] {.async.}
   ## will be set to it.
   ## 
   ## If the socket is disconnected, ``line`` will be set to ``""``.
+  ##
+  ## If the socket is disconnected in the middle of a line (before ``\r\L``
+  ## is read) then line will be set to ``""``.
+  ## The partial line **will be lost**.
   
   template addNLIfEmpty(): stmt =
     if result.len == 0:
@@ -901,13 +912,9 @@ proc recvLine*(p: PDispatcher, socket: TSocketHandle): PFuture[string] {.async.}
   result = ""
   var c = ""
   while true:
-    #echo("1")
     c = await p.recv(socket, 1)
-    #echo("Received ", c.len)
     if c.len == 0:
-      #echo("returning")
+      return ""
-      return
-    #echo("2")
     if c == "\r":
       c = await p.recv(socket, 1, MSG_PEEK)
       if c.len > 0 and c == "\L":
@@ -917,9 +924,16 @@ proc recvLine*(p: PDispatcher, socket: TSocketHandle): PFuture[string] {.async.}
     elif c == "\L":
       addNLIfEmpty()
       return
-    #echo("3")
     add(result.string, c)
-  #echo("4")
+
+var gDisp*{.threadvar.}: PDispatcher ## Global dispatcher
+gDisp = newDispatcher()
+
+proc runForever*() =
+  ## Begins a never ending global dispatcher poll loop.
+  while true:
+    gDisp.poll()
+
 
 when isMainModule:
   

lib/pure/asyncnet.nim

@@ -0,0 +1,147 @@
+import asyncdispatch
+import rawsockets
+import net
+
+when defined(ssl):
+  import openssl
+
+type
+  TAsyncSocket = object ## socket type
+    fd: TSocketHandle
+    case isBuffered: bool # determines whether this socket is buffered.
+    of true:
+      buffer: array[0..BufferSize, char]
+      currPos: int # current index in buffer
+      bufLen: int # current length of buffer
+    of false: nil
+    when defined(ssl):
+      case isSsl: bool
+      of true:
+        sslHandle: PSSL
+        sslContext: PSSLContext
+        sslNoHandshake: bool # True if needs handshake.
+        sslHasPeekChar: bool
+        sslPeekChar: char
+      of false: nil
+
+  PAsyncSocket* = ref TAsyncSocket
+
+# TODO: Save AF, domain etc info and reuse it in procs which need it like connect.
+
+proc newSocket(fd: TSocketHandle, isBuff: bool): PAsyncSocket =
+  assert fd != osInvalidSocket
+  new(result)
+  result.fd = fd
+  result.isBuffered = isBuff
+  if isBuff:
+    result.currPos = 0
+
+proc AsyncSocket*(domain: TDomain = AF_INET, typ: TType = SOCK_STREAM,
+    protocol: TProtocol = IPPROTO_TCP, buffered = true): PAsyncSocket =
+  ## Creates a new asynchronous socket.
+  result = newSocket(gDisp.socket(domain, typ, protocol), buffered)
+
+proc connect*(socket: PAsyncSocket, address: string, port: TPort,
+    af = AF_INET): PFuture[void] =
+  ## Connects ``socket`` to server at ``address:port``.
+  ##
+  ## Returns a ``PFuture`` which will complete when the connection succeeds
+  ## or an error occurs.
+  result = gDisp.connect(socket.fd, address, port, af)
+
+proc recv*(socket: PAsyncSocket, size: int,
+           flags: int = 0): PFuture[string] =
+  ## Reads ``size`` bytes from ``socket``. Returned future will complete once
+  ## all of the requested data is read. If socket is disconnected during the
+  ## recv operation then the future may complete with only a part of the
+  ## requested data read. If socket is disconnected and no data is available
+  ## to be read then the future will complete with a value of ``""``.
+  result = gDisp.recv(socket.fd, size, flags)
+
+proc send*(socket: PAsyncSocket, data: string): PFuture[void] =
+  ## Sends ``data`` to ``socket``. The returned future will complete once all
+  ## data has been sent.
+  result = gDisp.send(socket.fd, data)
+
+proc acceptAddr*(socket: PAsyncSocket): 
+      PFuture[tuple[address: string, client: PAsyncSocket]] =
+  ## Accepts a new connection. Returns a future containing the client socket
+  ## corresponding to that connection and the remote address of the client.
+  ## The future will complete when the connection is successfully accepted.
+  var retFuture = newFuture[tuple[address: string, client: PAsyncSocket]]()
+  var fut = gDisp.acceptAddr(socket.fd)
+  fut.callback =
+    proc (future: PFuture[tuple[address: string, client: TSocketHandle]]) =
+      assert future.finished
+      if future.failed:
+        retFuture.fail(future.readError)
+      else:
+        let resultTup = (future.read.address,
+                         newSocket(future.read.client, socket.isBuffered))
+        retFuture.complete(resultTup)
+  return retFuture
+
+proc accept*(socket: PAsyncSocket): PFuture[PAsyncSocket] =
+  ## Accepts a new connection. Returns a future containing the client socket
+  ## corresponding to that connection.
+  ## The future will complete when the connection is successfully accepted.
+  var retFut = newFuture[PAsyncSocket]()
+  var fut = acceptAddr(socket)
+  fut.callback =
+    proc (future: PFuture[tuple[address: string, client: PAsyncSocket]]) =
+      assert future.finished
+      if future.failed:
+        retFut.fail(future.readError)
+      else:
+        retFut.complete(future.read.client)
+  return retFut
+
+proc recvLine*(socket: PAsyncSocket): PFuture[string] {.async.} =
+  ## Reads a line of data from ``socket``. Returned future will complete once
+  ## a full line is read or an error occurs.
+  ##
+  ## If a full line is read ``\r\L`` is not
+  ## added to ``line``, however if solely ``\r\L`` is read then ``line``
+  ## will be set to it.
+  ## 
+  ## If the socket is disconnected, ``line`` will be set to ``""``.
+  ##
+  ## If the socket is disconnected in the middle of a line (before ``\r\L``
+  ## is read) then line will be set to ``""``.
+  ## The partial line **will be lost**.
+  
+  template addNLIfEmpty(): stmt =
+    if result.len == 0:
+      result.add("\c\L")
+
+  result = ""
+  var c = ""
+  while true:
+    c = await recv(socket, 1)
+    if c.len == 0:
+      return ""
+    if c == "\r":
+      c = await recv(socket, 1, MSG_PEEK)
+      if c.len > 0 and c == "\L":
+        discard await recv(socket, 1)
+      addNLIfEmpty()
+      return
+    elif c == "\L":
+      addNLIfEmpty()
+      return
+    add(result.string, c)
+
+when isMainModule:
+  proc main() {.async.} =
+    var sock = AsyncSocket()
+    await sock.connect("irc.freenode.net", TPort(6667))
+    while true:
+      let line = await sock.recvLine()
+      if line == "":
+        echo("Disconnected")
+        break
+      else:
+        echo("Got line: ", line)
+  main()
+  runForever()
+    

lib/pure/rawsockets.nim

@@ -10,6 +10,8 @@
 ## This module implements a low-level cross-platform sockets interface. Look
 ## at the ``net`` module for the higher-level version.
 
+# TODO: Clean up the exports a bit and everything else in general.
+
 import unsigned, os
 
 when hostos == "solaris":
@@ -17,16 +19,21 @@ when hostos == "solaris":
 
 when defined(Windows):
   import winlean
+  export WSAEWOULDBLOCK
 else:
   import posix
-  export fcntl, F_GETFL, O_NONBLOCK, F_SETFL
+  export fcntl, F_GETFL, O_NONBLOCK, F_SETFL, EAGAIN, EWOULDBLOCK
 
 export TSocketHandle, TSockaddr_in, TAddrinfo, INADDR_ANY, TSockAddr, TSockLen,
-  inet_ntoa, recv, `==`, connect, send, accept
+  inet_ntoa, recv, `==`, connect, send, accept, recvfrom, sendto
 
 export
   SO_ERROR,
-  SOL_SOCKET
+  SOL_SOCKET,
+  SOMAXCONN,
+  SO_ACCEPTCONN, SO_BROADCAST, SO_DEBUG, SO_DONTROUTE,
+  SO_KEEPALIVE, SO_OOBINLINE, SO_REUSEADDR,
+  MSG_PEEK
 
 type
   
@@ -138,8 +145,6 @@ else:
 proc socket*(domain: TDomain = AF_INET, typ: TType = SOCK_STREAM,
              protocol: TProtocol = IPPROTO_TCP): TSocketHandle =
   ## Creates a new socket; returns `InvalidSocket` if an error occurs.
-  
-  # TODO: The function which will use this will raise EOS.
   socket(toInt(domain), toInt(typ), toInt(protocol))
 
 proc close*(socket: TSocketHandle) =
@@ -154,14 +159,14 @@ proc close*(socket: TSocketHandle) =
 proc bindAddr*(socket: TSocketHandle, name: ptr TSockAddr, namelen: TSockLen): cint =
   result = bindSocket(socket, name, namelen)
 
-proc listen*(socket: TSocketHandle, backlog = SOMAXCONN) {.tags: [FReadIO].} =
+proc listen*(socket: TSocketHandle, backlog = SOMAXCONN): cint {.tags: [FReadIO].} =
   ## Marks ``socket`` as accepting connections. 
   ## ``Backlog`` specifies the maximum length of the 
   ## queue of pending connections.
   when defined(windows):
-    if winlean.listen(socket, cint(backlog)) < 0'i32: osError(osLastError())
+    result = winlean.listen(socket, cint(backlog))
   else:
-    if posix.listen(socket, cint(backlog)) < 0'i32: osError(osLastError())
+    result = posix.listen(socket, cint(backlog))
 
 proc getAddrInfo*(address: string, port: TPort, af: TDomain = AF_INET, typ: TType = SOCK_STREAM,
                  prot: TProtocol = IPPROTO_TCP): ptr TAddrInfo =
@@ -204,13 +209,110 @@ proc htonl*(x: int32): int32 =
   ## Converts 32-bit integers from host to network byte order. On machines
   ## where the host byte order is the same as network byte order, this is
   ## a no-op; otherwise, it performs a 4-byte swap operation.
-  result = sockets2.ntohl(x)
+  result = rawsockets.ntohl(x)
 
 proc htons*(x: int16): int16 =
   ## Converts 16-bit positive integers from host to network byte order.
   ## On machines where the host byte order is the same as network byte
   ## order, this is a no-op; otherwise, it performs a 2-byte swap operation.
-  result = sockets2.ntohs(x)
+  result = rawsockets.ntohs(x)
+
+proc getServByName*(name, proto: string): TServent {.tags: [FReadIO].} =
+  ## Searches the database from the beginning and finds the first entry for 
+  ## which the service name specified by ``name`` matches the s_name member
+  ## and the protocol name specified by ``proto`` matches the s_proto member.
+  ##
+  ## On posix this will search through the ``/etc/services`` file.
+  when defined(Windows):
+    var s = winlean.getservbyname(name, proto)
+  else:
+    var s = posix.getservbyname(name, proto)
+  if s == nil: raise newException(EOS, "Service not found.")
+  result.name = $s.s_name
+  result.aliases = cstringArrayToSeq(s.s_aliases)
+  result.port = TPort(s.s_port)
+  result.proto = $s.s_proto
+  
+proc getServByPort*(port: TPort, proto: string): TServent {.tags: [FReadIO].} = 
+  ## Searches the database from the beginning and finds the first entry for 
+  ## which the port specified by ``port`` matches the s_port member and the 
+  ## protocol name specified by ``proto`` matches the s_proto member.
+  ##
+  ## On posix this will search through the ``/etc/services`` file.
+  when defined(Windows):
+    var s = winlean.getservbyport(ze(int16(port)).cint, proto)
+  else:
+    var s = posix.getservbyport(ze(int16(port)).cint, proto)
+  if s == nil: raise newException(EOS, "Service not found.")
+  result.name = $s.s_name
+  result.aliases = cstringArrayToSeq(s.s_aliases)
+  result.port = TPort(s.s_port)
+  result.proto = $s.s_proto
+
+proc getHostByAddr*(ip: string): Thostent {.tags: [FReadIO].} =
+  ## This function will lookup the hostname of an IP Address.
+  var myaddr: TInAddr
+  myaddr.s_addr = inet_addr(ip)
+  
+  when defined(windows):
+    var s = winlean.gethostbyaddr(addr(myaddr), sizeof(myaddr).cuint,
+                                  cint(rawsockets.AF_INET))
+    if s == nil: osError(osLastError())
+  else:
+    var s = posix.gethostbyaddr(addr(myaddr), sizeof(myaddr).TSocklen, 
+                                cint(posix.AF_INET))
+    if s == nil:
+      raise newException(EOS, $hstrerror(h_errno))
+  
+  result.name = $s.h_name
+  result.aliases = cstringArrayToSeq(s.h_aliases)
+  when defined(windows): 
+    result.addrtype = TDomain(s.h_addrtype)
+  else:
+    if s.h_addrtype == posix.AF_INET:
+      result.addrtype = AF_INET
+    elif s.h_addrtype == posix.AF_INET6:
+      result.addrtype = AF_INET6
+    else:
+      raise newException(EOS, "unknown h_addrtype")
+  result.addrList = cstringArrayToSeq(s.h_addr_list)
+  result.length = int(s.h_length)
+
+proc getHostByName*(name: string): Thostent {.tags: [FReadIO].} = 
+  ## This function will lookup the IP address of a hostname.
+  when defined(Windows):
+    var s = winlean.gethostbyname(name)
+  else:
+    var s = posix.gethostbyname(name)
+  if s == nil: osError(osLastError())
+  result.name = $s.h_name
+  result.aliases = cstringArrayToSeq(s.h_aliases)
+  when defined(windows): 
+    result.addrtype = TDomain(s.h_addrtype)
+  else:
+    if s.h_addrtype == posix.AF_INET:
+      result.addrtype = AF_INET
+    elif s.h_addrtype == posix.AF_INET6:
+      result.addrtype = AF_INET6
+    else:
+      raise newException(EOS, "unknown h_addrtype")
+  result.addrList = cstringArrayToSeq(s.h_addr_list)
+  result.length = int(s.h_length)
+
+proc getSockName*(socket: TSocketHandle): TPort = 
+  ## returns the socket's associated port number.
+  var name: Tsockaddr_in
+  when defined(Windows):
+    name.sin_family = int16(ord(AF_INET))
+  else:
+    name.sin_family = posix.AF_INET
+  #name.sin_port = htons(cint16(port))
+  #name.sin_addr.s_addr = htonl(INADDR_ANY)
+  var namelen = sizeof(name).TSocklen
+  if getsockname(socket, cast[ptr TSockAddr](addr(name)),
+                 addr(namelen)) == -1'i32:
+    osError(osLastError())
+  result = TPort(rawsockets.ntohs(name.sin_port))
 
 proc getSockOptInt*(socket: TSocketHandle, level, optname: int): int {.
   tags: [FReadIO].} = 
@@ -230,6 +332,89 @@ proc setSockOptInt*(socket: TSocketHandle, level, optname, optval: int) {.
                 sizeof(value).TSocklen) < 0'i32:
     osError(osLastError())
 
+proc setBlocking*(s: TSocketHandle, blocking: bool) =
+  ## Sets blocking mode on socket.
+  ##
+  ## Raises EOS on error.
+  when defined(Windows):
+    var mode = clong(ord(not blocking)) # 1 for non-blocking, 0 for blocking
+    if ioctlsocket(s, FIONBIO, addr(mode)) == -1:
+      osError(osLastError())
+  else: # BSD sockets
+    var x: int = fcntl(s, F_GETFL, 0)
+    if x == -1:
+      osError(osLastError())
+    else:
+      var mode = if blocking: x and not O_NONBLOCK else: x or O_NONBLOCK
+      if fcntl(s, F_SETFL, mode) == -1:
+        osError(osLastError())
+
+proc timeValFromMilliseconds(timeout = 500): Ttimeval =
+  if timeout != -1:
+    var seconds = timeout div 1000
+    result.tv_sec = seconds.int32
+    result.tv_usec = ((timeout - seconds * 1000) * 1000).int32
+
+proc createFdSet(fd: var TFdSet, s: seq[TSocketHandle], m: var int) = 
+  FD_ZERO(fd)
+  for i in items(s): 
+    m = max(m, int(i))
+    FD_SET(i, fd)
+   
+proc pruneSocketSet(s: var seq[TSocketHandle], fd: var TFdSet) = 
+  var i = 0
+  var L = s.len
+  while i < L:
+    if FD_ISSET(s[i], fd) == 0'i32:
+      s[i] = s[L-1]
+      dec(L)
+    else:
+      inc(i)
+  setLen(s, L)
+
+proc select*(readfds: var seq[TSocketHandle], timeout = 500): int =
+  ## Traditional select function. This function will return the number of
+  ## sockets that are ready to be read from, written to, or which have errors.
+  ## If there are none; 0 is returned. 
+  ## ``Timeout`` is in miliseconds and -1 can be specified for no timeout.
+  ## 
+  ## A socket is removed from the specific ``seq`` when it has data waiting to
+  ## be read/written to or has errors (``exceptfds``).
+  var tv {.noInit.}: Ttimeval = timeValFromMilliseconds(timeout)
+  
+  var rd: TFdSet
+  var m = 0
+  createFdSet((rd), readfds, m)
+  
+  if timeout != -1:
+    result = int(select(cint(m+1), addr(rd), nil, nil, addr(tv)))
+  else:
+    result = int(select(cint(m+1), addr(rd), nil, nil, nil))
+  
+  pruneSocketSet(readfds, (rd))
+
+proc selectWrite*(writefds: var seq[TSocketHandle], 
+                  timeout = 500): int {.tags: [FReadIO].} =
+  ## When a socket in ``writefds`` is ready to be written to then a non-zero
+  ## value will be returned specifying the count of the sockets which can be
+  ## written to. The sockets which can be written to will also be removed
+  ## from ``writefds``.
+  ##
+  ## ``timeout`` is specified in miliseconds and ``-1`` can be specified for
+  ## an unlimited time.
+  var tv {.noInit.}: Ttimeval = timeValFromMilliseconds(timeout)
+  
+  var wr: TFdSet
+  var m = 0
+  createFdSet((wr), writefds, m)
+  
+  if timeout != -1:
+    result = int(select(cint(m+1), nil, addr(wr), nil, addr(tv)))
+  else:
+    result = int(select(cint(m+1), nil, addr(wr), nil, nil))
+  
+  pruneSocketSet(writefds, (wr))
+
 when defined(Windows):
   var wsa: TWSADATA
   if WSAStartup(0x0101'i16, addr wsa) != 0: OSError(OSLastError())

tests/async/tasyncawait.nim

@@ -3,7 +3,7 @@ discard """
   cmd: "nimrod cc --hints:on $# $#"
   output: "5000"
 """
-import asyncio2, sockets2, net, strutils
+import asyncio2, sockets2, net, strutils, os
 
 var disp = newDispatcher()
 var msgCount = 0
@@ -50,8 +50,19 @@ proc readMessages(disp: PDispatcher, client: TSocketHandle) {.async.} =
 proc createServer(disp: PDispatcher, port: TPort) {.async.} =
   var server = disp.socket()
   #disp.register(server)
-  server.bindAddr(port)
+  block:
-  server.listen()
+    var name: TSockaddr_in
+    when defined(windows):
+      name.sin_family = toInt(AF_INET).int16
+    else:
+      name.sin_family = toInt(AF_INET)
+    name.sin_port = htons(int16(port))
+    name.sin_addr.s_addr = htonl(INADDR_ANY)
+    if bindAddr(server, cast[ptr TSockAddr](addr(name)),
+                  sizeof(name).TSocklen) < 0'i32:
+      osError(osLastError())
+  
+  discard server.listen()
   while true:
     var client = await disp.accept(server)
     readMessages(disp, client)