nim-chronos/asyncdispatch2/transports/stream.nim

1071 lines
37 KiB
Nim
Raw Normal View History

2018-05-16 08:22:34 +00:00
#
# Asyncdispatch2 Stream Transport
# (c) Copyright 2018
# Status Research & Development GmbH
#
# Licensed under either of
# Apache License, version 2.0, (LICENSE-APACHEv2)
# MIT license (LICENSE-MIT)
2018-05-21 21:52:57 +00:00
import ../asyncloop, ../asyncsync, ../handles, ../sendfile
2018-05-16 08:22:34 +00:00
import common
import net, nativesockets, os, deques, strutils
when defined(windows):
import winlean
else:
import posix
type
2018-05-21 21:52:57 +00:00
VectorKind = enum
2018-05-22 21:03:13 +00:00
DataBuffer, # Simple buffer pointer/length
DataFile # File handle for sendfile/TransmitFile
2018-05-21 21:52:57 +00:00
type
StreamVector = object
2018-05-22 21:03:13 +00:00
kind: VectorKind # Writer vector source kind
buf: pointer # Writer buffer pointer
buflen: int # Writer buffer size
offset: uint # Writer vector offset
writer: Future[void] # Writer vector completion Future
2018-05-21 21:52:57 +00:00
2018-05-16 08:22:34 +00:00
TransportKind* {.pure.} = enum
2018-05-22 21:03:13 +00:00
Socket, # Socket transport
Pipe, # Pipe transport
File # File transport
2018-05-16 08:22:34 +00:00
type
StreamTransport* = ref object of RootRef
fd: AsyncFD # File descriptor
state: set[TransportState] # Current Transport state
reader: Future[void] # Current reader Future
buffer: seq[byte] # Reading buffer
offset: int # Reading buffer offset
error: ref Exception # Current error
queue: Deque[StreamVector] # Writer queue
future: Future[void] # Stream life future
2018-05-21 21:52:57 +00:00
transferred: int
2018-05-16 08:22:34 +00:00
case kind*: TransportKind
of TransportKind.Socket:
domain: Domain # Socket transport domain (IPv4/IPv6)
local: TransportAddress # Local address
remote: TransportAddress # Remote address
of TransportKind.Pipe:
2018-05-21 21:52:57 +00:00
todo1: int
2018-05-16 08:22:34 +00:00
of TransportKind.File:
2018-05-21 21:52:57 +00:00
todo2: int
2018-05-16 08:22:34 +00:00
2018-05-22 22:29:07 +00:00
StreamCallback* = proc(server: StreamServer,
client: StreamTransport,
2018-05-16 08:22:34 +00:00
udata: pointer): Future[void] {.gcsafe.}
2018-05-21 21:52:57 +00:00
## New connection callback
2018-05-16 08:22:34 +00:00
StreamServer* = ref object of SocketServer
function*: StreamCallback
proc remoteAddress*(transp: StreamTransport): TransportAddress =
## Returns ``transp`` remote socket address.
if transp.kind != TransportKind.Socket:
raise newException(TransportError, "Socket required!")
if transp.remote.port == Port(0):
var saddr: Sockaddr_storage
var slen = SockLen(sizeof(saddr))
if getpeername(SocketHandle(transp.fd), cast[ptr SockAddr](addr saddr),
addr slen) != 0:
raiseOsError(osLastError())
fromSockAddr(saddr, slen, transp.remote.address, transp.remote.port)
result = transp.remote
proc localAddress*(transp: StreamTransport): TransportAddress =
## Returns ``transp`` local socket address.
if transp.kind != TransportKind.Socket:
raise newException(TransportError, "Socket required!")
if transp.local.port == Port(0):
var saddr: Sockaddr_storage
var slen = SockLen(sizeof(saddr))
if getsockname(SocketHandle(transp.fd), cast[ptr SockAddr](addr saddr),
addr slen) != 0:
raiseOsError(osLastError())
fromSockAddr(saddr, slen, transp.local.address, transp.local.port)
result = transp.local
template setReadError(t, e: untyped) =
(t).state.incl(ReadError)
(t).error = newException(TransportOsError, osErrorMsg((e)))
template setWriteError(t, e: untyped) =
(t).state.incl(WriteError)
(t).error = newException(TransportOsError, osErrorMsg((e)))
template finishReader(t: untyped) =
var reader = (t).reader
(t).reader = nil
reader.complete()
template checkPending(t: untyped) =
if not isNil((t).reader):
raise newException(TransportError, "Read operation already pending!")
template shiftBuffer(t, c: untyped) =
if (t).offset > c:
moveMem(addr((t).buffer[0]), addr((t).buffer[(c)]), (t).offset - (c))
(t).offset = (t).offset - (c)
else:
(t).offset = 0
2018-05-21 21:52:57 +00:00
template shiftVectorBuffer(v, o: untyped) =
(v).buf = cast[pointer](cast[uint]((v).buf) + uint(o))
(v).buflen -= int(o)
template shiftVectorFile(v, o: untyped) =
(v).buf = cast[pointer](cast[uint]((v).buf) - cast[uint](o))
(v).offset += cast[uint]((o))
2018-05-16 08:22:34 +00:00
when defined(windows):
import winlean
type
WindowsStreamTransport = ref object of StreamTransport
2018-05-21 21:52:57 +00:00
rwsabuf: TWSABuf # Reader WSABUF
wwsabuf: TWSABuf # Writer WSABUF
2018-05-16 08:22:34 +00:00
rovl: CustomOverlapped # Reader OVERLAPPED structure
wovl: CustomOverlapped # Writer OVERLAPPED structure
roffset: int # Pending reading offset
WindowsStreamServer* = ref object of RootRef
server: SocketServer # Server object
2018-05-21 21:52:57 +00:00
domain: Domain # Current server domain (IPv4 or IPv6)
abuffer: array[128, byte] # Windows AcceptEx buffer
aovl: CustomOverlapped # AcceptEx OVERLAPPED structure
2018-05-16 08:22:34 +00:00
const SO_UPDATE_CONNECT_CONTEXT = 0x7010
template finishWriter(t: untyped) =
var vv = (t).queue.popFirst()
vv.writer.complete()
template zeroOvelappedOffset(t: untyped) =
(t).offset = 0
(t).offsetHigh = 0
template setOverlappedOffset(t, o: untyped) =
(t).offset = cast[int32](cast[uint64](o) and 0xFFFFFFFF'u64)
(t).offsetHigh = cast[int32](cast[uint64](o) shr 32)
2018-05-21 21:52:57 +00:00
template getFileSize(v: untyped): uint =
cast[uint]((v).buf)
2018-05-16 08:22:34 +00:00
2018-05-21 21:52:57 +00:00
template getFileHandle(v: untyped): Handle =
cast[Handle]((v).buflen)
2018-05-16 08:22:34 +00:00
template slideBuffer(t, o: untyped) =
2018-05-21 21:52:57 +00:00
(t).wwsabuf.buf = cast[cstring](cast[uint]((t).wwsabuf.buf) + uint(o))
(t).wwsabuf.len -= int32(o)
2018-05-16 08:22:34 +00:00
2018-05-21 21:52:57 +00:00
template setReaderWSABuffer(t: untyped) =
(t).rwsabuf.buf = cast[cstring](
2018-05-16 08:22:34 +00:00
cast[uint](addr t.buffer[0]) + uint((t).roffset))
2018-05-21 21:52:57 +00:00
(t).rwsabuf.len = int32(len((t).buffer) - (t).roffset)
2018-05-16 08:22:34 +00:00
2018-05-21 21:52:57 +00:00
template setWriterWSABuffer(t, v: untyped) =
(t).wwsabuf.buf = cast[cstring](v.buf)
(t).wwsabuf.len = cast[int32](v.buflen)
2018-05-16 08:22:34 +00:00
2018-05-22 16:27:20 +00:00
proc writeStreamLoop(udata: pointer) {.gcsafe, nimcall.} =
2018-05-16 08:22:34 +00:00
var bytesCount: int32
if isNil(udata):
return
2018-05-21 21:52:57 +00:00
var ovl = cast[PtrCustomOverlapped](udata)
2018-05-16 08:22:34 +00:00
var transp = cast[WindowsStreamTransport](ovl.data.udata)
while len(transp.queue) > 0:
if WritePending in transp.state:
## Continuation
transp.state.excl(WritePending)
let err = transp.wovl.data.errCode
if err == OSErrorCode(-1):
bytesCount = transp.wovl.data.bytesCount
var vector = transp.queue.popFirst()
if bytesCount == 0:
vector.writer.complete()
else:
if transp.kind == TransportKind.Socket:
if vector.kind == VectorKind.DataBuffer:
2018-05-21 21:52:57 +00:00
if bytesCount < transp.wwsabuf.len:
vector.shiftVectorBuffer(bytesCount)
2018-05-16 08:22:34 +00:00
transp.queue.addFirst(vector)
else:
vector.writer.complete()
else:
if uint(bytesCount) < getFileSize(vector):
2018-05-21 21:52:57 +00:00
vector.shiftVectorFile(bytesCount)
2018-05-16 08:22:34 +00:00
transp.queue.addFirst(vector)
else:
vector.writer.complete()
else:
transp.setWriteError(err)
transp.finishWriter()
else:
## Initiation
transp.state.incl(WritePending)
if transp.kind == TransportKind.Socket:
let sock = SocketHandle(transp.wovl.data.fd)
var vector = transp.queue.popFirst()
if vector.kind == VectorKind.DataBuffer:
2018-05-16 08:22:34 +00:00
transp.wovl.zeroOvelappedOffset()
2018-05-21 21:52:57 +00:00
transp.setWriterWSABuffer(vector)
let ret = WSASend(sock, addr transp.wwsabuf, 1,
2018-05-16 08:22:34 +00:00
addr bytesCount, DWORD(0),
cast[POVERLAPPED](addr transp.wovl), nil)
if ret != 0:
let err = osLastError()
if int(err) == ERROR_OPERATION_ABORTED:
2018-05-21 21:52:57 +00:00
transp.state.excl(WritePending)
transp.state.incl(WritePaused)
elif int(err) == ERROR_IO_PENDING:
transp.queue.addFirst(vector)
else:
2018-05-16 08:22:34 +00:00
transp.state.excl(WritePending)
transp.setWriteError(err)
2018-05-21 21:52:57 +00:00
vector.writer.complete()
else:
transp.queue.addFirst(vector)
2018-05-16 08:22:34 +00:00
else:
let loop = getGlobalDispatcher()
var size: int32
var flags: int32
if getFileSize(vector) > 2_147_483_646'u:
size = 2_147_483_646
else:
size = int32(getFileSize(vector))
transp.wovl.setOverlappedOffset(vector.offset)
var ret = loop.transmitFile(sock, getFileHandle(vector), size, 0,
cast[POVERLAPPED](addr transp.wovl),
nil, flags)
if ret == 0:
let err = osLastError()
if int(err) == ERROR_OPERATION_ABORTED:
2018-05-21 21:52:57 +00:00
transp.state.excl(WritePending)
transp.state.incl(WritePaused)
elif int(err) == ERROR_IO_PENDING:
transp.queue.addFirst(vector)
else:
2018-05-16 08:22:34 +00:00
transp.state.excl(WritePending)
transp.setWriteError(err)
2018-05-21 21:52:57 +00:00
vector.writer.complete()
else:
transp.queue.addFirst(vector)
2018-05-16 08:22:34 +00:00
break
if len(transp.queue) == 0:
transp.state.incl(WritePaused)
2018-05-22 16:27:20 +00:00
proc readStreamLoop(udata: pointer) {.gcsafe, nimcall.} =
2018-05-16 08:22:34 +00:00
if isNil(udata):
return
2018-05-21 21:52:57 +00:00
var ovl = cast[PtrCustomOverlapped](udata)
2018-05-16 08:22:34 +00:00
var transp = cast[WindowsStreamTransport](ovl.data.udata)
while true:
if ReadPending in transp.state:
## Continuation
2018-05-21 21:52:57 +00:00
transp.state.excl(ReadPending)
2018-05-16 08:22:34 +00:00
if ReadClosed in transp.state:
break
let err = transp.rovl.data.errCode
if err == OSErrorCode(-1):
let bytesCount = transp.rovl.data.bytesCount
2018-05-21 21:52:57 +00:00
transp.transferred += bytesCount
2018-05-16 08:22:34 +00:00
if bytesCount == 0:
transp.state.incl(ReadEof)
transp.state.incl(ReadPaused)
else:
if transp.offset != transp.roffset:
moveMem(addr transp.buffer[transp.offset],
addr transp.buffer[transp.roffset],
bytesCount)
transp.offset += bytesCount
transp.roffset = transp.offset
if transp.offset == len(transp.buffer):
transp.state.incl(ReadPaused)
else:
transp.setReadError(err)
if not isNil(transp.reader):
transp.finishReader()
2018-05-21 21:52:57 +00:00
if ReadPaused in transp.state:
# Transport buffer is full, so we will not continue on reading.
break
2018-05-16 08:22:34 +00:00
else:
## Initiation
2018-05-21 21:52:57 +00:00
if transp.state * {ReadEof, ReadClosed, ReadError} == {}:
2018-05-16 08:22:34 +00:00
var flags = DWORD(0)
var bytesCount: int32 = 0
transp.state.excl(ReadPaused)
transp.state.incl(ReadPending)
if transp.kind == TransportKind.Socket:
let sock = SocketHandle(transp.rovl.data.fd)
2018-05-21 21:52:57 +00:00
transp.roffset = transp.offset
transp.setReaderWSABuffer()
let ret = WSARecv(sock, addr transp.rwsabuf, 1,
2018-05-16 08:22:34 +00:00
addr bytesCount, addr flags,
cast[POVERLAPPED](addr transp.rovl), nil)
if ret != 0:
let err = osLastError()
if int(err) == ERROR_OPERATION_ABORTED:
2018-05-21 21:52:57 +00:00
transp.state.excl(ReadPending)
2018-05-16 08:22:34 +00:00
transp.state.incl(ReadPaused)
elif int32(err) != ERROR_IO_PENDING:
transp.setReadError(err)
if not isNil(transp.reader):
transp.finishReader()
## Finish Loop
break
proc newStreamSocketTransport(sock: AsyncFD, bufsize: int): StreamTransport =
2018-05-21 21:52:57 +00:00
var transp = WindowsStreamTransport(kind: TransportKind.Socket)
transp.fd = sock
transp.rovl.data = CompletionData(fd: sock, cb: readStreamLoop,
udata: cast[pointer](transp))
transp.wovl.data = CompletionData(fd: sock, cb: writeStreamLoop,
udata: cast[pointer](transp))
transp.buffer = newSeq[byte](bufsize)
transp.state = {ReadPaused, WritePaused}
transp.queue = initDeque[StreamVector]()
transp.future = newFuture[void]("stream.socket.transport")
GC_ref(transp)
result = cast[StreamTransport](transp)
2018-05-16 08:22:34 +00:00
proc bindToDomain(handle: AsyncFD, domain: Domain): bool =
result = true
if domain == Domain.AF_INET6:
var saddr: Sockaddr_in6
saddr.sin6_family = int16(toInt(domain))
if bindAddr(SocketHandle(handle), cast[ptr SockAddr](addr(saddr)),
sizeof(saddr).SockLen) != 0'i32:
result = false
else:
var saddr: Sockaddr_in
saddr.sin_family = int16(toInt(domain))
if bindAddr(SocketHandle(handle), cast[ptr SockAddr](addr(saddr)),
sizeof(saddr).SockLen) != 0'i32:
result = false
proc connect*(address: TransportAddress,
bufferSize = DefaultStreamBufferSize): Future[StreamTransport] =
let loop = getGlobalDispatcher()
var
saddr: Sockaddr_storage
slen: SockLen
sock: AsyncFD
povl: RefCustomOverlapped
var retFuture = newFuture[StreamTransport]("stream.transport.connect")
toSockAddr(address.address, address.port, saddr, slen)
sock = createAsyncSocket(address.address.getDomain(), SockType.SOCK_STREAM,
Protocol.IPPROTO_TCP)
if sock == asyncInvalidSocket:
result.fail(newException(OSError, osErrorMsg(osLastError())))
if not bindToDomain(sock, address.address.getDomain()):
sock.closeAsyncSocket()
result.fail(newException(OSError, osErrorMsg(osLastError())))
proc continuation(udata: pointer) =
2018-05-22 16:27:20 +00:00
var ovl = cast[RefCustomOverlapped](udata)
2018-05-16 08:22:34 +00:00
if not retFuture.finished:
if ovl.data.errCode == OSErrorCode(-1):
if setsockopt(SocketHandle(sock), cint(SOL_SOCKET),
cint(SO_UPDATE_CONNECT_CONTEXT), nil,
SockLen(0)) != 0'i32:
sock.closeAsyncSocket()
retFuture.fail(newException(OSError, osErrorMsg(osLastError())))
else:
retFuture.complete(newStreamSocketTransport(povl.data.fd,
bufferSize))
else:
sock.closeAsyncSocket()
retFuture.fail(newException(OSError, osErrorMsg(ovl.data.errCode)))
2018-05-22 16:27:20 +00:00
GC_unref(ovl)
2018-05-16 08:22:34 +00:00
povl = RefCustomOverlapped()
2018-05-22 16:27:20 +00:00
GC_ref(povl)
2018-05-16 08:22:34 +00:00
povl.data = CompletionData(fd: sock, cb: continuation)
var res = loop.connectEx(SocketHandle(sock),
cast[ptr SockAddr](addr saddr),
DWORD(slen), nil, 0, nil,
cast[POVERLAPPED](povl))
# We will not process immediate completion, to avoid undefined behavior.
if not res:
let err = osLastError()
if int32(err) != ERROR_IO_PENDING:
2018-05-22 16:27:20 +00:00
GC_unref(povl)
2018-05-16 08:22:34 +00:00
sock.closeAsyncSocket()
retFuture.fail(newException(OSError, osErrorMsg(err)))
return retFuture
proc acceptAddr(server: WindowsStreamServer): Future[AsyncFD] =
var retFuture = newFuture[AsyncFD]("transport.acceptAddr")
let loop = getGlobalDispatcher()
var sock = createAsyncSocket(server.domain, SockType.SOCK_STREAM,
Protocol.IPPROTO_TCP)
if sock == asyncInvalidSocket:
retFuture.fail(newException(OSError, osErrorMsg(osLastError())))
var dwBytesReceived = DWORD(0)
let dwReceiveDataLength = DWORD(0)
let dwLocalAddressLength = DWORD(sizeof(Sockaddr_in6) + 16)
let dwRemoteAddressLength = DWORD(sizeof(Sockaddr_in6) + 16)
proc continuation(udata: pointer) =
2018-05-21 21:52:57 +00:00
var ovl = cast[PtrCustomOverlapped](udata)
2018-05-16 08:22:34 +00:00
if not retFuture.finished:
if server.server.status in {Stopped, Paused}:
sock.closeAsyncSocket()
retFuture.complete(asyncInvalidSocket)
else:
if ovl.data.errCode == OSErrorCode(-1):
if setsockopt(SocketHandle(sock), cint(SOL_SOCKET),
cint(SO_UPDATE_ACCEPT_CONTEXT),
addr server.server.sock,
SockLen(sizeof(SocketHandle))) != 0'i32:
sock.closeAsyncSocket()
retFuture.fail(newException(OSError, osErrorMsg(osLastError())))
else:
retFuture.complete(sock)
else:
sock.closeAsyncSocket()
retFuture.fail(newException(OSError, osErrorMsg(ovl.data.errCode)))
server.aovl.data.fd = server.server.sock
server.aovl.data.cb = continuation
let res = loop.acceptEx(SocketHandle(server.server.sock),
SocketHandle(sock), addr server.abuffer[0],
dwReceiveDataLength, dwLocalAddressLength,
dwRemoteAddressLength, addr dwBytesReceived,
cast[POVERLAPPED](addr server.aovl))
if not res:
let err = osLastError()
if int32(err) != ERROR_IO_PENDING:
retFuture.fail(newException(OSError, osErrorMsg(err)))
return retFuture
proc serverLoop(server: StreamServer): Future[void] {.async.} =
## TODO: This procedure must be reviewed, when cancellation support
## will be added
var wserver = new WindowsStreamServer
wserver.server = server
wserver.domain = server.local.address.getDomain()
await server.actEvent.wait()
server.actEvent.clear()
var acceptFut: Future[AsyncFD]
2018-05-16 08:22:34 +00:00
if server.action == ServerCommand.Start:
var eventFut = server.actEvent.wait()
while true:
if server.status in {Paused}:
await eventFut
else:
acceptFut = acceptAddr(wserver)
await eventFut or acceptFut
2018-05-16 08:22:34 +00:00
if eventFut.finished:
server.actEvent.clear()
eventFut = server.actEvent.wait()
2018-05-16 08:22:34 +00:00
if server.action == ServerCommand.Start:
if server.status in {Stopped, Paused}:
server.status = Running
elif server.action == ServerCommand.Stop:
if server.status in {Running}:
server.status = Stopped
break
elif server.status in {Paused}:
server.status = Stopped
break
elif server.action == ServerCommand.Pause:
if server.status in {Running}:
server.status = Paused
2018-05-16 08:22:34 +00:00
if acceptFut.finished:
if not acceptFut.failed:
var sock = acceptFut.read()
if sock != asyncInvalidSocket:
2018-05-22 22:29:07 +00:00
discard server.function(server,
2018-05-16 08:22:34 +00:00
newStreamSocketTransport(sock, server.bufferSize),
server.udata)
proc resumeRead(transp: StreamTransport) {.inline.} =
var wtransp = cast[WindowsStreamTransport](transp)
wtransp.state.excl(ReadPaused)
readStreamLoop(cast[pointer](addr wtransp.rovl))
proc resumeWrite(transp: StreamTransport) {.inline.} =
var wtransp = cast[WindowsStreamTransport](transp)
wtransp.state.excl(WritePaused)
writeStreamLoop(cast[pointer](addr wtransp.wovl))
else:
import posix
type
UnixStreamTransport* = ref object of StreamTransport
template getVectorBuffer(v: untyped): pointer =
cast[pointer](cast[uint]((v).buf) + uint((v).boffset))
template getVectorLength(v: untyped): int =
cast[int]((v).buflen - int((v).boffset))
template initBufferStreamVector(v, p, n, t: untyped) =
(v).kind = DataBuffer
(v).buf = cast[pointer]((p))
(v).buflen = int(n)
(v).writer = (t)
proc writeStreamLoop(udata: pointer) {.gcsafe.} =
var cdata = cast[ptr CompletionData](udata)
var transp = cast[UnixStreamTransport](cdata.udata)
let fd = SocketHandle(cdata.fd)
if not isNil(transp):
if len(transp.queue) > 0:
var vector = transp.queue.popFirst()
while true:
if transp.kind == TransportKind.Socket:
if vector.kind == VectorKind.DataBuffer:
let res = posix.send(fd, vector.buf, vector.buflen, MSG_NOSIGNAL)
if res >= 0:
if vector.buflen - res == 0:
vector.writer.complete()
else:
vector.shiftVectorBuffer(res)
transp.queue.addFirst(vector)
else:
let err = osLastError()
if int(err) == EINTR:
continue
else:
transp.setWriteError(err)
vector.writer.complete()
else:
2018-05-21 21:52:57 +00:00
let res = sendfile(int(fd), cast[int](vector.buflen),
int(vector.offset),
cast[int](vector.buf))
if res >= 0:
if cast[int](vector.buf) - res == 0:
vector.writer.complete()
else:
vector.shiftVectorFile(res)
transp.queue.addFirst(vector)
else:
let err = osLastError()
if int(err) == EINTR:
continue
else:
transp.setWriteError(err)
vector.writer.complete()
break
2018-05-16 08:22:34 +00:00
else:
transp.state.incl(WritePaused)
transp.fd.removeWriter()
proc readStreamLoop(udata: pointer) {.gcsafe.} =
var cdata = cast[ptr CompletionData](udata)
var transp = cast[UnixStreamTransport](cdata.udata)
let fd = SocketHandle(cdata.fd)
if not isNil(transp):
while true:
var res = posix.recv(fd, addr transp.buffer[transp.offset],
len(transp.buffer) - transp.offset, cint(0))
if res < 0:
let err = osLastError()
if int(err) == EINTR:
continue
elif int(err) in {ECONNRESET}:
transp.state.incl(ReadEof)
transp.state.incl(ReadPaused)
cdata.fd.removeReader()
else:
transp.setReadError(err)
cdata.fd.removeReader()
elif res == 0:
transp.state.incl(ReadEof)
transp.state.incl(ReadPaused)
cdata.fd.removeReader()
else:
transp.offset += res
if transp.offset == len(transp.buffer):
transp.state.incl(ReadPaused)
cdata.fd.removeReader()
if not isNil(transp.reader):
transp.finishReader()
break
proc newStreamSocketTransport(sock: AsyncFD, bufsize: int): StreamTransport =
2018-05-21 21:52:57 +00:00
var transp = UnixStreamTransport(kind: TransportKind.Socket)
transp.fd = sock
transp.buffer = newSeq[byte](bufsize)
transp.state = {ReadPaused, WritePaused}
transp.queue = initDeque[StreamVector]()
transp.future = newFuture[void]("socket.stream.transport")
GC_ref(transp)
result = cast[StreamTransport](transp)
2018-05-16 08:22:34 +00:00
proc connect*(address: TransportAddress,
bufferSize = DefaultStreamBufferSize): Future[StreamTransport] =
2018-05-22 21:03:13 +00:00
## Connect to ``address`` and create new transport for this connection.
2018-05-16 08:22:34 +00:00
var
saddr: Sockaddr_storage
slen: SockLen
sock: AsyncFD
var retFuture = newFuture[StreamTransport]("transport.connect")
toSockAddr(address.address, address.port, saddr, slen)
sock = createAsyncSocket(address.address.getDomain(), SockType.SOCK_STREAM,
Protocol.IPPROTO_TCP)
if sock == asyncInvalidSocket:
result.fail(newException(OSError, osErrorMsg(osLastError())))
proc continuation(udata: pointer) =
var data = cast[ptr CompletionData](udata)
var err = 0
if not data.fd.getSocketError(err):
sock.closeAsyncSocket()
retFuture.fail(newException(OSError, osErrorMsg(osLastError())))
return
if err != 0:
sock.closeAsyncSocket()
retFuture.fail(newException(OSError, osErrorMsg(OSErrorCode(err))))
return
data.fd.removeWriter()
retFuture.complete(newStreamSocketTransport(data.fd, bufferSize))
while true:
var res = posix.connect(SocketHandle(sock),
cast[ptr SockAddr](addr saddr), slen)
if res == 0:
retFuture.complete(newStreamSocketTransport(sock, bufferSize))
break
else:
let err = osLastError()
if int(err) == EINTR:
continue
elif int(err) == EINPROGRESS:
sock.addWriter(continuation)
break
else:
sock.closeAsyncSocket()
retFuture.fail(newException(OSError, osErrorMsg(err)))
break
return retFuture
proc serverCallback(udata: pointer) =
var
saddr: Sockaddr_storage
slen: SockLen
var server = cast[StreamServer](cast[ptr CompletionData](udata).udata)
while true:
let res = posix.accept(SocketHandle(server.sock),
cast[ptr SockAddr](addr saddr), addr slen)
if int(res) > 0:
let sock = wrapAsyncSocket(res)
if sock != asyncInvalidSocket:
2018-05-22 22:29:07 +00:00
discard server.function(server,
2018-05-16 08:22:34 +00:00
newStreamSocketTransport(sock, server.bufferSize),
server.udata)
break
else:
let err = osLastError()
if int(err) == EINTR:
continue
elif int(err) in {EBADF, EINVAL, ENOTSOCK, EOPNOTSUPP, EPROTO}:
## Critical unrecoverable error
raiseOsError(err)
proc serverLoop(server: SocketServer): Future[void] {.async.} =
while true:
await server.actEvent.wait()
server.actEvent.clear()
if server.action == ServerCommand.Start:
if server.status in {Stopped, Paused, Starting}:
addReader(server.sock, serverCallback,
cast[pointer](server))
server.status = Running
elif server.action == ServerCommand.Stop:
if server.status in {Running}:
removeReader(server.sock)
server.status = Stopped
break
elif server.status in {Paused}:
server.status = Stopped
break
elif server.action == ServerCommand.Pause:
if server.status in {Running}:
removeReader(server.sock)
server.status = Paused
proc resumeRead(transp: StreamTransport) {.inline.} =
transp.state.excl(ReadPaused)
addReader(transp.fd, readStreamLoop, cast[pointer](transp))
proc resumeWrite(transp: StreamTransport) {.inline.} =
transp.state.excl(WritePaused)
addWriter(transp.fd, writeStreamLoop, cast[pointer](transp))
proc start*(server: SocketServer) =
2018-05-21 21:52:57 +00:00
## Starts ``server``.
2018-05-16 08:22:34 +00:00
server.action = Start
server.actEvent.fire()
proc stop*(server: SocketServer) =
2018-05-22 21:03:13 +00:00
## Stops ``server``.
2018-05-16 08:22:34 +00:00
server.action = Stop
server.actEvent.fire()
proc pause*(server: SocketServer) =
2018-05-22 21:03:13 +00:00
## Pause ``server``.
2018-05-16 08:22:34 +00:00
server.action = Pause
server.actEvent.fire()
proc join*(server: SocketServer) {.async.} =
2018-05-21 21:52:57 +00:00
## Waits until ``server`` is not stopped.
if not server.loopFuture.finished:
await server.loopFuture
proc close*(server: SocketServer) =
## Release ``server`` resources.
GC_unref(server)
2018-05-16 08:22:34 +00:00
proc createStreamServer*(host: TransportAddress,
flags: set[ServerFlags],
cbproc: StreamCallback,
sock: AsyncFD = asyncInvalidSocket,
backlog: int = 100,
bufferSize: int = DefaultStreamBufferSize,
udata: pointer = nil): StreamServer =
2018-05-22 21:03:13 +00:00
## Create new TCP server
2018-05-16 08:22:34 +00:00
var
saddr: Sockaddr_storage
slen: SockLen
serverSocket: AsyncFD
if sock == asyncInvalidSocket:
serverSocket = createAsyncSocket(host.address.getDomain(),
SockType.SOCK_STREAM,
Protocol.IPPROTO_TCP)
if serverSocket == asyncInvalidSocket:
raiseOsError(osLastError())
else:
if not setSocketBlocking(SocketHandle(sock), false):
raiseOsError(osLastError())
register(sock)
serverSocket = sock
if ServerFlags.ReuseAddr in flags:
if not setSockOpt(serverSocket, SOL_SOCKET, SO_REUSEADDR, 1):
let err = osLastError()
if sock == asyncInvalidSocket:
closeAsyncSocket(serverSocket)
raiseOsError(err)
toSockAddr(host.address, host.port, saddr, slen)
if bindAddr(SocketHandle(serverSocket), cast[ptr SockAddr](addr saddr),
slen) != 0:
let err = osLastError()
if sock == asyncInvalidSocket:
closeAsyncSocket(serverSocket)
raiseOsError(err)
2018-05-21 21:52:57 +00:00
2018-05-16 08:22:34 +00:00
if nativesockets.listen(SocketHandle(serverSocket), cint(backlog)) != 0:
let err = osLastError()
if sock == asyncInvalidSocket:
closeAsyncSocket(serverSocket)
raiseOsError(err)
result = StreamServer()
result.sock = serverSocket
result.function = cbproc
result.bufferSize = bufferSize
result.status = Starting
result.actEvent = newAsyncEvent()
result.udata = udata
result.local = host
2018-05-21 21:52:57 +00:00
GC_ref(result)
2018-05-16 08:22:34 +00:00
result.loopFuture = serverLoop(result)
proc write*(transp: StreamTransport, pbytes: pointer,
nbytes: int): Future[int] {.async.} =
2018-05-22 21:03:13 +00:00
## Write data from buffer ``pbytes`` with size ``nbytes`` to transport
## ``transp``.
2018-05-16 08:22:34 +00:00
checkClosed(transp)
var waitFuture = newFuture[void]("transport.write")
2018-05-21 21:52:57 +00:00
var vector = StreamVector(kind: DataBuffer, writer: waitFuture,
buf: pbytes, buflen: nbytes)
2018-05-16 08:22:34 +00:00
transp.queue.addLast(vector)
if WritePaused in transp.state:
transp.resumeWrite()
await vector.writer
if WriteError in transp.state:
raise transp.getError()
result = nbytes
2018-05-21 21:52:57 +00:00
proc writeFile*(transp: StreamTransport, handle: int,
offset: uint = 0,
size: int = 0): Future[void] {.async.} =
2018-05-22 21:03:13 +00:00
## Write data from file descriptor ``handle`` to transport ``transp``.
##
## You can specify starting ``offset`` in opened file and number of bytes
## to transfer from file to transport via ``size``.
2018-05-21 21:52:57 +00:00
if transp.kind != TransportKind.Socket:
raise newException(TransportError, "You can transmit files only to sockets")
checkClosed(transp)
var waitFuture = newFuture[void]("transport.writeFile")
var vector = StreamVector(kind: DataFile, writer: waitFuture,
buf: cast[pointer](size), offset: offset,
buflen: handle)
transp.queue.addLast(vector)
if WritePaused in transp.state:
transp.resumeWrite()
await vector.writer
if WriteError in transp.state:
raise transp.getError()
2018-05-16 08:22:34 +00:00
proc readExactly*(transp: StreamTransport, pbytes: pointer,
2018-05-21 21:52:57 +00:00
nbytes: int) {.async.} =
2018-05-22 21:03:13 +00:00
## Read exactly ``nbytes`` bytes from transport ``transp`` and store it to
## ``pbytes``.
##
## If EOF is received and ``nbytes`` is not yet readed, the procedure
## will raise ``TransportIncompleteError``.
2018-05-16 08:22:34 +00:00
checkClosed(transp)
checkPending(transp)
var index = 0
while true:
if transp.offset == 0:
if (ReadError in transp.state):
raise transp.getError()
if (ReadEof in transp.state) or (ReadClosed in transp.state):
raise newException(TransportIncompleteError, "Data incomplete!")
if transp.offset >= (nbytes - index):
copyMem(cast[pointer](cast[uint](pbytes) + uint(index)),
addr(transp.buffer[0]), nbytes - index)
transp.shiftBuffer(nbytes - index)
break
else:
2018-05-21 21:52:57 +00:00
if transp.offset != 0:
copyMem(cast[pointer](cast[uint](pbytes) + uint(index)),
addr(transp.buffer[0]), transp.offset)
index += transp.offset
transp.reader = newFuture[void]("stream.transport.readExactly")
2018-05-16 08:22:34 +00:00
transp.offset = 0
if ReadPaused in transp.state:
transp.resumeRead()
await transp.reader
2018-05-21 21:52:57 +00:00
2018-05-16 08:22:34 +00:00
# we are no longer need data
transp.reader = nil
proc readOnce*(transp: StreamTransport, pbytes: pointer,
nbytes: int): Future[int] {.async.} =
## Perform one read operation on transport ``transp``.
checkClosed(transp)
checkPending(transp)
while true:
if transp.offset == 0:
if (ReadError in transp.state):
raise transp.getError()
if (ReadEof in transp.state) or (ReadClosed in transp.state):
result = 0
break
2018-05-21 21:52:57 +00:00
transp.reader = newFuture[void]("stream.transport.readOnce")
2018-05-16 08:22:34 +00:00
if ReadPaused in transp.state:
transp.resumeRead()
await transp.reader
transp.reader = nil
else:
if transp.offset > nbytes:
copyMem(pbytes, addr(transp.buffer[0]), nbytes)
transp.shiftBuffer(nbytes)
result = nbytes
else:
copyMem(pbytes, addr(transp.buffer[0]), transp.offset)
result = transp.offset
break
proc readUntil*(transp: StreamTransport, pbytes: pointer, nbytes: int,
sep: seq[byte]): Future[int] {.async.} =
2018-05-22 21:03:13 +00:00
## Read data from the transport ``transp`` until separator ``sep`` is found.
##
## On success, the data and separator will be removed from the internal
## buffer (consumed). Returned data will NOT include the separator at the end.
2018-05-16 08:22:34 +00:00
checkClosed(transp)
checkPending(transp)
var dest = cast[ptr UncheckedArray[byte]](pbytes)
var state = 0
var k = 0
var index = 0
while true:
if (transp.offset - index) == 0:
if ReadError in transp.state:
transp.shiftBuffer(index)
raise transp.getError()
if (ReadEof in transp.state) or (ReadClosed in transp.state):
transp.shiftBuffer(index)
raise newException(TransportIncompleteError, "Data incomplete!")
index = 0
while index < transp.offset:
let ch = transp.buffer[index]
if sep[state] == ch:
inc(state)
else:
state = 0
if k < nbytes:
dest[k] = ch
inc(k)
else:
raise newException(TransportLimitError, "Limit reached!")
if state == len(sep):
transp.shiftBuffer(index + 1)
break
inc(index)
if state == len(sep):
result = k
break
else:
if (transp.offset - index) == 0:
2018-05-21 21:52:57 +00:00
transp.reader = newFuture[void]("stream.transport.readUntil")
2018-05-16 08:22:34 +00:00
if ReadPaused in transp.state:
transp.resumeRead()
await transp.reader
# we are no longer need data
transp.reader = nil
proc readLine*(transp: StreamTransport, limit = 0,
sep = "\r\n"): Future[string] {.async.} =
2018-05-22 21:03:13 +00:00
## Read one line from transport ``transp``, where "line" is a sequence of
## bytes ending with ``sep`` (default is "\r\n").
##
## If EOF is received, and ``sep`` was not found, the method will return the
## partial read bytes.
##
## If the EOF was received and the internal buffer is empty, return an
## empty string.
##
## If ``limit`` more then 0, then read is limited to ``limit`` bytes.
2018-05-16 08:22:34 +00:00
checkClosed(transp)
checkPending(transp)
result = ""
var lim = if limit <= 0: -1 else: limit
var state = 0
var index = 0
while true:
if (transp.offset - index) == 0:
if (ReadError in transp.state):
transp.shiftBuffer(index)
raise transp.getError()
if (ReadEof in transp.state) or (ReadClosed in transp.state):
transp.shiftBuffer(index)
break
index = 0
while index < transp.offset:
let ch = char(transp.buffer[index])
if sep[state] == ch:
inc(state)
if state == len(sep):
transp.shiftBuffer(index + 1)
break
else:
state = 0
result.add(ch)
if len(result) == lim:
transp.shiftBuffer(index + 1)
break
inc(index)
if (state == len(sep)) or (lim == len(result)):
break
else:
if (transp.offset - index) == 0:
2018-05-21 21:52:57 +00:00
transp.reader = newFuture[void]("stream.transport.readLine")
2018-05-16 08:22:34 +00:00
if ReadPaused in transp.state:
transp.resumeRead()
await transp.reader
# we are no longer need data
transp.reader = nil
proc read*(transp: StreamTransport, n = -1): Future[seq[byte]] {.async.} =
checkClosed(transp)
checkPending(transp)
result = newSeq[byte]()
while true:
if (ReadError in transp.state):
raise transp.getError()
if (ReadEof in transp.state) or (ReadClosed in transp.state):
break
if transp.offset > 0:
let s = len(result)
let o = s + transp.offset
if n == -1:
# grabbing all incoming data, until EOF
result.setLen(o)
copyMem(cast[pointer](addr result[s]), addr(transp.buffer[0]),
transp.offset)
transp.offset = 0
else:
if transp.offset >= (n - s):
# size of buffer data is more then we need, grabbing only part
let part = transp.offset - (n - s)
result.setLen(n)
copyMem(cast[pointer](addr result[s]), addr(transp.buffer[0]),
part)
transp.shiftBuffer(part)
break
else:
# there not enough data in buffer, grabbing all
result.setLen(o)
copyMem(cast[pointer](addr result[s]), addr(transp.buffer[0]),
transp.offset)
transp.offset = 0
2018-05-21 21:52:57 +00:00
transp.reader = newFuture[void]("stream.transport.read")
2018-05-16 08:22:34 +00:00
if ReadPaused in transp.state:
transp.resumeRead()
await transp.reader
# we are no longer need data
transp.reader = nil
proc atEof*(transp: StreamTransport): bool {.inline.} =
## Returns ``true`` if ``transp`` is at EOF.
result = (transp.offset == 0) and (ReadEof in transp.state) and
(ReadPaused in transp.state)
proc join*(transp: StreamTransport) {.async.} =
## Wait until ``transp`` will not be closed.
2018-05-21 21:52:57 +00:00
if not transp.future.finished:
await transp.future
2018-05-16 08:22:34 +00:00
proc close*(transp: StreamTransport) =
## Closes and frees resources of transport ``transp``.
if ReadClosed notin transp.state and WriteClosed notin transp.state:
when defined(windows):
discard cancelIo(Handle(transp.fd))
closeAsyncSocket(transp.fd)
transp.state.incl(WriteClosed)
transp.state.incl(ReadClosed)
transp.future.complete()
2018-05-21 21:52:57 +00:00
GC_unref(transp)