Implement dial, support IPv6 in httpclient (#5763)
* Implement dial, support IPv6 in httpclient Added ``dial`` procedure to networking modules: ``net``, ``asyncdispatch``, ``asyncnet``. It merges socket creation, address resolution, and connection into single step. When using ``dial``, you don't have to worry about IPv4 vs IPv6 problem. Fixed addrInfo loop in connect to behave properly. Previously it would stop on first non-immediate failure, instead of continuing and trying the remaining addresses. Fixed newAsyncNativeSocket to raise proper error if socket creation fails. Fixes: #3811 * Check domain during connect() only on non-Windows This is how it was in the previous implementation of connect(). * Call 'osLastError' before 'close' in net.dial * Record osLastError before freeAddrInfo in net.dial * Add missing docs for 'dial' proc * Optimize dial to create one FD per domain, add tests And make async IPv6 servers work on Windows. * Add IPv6 test to uri module * Fix getAddrString error handling
This commit is contained in:
Ruslan Mustakov
Andreas Rumpf
parent
6377b52d8e
commit
ecf278c467
13 changed files with 559 additions and 334 deletions
|
|
@ -9,7 +9,7 @@
|
|||
|
||||
include "system/inclrtl"
|
||||
|
||||
import os, oids, tables, strutils, times, heapqueue
|
||||
import os, oids, tables, strutils, times, heapqueue, options
|
||||
|
||||
import nativesockets, net, deques
|
||||
|
||||
|
|
@ -385,68 +385,6 @@ when defined(windows) or defined(nimdoc):
|
|||
dwRemoteAddressLength, LocalSockaddr, LocalSockaddrLength,
|
||||
RemoteSockaddr, RemoteSockaddrLength)
|
||||
|
||||
proc connect*(socket: AsyncFD, address: string, port: Port,
|
||||
domain = nativesockets.AF_INET): Future[void] =
|
||||
## Connects ``socket`` to server at ``address:port``.
|
||||
##
|
||||
## Returns a ``Future`` which will complete when the connection succeeds
|
||||
## or an error occurs.
|
||||
verifyPresence(socket)
|
||||
var retFuture = newFuture[void]("connect")
|
||||
# Apparently ``ConnectEx`` expects the socket to be initially bound:
|
||||
var saddr: Sockaddr_in
|
||||
saddr.sin_family = int16(toInt(domain))
|
||||
saddr.sin_port = 0
|
||||
saddr.sin_addr.s_addr = INADDR_ANY
|
||||
if bindAddr(socket.SocketHandle, cast[ptr SockAddr](addr(saddr)),
|
||||
sizeof(saddr).SockLen) < 0'i32:
|
||||
raiseOSError(osLastError())
|
||||
|
||||
var aiList = getAddrInfo(address, port, domain)
|
||||
var success = false
|
||||
var lastError: OSErrorCode
|
||||
var it = aiList
|
||||
while it != nil:
|
||||
# "the OVERLAPPED structure must remain valid until the I/O completes"
|
||||
# http://blogs.msdn.com/b/oldnewthing/archive/2011/02/02/10123392.aspx
|
||||
var ol = PCustomOverlapped()
|
||||
GC_ref(ol)
|
||||
ol.data = CompletionData(fd: socket, cb:
|
||||
proc (fd: AsyncFD, bytesCount: Dword, errcode: OSErrorCode) =
|
||||
if not retFuture.finished:
|
||||
if errcode == OSErrorCode(-1):
|
||||
retFuture.complete()
|
||||
else:
|
||||
retFuture.fail(newException(OSError, osErrorMsg(errcode)))
|
||||
)
|
||||
|
||||
var ret = connectEx(socket.SocketHandle, it.ai_addr,
|
||||
sizeof(Sockaddr_in).cint, nil, 0, nil,
|
||||
cast[POVERLAPPED](ol))
|
||||
if ret:
|
||||
# Request to connect completed immediately.
|
||||
success = true
|
||||
retFuture.complete()
|
||||
# We don't deallocate ``ol`` here because even though this completed
|
||||
# immediately poll will still be notified about its completion and it will
|
||||
# free ``ol``.
|
||||
break
|
||||
else:
|
||||
lastError = osLastError()
|
||||
if lastError.int32 == ERROR_IO_PENDING:
|
||||
# In this case ``ol`` will be deallocated in ``poll``.
|
||||
success = true
|
||||
break
|
||||
else:
|
||||
GC_unref(ol)
|
||||
success = false
|
||||
it = it.ai_next
|
||||
|
||||
freeAddrInfo(aiList)
|
||||
if not success:
|
||||
retFuture.fail(newException(OSError, osErrorMsg(lastError)))
|
||||
return retFuture
|
||||
|
||||
proc recv*(socket: AsyncFD, size: int,
|
||||
flags = {SocketFlag.SafeDisconn}): Future[string] =
|
||||
## Reads **up to** ``size`` bytes from ``socket``. Returned future will
|
||||
|
|
@ -754,8 +692,8 @@ when defined(windows) or defined(nimdoc):
|
|||
var lpOutputBuf = newString(lpOutputLen)
|
||||
var dwBytesReceived: Dword
|
||||
let dwReceiveDataLength = 0.Dword # We don't want any data to be read.
|
||||
let dwLocalAddressLength = Dword(sizeof(Sockaddr_in) + 16)
|
||||
let dwRemoteAddressLength = Dword(sizeof(Sockaddr_in) + 16)
|
||||
let dwLocalAddressLength = Dword(sizeof(Sockaddr_in6) + 16)
|
||||
let dwRemoteAddressLength = Dword(sizeof(Sockaddr_in6) + 16)
|
||||
|
||||
template failAccept(errcode) =
|
||||
if flags.isDisconnectionError(errcode):
|
||||
|
|
@ -785,12 +723,14 @@ when defined(windows) or defined(nimdoc):
|
|||
dwLocalAddressLength, dwRemoteAddressLength,
|
||||
addr localSockaddr, addr localLen,
|
||||
addr remoteSockaddr, addr remoteLen)
|
||||
register(clientSock.AsyncFD)
|
||||
# TODO: IPv6. Check ``sa_family``. http://stackoverflow.com/a/9212542/492186
|
||||
retFuture.complete(
|
||||
(address: $inet_ntoa(cast[ptr Sockaddr_in](remoteSockAddr).sin_addr),
|
||||
client: clientSock.AsyncFD)
|
||||
)
|
||||
try:
|
||||
let address = getAddrString(remoteSockAddr)
|
||||
register(clientSock.AsyncFD)
|
||||
retFuture.complete((address: address, client: clientSock.AsyncFD))
|
||||
except:
|
||||
# getAddrString may raise
|
||||
clientSock.close()
|
||||
retFuture.fail(getCurrentException())
|
||||
|
||||
var ol = PCustomOverlapped()
|
||||
GC_ref(ol)
|
||||
|
|
@ -823,20 +763,6 @@ when defined(windows) or defined(nimdoc):
|
|||
|
||||
return retFuture
|
||||
|
||||
proc newAsyncNativeSocket*(domain, sockType, protocol: cint): AsyncFD =
|
||||
## Creates a new socket and registers it with the dispatcher implicitly.
|
||||
result = newNativeSocket(domain, sockType, protocol).AsyncFD
|
||||
result.SocketHandle.setBlocking(false)
|
||||
register(result)
|
||||
|
||||
proc newAsyncNativeSocket*(domain: Domain = nativesockets.AF_INET,
|
||||
sockType: SockType = SOCK_STREAM,
|
||||
protocol: Protocol = IPPROTO_TCP): AsyncFD =
|
||||
## Creates a new socket and registers it with the dispatcher implicitly.
|
||||
result = newNativeSocket(domain, sockType, protocol).AsyncFD
|
||||
result.SocketHandle.setBlocking(false)
|
||||
register(result)
|
||||
|
||||
proc closeSocket*(socket: AsyncFD) =
|
||||
## Closes a socket and ensures that it is unregistered.
|
||||
socket.SocketHandle.close()
|
||||
|
|
@ -1015,23 +941,6 @@ else:
|
|||
var data = PData(fd: fd, readCBs: @[], writeCBs: @[])
|
||||
p.selector.register(fd.SocketHandle, {}, data.RootRef)
|
||||
|
||||
proc newAsyncNativeSocket*(domain: cint, sockType: cint,
|
||||
protocol: cint): AsyncFD =
|
||||
result = newNativeSocket(domain, sockType, protocol).AsyncFD
|
||||
result.SocketHandle.setBlocking(false)
|
||||
when defined(macosx):
|
||||
result.SocketHandle.setSockOptInt(SOL_SOCKET, SO_NOSIGPIPE, 1)
|
||||
register(result)
|
||||
|
||||
proc newAsyncNativeSocket*(domain: Domain = AF_INET,
|
||||
sockType: SockType = SOCK_STREAM,
|
||||
protocol: Protocol = IPPROTO_TCP): AsyncFD =
|
||||
result = newNativeSocket(domain, sockType, protocol).AsyncFD
|
||||
result.SocketHandle.setBlocking(false)
|
||||
when defined(macosx):
|
||||
result.SocketHandle.setSockOptInt(SOL_SOCKET, SO_NOSIGPIPE, 1)
|
||||
register(result)
|
||||
|
||||
proc closeSocket*(sock: AsyncFD) =
|
||||
let disp = getGlobalDispatcher()
|
||||
disp.selector.unregister(sock.SocketHandle)
|
||||
|
|
@ -1115,50 +1024,6 @@ else:
|
|||
# Callback queue processing
|
||||
processPendingCallbacks(p)
|
||||
|
||||
proc connect*(socket: AsyncFD, address: string, port: Port,
|
||||
domain = AF_INET): Future[void] =
|
||||
var retFuture = newFuture[void]("connect")
|
||||
|
||||
proc cb(fd: AsyncFD): bool =
|
||||
var ret = SocketHandle(fd).getSockOptInt(cint(SOL_SOCKET), cint(SO_ERROR))
|
||||
if ret == 0:
|
||||
# We have connected.
|
||||
retFuture.complete()
|
||||
return true
|
||||
elif ret == EINTR:
|
||||
# interrupted, keep waiting
|
||||
return false
|
||||
else:
|
||||
retFuture.fail(newException(OSError, osErrorMsg(OSErrorCode(ret))))
|
||||
return true
|
||||
|
||||
assert getSockDomain(socket.SocketHandle) == domain
|
||||
var aiList = getAddrInfo(address, port, domain)
|
||||
var success = false
|
||||
var lastError: OSErrorCode
|
||||
var it = aiList
|
||||
while it != nil:
|
||||
var ret = connect(socket.SocketHandle, it.ai_addr, it.ai_addrlen.Socklen)
|
||||
if ret == 0:
|
||||
# Request to connect completed immediately.
|
||||
success = true
|
||||
retFuture.complete()
|
||||
break
|
||||
else:
|
||||
lastError = osLastError()
|
||||
if lastError.int32 == EINTR or lastError.int32 == EINPROGRESS:
|
||||
success = true
|
||||
addWrite(socket, cb)
|
||||
break
|
||||
else:
|
||||
success = false
|
||||
it = it.ai_next
|
||||
|
||||
freeAddrInfo(aiList)
|
||||
if not success:
|
||||
retFuture.fail(newException(OSError, osErrorMsg(lastError)))
|
||||
return retFuture
|
||||
|
||||
proc recv*(socket: AsyncFD, size: int,
|
||||
flags = {SocketFlag.SafeDisconn}): Future[string] =
|
||||
var retFuture = newFuture[string]("recv")
|
||||
|
|
@ -1320,11 +1185,20 @@ else:
|
|||
else:
|
||||
retFuture.fail(newException(OSError, osErrorMsg(lastError)))
|
||||
else:
|
||||
register(client.AsyncFD)
|
||||
retFuture.complete((getAddrString(cast[ptr SockAddr](addr sockAddress)), client.AsyncFD))
|
||||
try:
|
||||
let address = getAddrString(cast[ptr SockAddr](addr sockAddress))
|
||||
register(client.AsyncFD)
|
||||
retFuture.complete((address, client.AsyncFD))
|
||||
except:
|
||||
# getAddrString may raise
|
||||
client.close()
|
||||
retFuture.fail(getCurrentException())
|
||||
addRead(socket, cb)
|
||||
return retFuture
|
||||
|
||||
# Common procedures between current and upcoming asyncdispatch
|
||||
include includes.asynccommon
|
||||
|
||||
proc sleepAsync*(ms: int): Future[void] =
|
||||
## Suspends the execution of the current async procedure for the next
|
||||
## ``ms`` milliseconds.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue