codex-storage
/
nim-chronos
mirror of https://github.com/codex-storage/nim-chronos.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

deprecate `callback=`, UDP fixes (fixes #491) (#492) 

Using the callback setter may lead to callbacks owned by others being
reset, which is unexpected.

* don't crash on zero-length UDP writes
This commit is contained in:
Jacek Sieka 2024-01-19 09:21:10 +01:00 committed by GitHub
parent 1021a7d294
commit 3ca2c5e6b5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 25 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
chronos/internal/asyncfutures.nim
View File

@ -330,7 +330,8 @@ proc removeCallback*(future: FutureBase, cb: CallbackFunc,
proc removeCallback*(future: FutureBase, cb: CallbackFunc) = proc removeCallback*(future: FutureBase, cb: CallbackFunc) =
future.removeCallback(cb, cast[pointer](future)) future.removeCallback(cb, cast[pointer](future))
proc `callback=`*(future: FutureBase, cb: CallbackFunc, udata: pointer) = proc `callback=`*(future: FutureBase, cb: CallbackFunc, udata: pointer) {.
deprecated: "use addCallback/removeCallback/clearCallbacks to manage the callback list".} =
## Clears the list of callbacks and sets the callback proc to be called when ## Clears the list of callbacks and sets the callback proc to be called when
## the future completes. ## the future completes.
## ##
@ -341,11 +342,14 @@ proc `callback=`*(future: FutureBase, cb: CallbackFunc, udata: pointer) =
future.clearCallbacks future.clearCallbacks
future.addCallback(cb, udata) future.addCallback(cb, udata)
proc `callback=`*(future: FutureBase, cb: CallbackFunc) = proc `callback=`*(future: FutureBase, cb: CallbackFunc) {.
deprecated: "use addCallback/removeCallback/clearCallbacks instead to manage the callback list".} =
## Sets the callback proc to be called when the future completes. ## Sets the callback proc to be called when the future completes.
## ##
## If future has already completed then ``cb`` will be called immediately. ## If future has already completed then ``cb`` will be called immediately.
{.push warning[Deprecated]: off.}
`callback=`(future, cb, cast[pointer](future)) `callback=`(future, cb, cast[pointer](future))
{.pop.}
proc `cancelCallback=`*(future: FutureBase, cb: CallbackFunc) = proc `cancelCallback=`*(future: FutureBase, cb: CallbackFunc) =
## Sets the callback procedure to be called when the future is cancelled. ## Sets the callback procedure to be called when the future is cancelled.

38
chronos/transports/datagram.nim
View File

@ -13,6 +13,7 @@ import std/deques
when not(defined(windows)): import ".."/selectors2 when not(defined(windows)): import ".."/selectors2
import ".."/[asyncloop, config, osdefs, oserrno, osutils, handles] import ".."/[asyncloop, config, osdefs, oserrno, osutils, handles]
import "."/common import "."/common
import stew/ptrops
type type
VectorKind = enum VectorKind = enum
@ -119,7 +120,7 @@ when defined(windows):
## Initiation ## Initiation
transp.state.incl(WritePending) transp.state.incl(WritePending)
let fd = SocketHandle(transp.fd) let fd = SocketHandle(transp.fd)
var vector = transp.queue.popFirst() let vector = transp.queue.popFirst()
transp.setWriterWSABuffer(vector) transp.setWriterWSABuffer(vector)
let ret = let ret =
if vector.kind == WithAddress: if vector.kind == WithAddress:
@ -365,7 +366,7 @@ when defined(windows):
udata: cast[pointer](res)) udata: cast[pointer](res))
res.wovl.data = CompletionData(cb: writeDatagramLoop, res.wovl.data = CompletionData(cb: writeDatagramLoop,
udata: cast[pointer](res)) udata: cast[pointer](res))
res.rwsabuf = WSABUF(buf: cast[cstring](addr res.buffer[0]), res.rwsabuf = WSABUF(buf: cast[cstring](baseAddr res.buffer),
len: ULONG(len(res.buffer))) len: ULONG(len(res.buffer)))
GC_ref(res) GC_ref(res)
# Start tracking transport # Start tracking transport
@ -392,7 +393,7 @@ else:
else: else:
while true: while true:
transp.ralen = SockLen(sizeof(Sockaddr_storage)) transp.ralen = SockLen(sizeof(Sockaddr_storage))
var res = osdefs.recvfrom(fd, addr transp.buffer[0], var res = osdefs.recvfrom(fd, baseAddr transp.buffer,
cint(len(transp.buffer)), cint(0), cint(len(transp.buffer)), cint(0),
cast[ptr SockAddr](addr transp.raddr), cast[ptr SockAddr](addr transp.raddr),
addr transp.ralen) addr transp.ralen)
@ -424,7 +425,7 @@ else:
transp.state.incl({WritePaused}) transp.state.incl({WritePaused})
else: else:
if len(transp.queue) > 0: if len(transp.queue) > 0:
var vector = transp.queue.popFirst() let vector = transp.queue.popFirst()
while true: while true:
if vector.kind == WithAddress: if vector.kind == WithAddress:
toSAddr(vector.address, transp.waddr, transp.walen) toSAddr(vector.address, transp.waddr, transp.walen)
@ -826,7 +827,7 @@ proc newDatagramTransport6*[T](cbproc: UnsafeDatagramCallback,
proc join*(transp: DatagramTransport): Future[void] {. proc join*(transp: DatagramTransport): Future[void] {.
async: (raw: true, raises: [CancelledError]).} = async: (raw: true, raises: [CancelledError]).} =
## Wait until the transport ``transp`` will be closed. ## Wait until the transport ``transp`` will be closed.
var retFuture = newFuture[void]("datagram.transport.join") let retFuture = newFuture[void]("datagram.transport.join")
proc continuation(udata: pointer) {.gcsafe.} = proc continuation(udata: pointer) {.gcsafe.} =
retFuture.complete() retFuture.complete()
@ -858,12 +859,12 @@ proc send*(transp: DatagramTransport, pbytes: pointer,
async: (raw: true, raises: [TransportError, CancelledError]).} = async: (raw: true, raises: [TransportError, CancelledError]).} =
## Send buffer with pointer ``pbytes`` and size ``nbytes`` using transport ## Send buffer with pointer ``pbytes`` and size ``nbytes`` using transport
## ``transp`` to remote destination address which was bounded on transport. ## ``transp`` to remote destination address which was bounded on transport.
var retFuture = newFuture[void]("datagram.transport.send(pointer)") let retFuture = newFuture[void]("datagram.transport.send(pointer)")
transp.checkClosed(retFuture) transp.checkClosed(retFuture)
if transp.remote.port == Port(0): if transp.remote.port == Port(0):
retFuture.fail(newException(TransportError, "Remote peer not set!")) retFuture.fail(newException(TransportError, "Remote peer not set!"))
return retFuture return retFuture
var vector = GramVector(kind: WithoutAddress, buf: pbytes, buflen: nbytes, let vector = GramVector(kind: WithoutAddress, buf: pbytes, buflen: nbytes,
writer: retFuture) writer: retFuture)
transp.queue.addLast(vector) transp.queue.addLast(vector)
if WritePaused in transp.state: if WritePaused in transp.state:
@ -877,14 +878,14 @@ proc send*(transp: DatagramTransport, msg: sink string,
async: (raw: true, raises: [TransportError, CancelledError]).} = async: (raw: true, raises: [TransportError, CancelledError]).} =
## Send string ``msg`` using transport ``transp`` to remote destination ## Send string ``msg`` using transport ``transp`` to remote destination
## address which was bounded on transport. ## address which was bounded on transport.
var retFuture = newFuture[void]("datagram.transport.send(string)") let retFuture = newFuture[void]("datagram.transport.send(string)")
transp.checkClosed(retFuture) transp.checkClosed(retFuture)
let length = if msglen <= 0: len(msg) else: msglen let length = if msglen <= 0: len(msg) else: msglen
var localCopy = chronosMoveSink(msg) var localCopy = chronosMoveSink(msg)
retFuture.addCallback(proc(_: pointer) = reset(localCopy)) retFuture.addCallback(proc(_: pointer) = reset(localCopy))
let vector = GramVector(kind: WithoutAddress, buf: addr localCopy[0], let vector = GramVector(kind: WithoutAddress, buf: baseAddr localCopy,
buflen: length, buflen: length,
writer: retFuture) writer: retFuture)
@ -900,14 +901,14 @@ proc send*[T](transp: DatagramTransport, msg: sink seq[T],
async: (raw: true, raises: [TransportError, CancelledError]).} = async: (raw: true, raises: [TransportError, CancelledError]).} =
## Send string ``msg`` using transport ``transp`` to remote destination ## Send string ``msg`` using transport ``transp`` to remote destination
## address which was bounded on transport. ## address which was bounded on transport.
var retFuture = newFuture[void]("datagram.transport.send(seq)") let retFuture = newFuture[void]("datagram.transport.send(seq)")
transp.checkClosed(retFuture) transp.checkClosed(retFuture)
let length = if msglen <= 0: (len(msg) * sizeof(T)) else: (msglen * sizeof(T)) let length = if msglen <= 0: (len(msg) * sizeof(T)) else: (msglen * sizeof(T))
var localCopy = chronosMoveSink(msg) var localCopy = chronosMoveSink(msg)
retFuture.addCallback(proc(_: pointer) = reset(localCopy)) retFuture.addCallback(proc(_: pointer) = reset(localCopy))
let vector = GramVector(kind: WithoutAddress, buf: addr localCopy[0], let vector = GramVector(kind: WithoutAddress, buf: baseAddr localCopy,
buflen: length, buflen: length,
writer: retFuture) writer: retFuture)
transp.queue.addLast(vector) transp.queue.addLast(vector)
@ -922,7 +923,7 @@ proc sendTo*(transp: DatagramTransport, remote: TransportAddress,
async: (raw: true, raises: [TransportError, CancelledError]).} = async: (raw: true, raises: [TransportError, CancelledError]).} =
## Send buffer with pointer ``pbytes`` and size ``nbytes`` using transport ## Send buffer with pointer ``pbytes`` and size ``nbytes`` using transport
## ``transp`` to remote destination address ``remote``. ## ``transp`` to remote destination address ``remote``.
var retFuture = newFuture[void]("datagram.transport.sendTo(pointer)") let retFuture = newFuture[void]("datagram.transport.sendTo(pointer)")
transp.checkClosed(retFuture) transp.checkClosed(retFuture)
let vector = GramVector(kind: WithAddress, buf: pbytes, buflen: nbytes, let vector = GramVector(kind: WithAddress, buf: pbytes, buflen: nbytes,
writer: retFuture, address: remote) writer: retFuture, address: remote)
@ -938,14 +939,14 @@ proc sendTo*(transp: DatagramTransport, remote: TransportAddress,
async: (raw: true, raises: [TransportError, CancelledError]).} = async: (raw: true, raises: [TransportError, CancelledError]).} =
## Send string ``msg`` using transport ``transp`` to remote destination ## Send string ``msg`` using transport ``transp`` to remote destination
## address ``remote``. ## address ``remote``.
var retFuture = newFuture[void]("datagram.transport.sendTo(string)") let retFuture = newFuture[void]("datagram.transport.sendTo(string)")
transp.checkClosed(retFuture) transp.checkClosed(retFuture)
let length = if msglen <= 0: len(msg) else: msglen let length = if msglen <= 0: len(msg) else: msglen
var localCopy = chronosMoveSink(msg) var localCopy = chronosMoveSink(msg)
retFuture.addCallback(proc(_: pointer) = reset(localCopy)) retFuture.addCallback(proc(_: pointer) = reset(localCopy))
let vector = GramVector(kind: WithAddress, buf: addr localCopy[0], let vector = GramVector(kind: WithAddress, buf: baseAddr localCopy,
buflen: length, buflen: length,
writer: retFuture, writer: retFuture,
address: remote) address: remote)
@ -961,15 +962,15 @@ proc sendTo*[T](transp: DatagramTransport, remote: TransportAddress,
async: (raw: true, raises: [TransportError, CancelledError]).} = async: (raw: true, raises: [TransportError, CancelledError]).} =
## Send sequence ``msg`` using transport ``transp`` to remote destination ## Send sequence ``msg`` using transport ``transp`` to remote destination
## address ``remote``. ## address ``remote``.
var retFuture = newFuture[void]("datagram.transport.sendTo(seq)") let retFuture = newFuture[void]("datagram.transport.sendTo(seq)")
transp.checkClosed(retFuture) transp.checkClosed(retFuture)
let length = if msglen <= 0: (len(msg) * sizeof(T)) else: (msglen * sizeof(T)) let length = if msglen <= 0: (len(msg) * sizeof(T)) else: (msglen * sizeof(T))
var localCopy = chronosMoveSink(msg) var localCopy = chronosMoveSink(msg)
retFuture.addCallback(proc(_: pointer) = reset(localCopy)) retFuture.addCallback(proc(_: pointer) = reset(localCopy))
let vector = GramVector(kind: WithAddress, buf: addr localCopy[0], let vector = GramVector(kind: WithAddress, buf: baseAddr localCopy,
buflen: length, buflen: length,
writer: cast[Future[void]](retFuture), writer: retFuture,
address: remote) address: remote)
transp.queue.addLast(vector) transp.queue.addLast(vector)
if WritePaused in transp.state: if WritePaused in transp.state:
@ -993,7 +994,6 @@ proc peekMessage*(transp: DatagramTransport, msg: var seq[byte],
proc getMessage*(transp: DatagramTransport): seq[byte] {. proc getMessage*(transp: DatagramTransport): seq[byte] {.
raises: [TransportError].} = raises: [TransportError].} =
## Copy data from internal message buffer and return result. ## Copy data from internal message buffer and return result.
var default: seq[byte]
if ReadError in transp.state: if ReadError in transp.state:
transp.state.excl(ReadError) transp.state.excl(ReadError)
raise transp.getError() raise transp.getError()
@ -1002,7 +1002,7 @@ proc getMessage*(transp: DatagramTransport): seq[byte] {.
copyMem(addr res[0], addr transp.buffer[0], transp.buflen) copyMem(addr res[0], addr transp.buffer[0], transp.buflen)
res res
else: else:
default default(seq[byte])
proc getUserData*[T](transp: DatagramTransport): T {.inline.} = proc getUserData*[T](transp: DatagramTransport): T {.inline.} =
## Obtain user data stored in ``transp`` object. ## Obtain user data stored in ``transp`` object.