fix(gossip): use async UDP socket to avoid blocking the event loop
The gossip listener used synchronous newSocket + recvFrom, which blocks the async event loop and prevents other async operations from running while waiting for UDP packets. Switch to newAsyncSocket + async recvFrom so the loop can continue processing other tasks between gossip rounds.
This commit is contained in:
@@ -4,6 +4,7 @@ import std/random
|
|||||||
import std/monotimes
|
import std/monotimes
|
||||||
import std/asyncdispatch
|
import std/asyncdispatch
|
||||||
import std/net
|
import std/net
|
||||||
|
import std/asyncnet
|
||||||
import std/strutils
|
import std/strutils
|
||||||
import std/streams
|
import std/streams
|
||||||
import std/os
|
import std/os
|
||||||
@@ -36,7 +37,7 @@ type
|
|||||||
fanout*: int # number of nodes to gossip to per round
|
fanout*: int # number of nodes to gossip to per round
|
||||||
gossipPort*: int
|
gossipPort*: int
|
||||||
running*: bool
|
running*: bool
|
||||||
sock*: Socket
|
sock*: AsyncSocket
|
||||||
onJoin*: proc(node: GossipNode) {.gcsafe.}
|
onJoin*: proc(node: GossipNode) {.gcsafe.}
|
||||||
onLeave*: proc(nodeId: string) {.gcsafe.}
|
onLeave*: proc(nodeId: string) {.gcsafe.}
|
||||||
onSuspect*: proc(nodeId: string) {.gcsafe.}
|
onSuspect*: proc(nodeId: string) {.gcsafe.}
|
||||||
@@ -305,25 +306,21 @@ proc startGossipRound*(gp: GossipProtocol, intervalMs: int = 2000) {.async.} =
|
|||||||
gp.broadcastGossip()
|
gp.broadcastGossip()
|
||||||
|
|
||||||
proc startGossipListener*(gp: GossipProtocol) {.async.} =
|
proc startGossipListener*(gp: GossipProtocol) {.async.} =
|
||||||
gp.sock = newSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
|
gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
|
||||||
gp.sock.setSockOpt(OptReuseAddr, true)
|
gp.sock.setSockOpt(OptReuseAddr, true)
|
||||||
gp.sock.bindAddr(Port(gp.gossipPort))
|
gp.sock.bindAddr(Port(gp.gossipPort))
|
||||||
gp.running = true
|
gp.running = true
|
||||||
while gp.running:
|
while gp.running:
|
||||||
try:
|
try:
|
||||||
var data = newString(65535)
|
let (data, senderAddr, senderPort) = await gp.sock.recvFrom(65535)
|
||||||
var senderAddr = ""
|
if data.len > 0:
|
||||||
var senderPort: Port
|
|
||||||
let bytesRead = gp.sock.recvFrom(data, 65535, senderAddr, senderPort)
|
|
||||||
if bytesRead > 0:
|
|
||||||
data.setLen(bytesRead)
|
|
||||||
gp.handleIncomingGossip(data, senderAddr)
|
gp.handleIncomingGossip(data, senderAddr)
|
||||||
except:
|
except:
|
||||||
# Small sleep on error to avoid spin
|
# Small sleep on error to avoid spin
|
||||||
gp.sock.close()
|
gp.sock.close()
|
||||||
# Recreate socket for next iteration
|
# Recreate socket for next iteration
|
||||||
try:
|
try:
|
||||||
gp.sock = newSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
|
gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
|
||||||
gp.sock.setSockOpt(OptReuseAddr, true)
|
gp.sock.setSockOpt(OptReuseAddr, true)
|
||||||
gp.sock.bindAddr(Port(gp.gossipPort))
|
gp.sock.bindAddr(Port(gp.gossipPort))
|
||||||
except:
|
except:
|
||||||
|
|||||||
Reference in New Issue
Block a user