mirror of
https://github.com/logos-storage/swarmsim.git
synced 2026-01-08 00:33:06 +00:00
add basic peer lifecycle
This commit is contained in:
parent
ee3cc72213
commit
df3bd77384
@ -2,6 +2,7 @@ import std/options
|
|||||||
import std/random
|
import std/random
|
||||||
import std/hashes
|
import std/hashes
|
||||||
import sequtils
|
import sequtils
|
||||||
|
import sugar
|
||||||
|
|
||||||
import ./types
|
import ./types
|
||||||
import ./message
|
import ./message
|
||||||
@ -15,6 +16,33 @@ export protocol
|
|||||||
export eventdrivenengine
|
export eventdrivenengine
|
||||||
export Peer
|
export Peer
|
||||||
|
|
||||||
|
type
|
||||||
|
PeerLifecycleChange* = ref object of SchedulableEvent
|
||||||
|
peer: Peer
|
||||||
|
network: Network
|
||||||
|
event: LifecycleEventType
|
||||||
|
|
||||||
|
method atScheduledTime*(self: PeerLifecycleChange,
|
||||||
|
engine: EventDrivenEngine): void =
|
||||||
|
|
||||||
|
let peer = self.peer
|
||||||
|
let oldState = peer.up
|
||||||
|
|
||||||
|
# XXX We're somewhat lax with state machine transitions and will allow
|
||||||
|
# self-transitions...
|
||||||
|
let newState = case self.event:
|
||||||
|
of started, up:
|
||||||
|
true
|
||||||
|
of down:
|
||||||
|
false
|
||||||
|
|
||||||
|
peer.up = newState
|
||||||
|
|
||||||
|
# ... but self-transitions do not get reported downstream.
|
||||||
|
if oldState != newState:
|
||||||
|
self.peer.protocols.values.toSeq.apply(p =>
|
||||||
|
p.onLifecycleEventType(self.peer, self.event, self.network))
|
||||||
|
|
||||||
proc getProtocol*(self: Peer, id: string): Option[Protocol] =
|
proc getProtocol*(self: Peer, id: string): Option[Protocol] =
|
||||||
if self.protocols.hasKey(id):
|
if self.protocols.hasKey(id):
|
||||||
return self.protocols[id].some
|
return self.protocols[id].some
|
||||||
@ -34,6 +62,21 @@ proc deliver*(self: Peer, message: Message, engine: EventDrivenEngine,
|
|||||||
self.deliverForType(message.messageType, message, engine, network)
|
self.deliverForType(message.messageType, message, engine, network)
|
||||||
self.deliverForType(Message.allMessages, message, engine, network)
|
self.deliverForType(Message.allMessages, message, engine, network)
|
||||||
|
|
||||||
|
proc scheduleLifecycleChange*(self: Peer, event: LifecycleEventType,
|
||||||
|
network: Network, time: uint64): void =
|
||||||
|
network.engine.schedule(PeerLifecycleChange(
|
||||||
|
peer: self,
|
||||||
|
network: network,
|
||||||
|
event: event,
|
||||||
|
time: time
|
||||||
|
))
|
||||||
|
|
||||||
|
proc startAt*(self: Peer, network: Network, time: uint64): void =
|
||||||
|
self.scheduleLifecycleChange(started, network, time)
|
||||||
|
|
||||||
|
proc start*(self: Peer, network: Network): void =
|
||||||
|
self.startAt(network, network.engine.currentTime)
|
||||||
|
|
||||||
proc initPeer*(self: Peer, protocols: seq[Protocol],
|
proc initPeer*(self: Peer, protocols: seq[Protocol],
|
||||||
peerId: Option[int] = none(int)): Peer =
|
peerId: Option[int] = none(int)): Peer =
|
||||||
|
|
||||||
|
|||||||
@ -14,3 +14,11 @@ method deliver*(
|
|||||||
network: Network
|
network: Network
|
||||||
): void {.base.} =
|
): void {.base.} =
|
||||||
raise newException(CatchableError, "Method without implementation override")
|
raise newException(CatchableError, "Method without implementation override")
|
||||||
|
|
||||||
|
method onLifecycleEventType*(
|
||||||
|
self: Protocol,
|
||||||
|
peer: Peer,
|
||||||
|
event: LifecycleEventType,
|
||||||
|
network: Network
|
||||||
|
): void {.base.} =
|
||||||
|
discard
|
||||||
|
|||||||
@ -37,8 +37,14 @@ type
|
|||||||
messageTypes*: seq[string]
|
messageTypes*: seq[string]
|
||||||
|
|
||||||
type
|
type
|
||||||
|
LifecycleEventType* = enum
|
||||||
|
started
|
||||||
|
up
|
||||||
|
down
|
||||||
|
|
||||||
Peer* = ref object of RootObj
|
Peer* = ref object of RootObj
|
||||||
peerId*: int
|
peerId*: int
|
||||||
|
up*: bool = false
|
||||||
protocols*: Table[string, Protocol]
|
protocols*: Table[string, Protocol]
|
||||||
dispatch*: MultiTable[string, Protocol]
|
dispatch*: MultiTable[string, Protocol]
|
||||||
|
|
||||||
|
|||||||
@ -9,9 +9,11 @@ import ../helpers/testpeer
|
|||||||
|
|
||||||
suite "block exchange":
|
suite "block exchange":
|
||||||
|
|
||||||
test "should respond to want-block message with a list of the blocks it has":
|
setup:
|
||||||
let engine = EventDrivenEngine()
|
let engine = EventDrivenEngine()
|
||||||
let network = Network(engine: engine)
|
let network = Network(engine: engine)
|
||||||
|
|
||||||
|
test "should respond to want-have message with a list of the blocks it has":
|
||||||
let sender = TestPeer.new(network)
|
let sender = TestPeer.new(network)
|
||||||
|
|
||||||
let bex = BlockExchangeProtocol.new()
|
let bex = BlockExchangeProtocol.new()
|
||||||
@ -41,3 +43,6 @@ suite "block exchange":
|
|||||||
let response = Have(sender.inbox.messages[0])
|
let response = Have(sender.inbox.messages[0])
|
||||||
|
|
||||||
check(response.haves == toIntSet([0, 1, 4]))
|
check(response.haves == toIntSet([0, 1, 4]))
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@ -10,6 +10,7 @@ import ../helpers/inbox
|
|||||||
import ../helpers/types
|
import ../helpers/types
|
||||||
|
|
||||||
suite "network":
|
suite "network":
|
||||||
|
|
||||||
test "should dispatch message to the correct peer":
|
test "should dispatch message to the correct peer":
|
||||||
|
|
||||||
let engine = EventDrivenEngine()
|
let engine = EventDrivenEngine()
|
||||||
|
|||||||
@ -9,6 +9,8 @@ import swarmsim/engine/message
|
|||||||
import ../helpers/inbox
|
import ../helpers/inbox
|
||||||
import ../helpers/types
|
import ../helpers/types
|
||||||
|
|
||||||
|
import pretty
|
||||||
|
|
||||||
# We need this here as otherwise for some reason the nim compiler trips.
|
# We need this here as otherwise for some reason the nim compiler trips.
|
||||||
proc `$`*(m: Message): string = repr m
|
proc `$`*(m: Message): string = repr m
|
||||||
|
|
||||||
@ -79,7 +81,6 @@ suite "peer":
|
|||||||
|
|
||||||
check(i1.messages == @[m1, m2])
|
check(i1.messages == @[m1, m2])
|
||||||
|
|
||||||
|
|
||||||
test "should deliver all message types when listening to Message.allMessages":
|
test "should deliver all message types when listening to Message.allMessages":
|
||||||
let i1 = Inbox(protocolId: "protocol1", messageTypes: @[Message.allMessages])
|
let i1 = Inbox(protocolId: "protocol1", messageTypes: @[Message.allMessages])
|
||||||
|
|
||||||
@ -95,3 +96,36 @@ suite "peer":
|
|||||||
|
|
||||||
check(i1.messages == @[m1, m2, m3])
|
check(i1.messages == @[m1, m2, m3])
|
||||||
|
|
||||||
|
test "should be up after start":
|
||||||
|
let i1 = Inbox(protocolId: "lifecycleListener", messageTypes: @[])
|
||||||
|
let peer = Peer.new(protocols = @[Protocol i1])
|
||||||
|
|
||||||
|
peer.startAt(network, 5)
|
||||||
|
|
||||||
|
check(not peer.up)
|
||||||
|
|
||||||
|
engine.run()
|
||||||
|
|
||||||
|
check((repr i1.events) == (repr @[
|
||||||
|
LifecycleEvent(event: started, time: 5.uint64)]))
|
||||||
|
|
||||||
|
check(peer.up)
|
||||||
|
|
||||||
|
test "should not be up after going down":
|
||||||
|
let i1 = Inbox(protocolId: "lifecycleListener", messageTypes: @[])
|
||||||
|
|
||||||
|
let peer = Peer.new(protocols = @[Protocol i1])
|
||||||
|
|
||||||
|
peer.start(network)
|
||||||
|
peer.scheduleLifecycleChange(
|
||||||
|
event = down,
|
||||||
|
network = network,
|
||||||
|
time = 10
|
||||||
|
)
|
||||||
|
|
||||||
|
engine.run()
|
||||||
|
|
||||||
|
check((repr i1.events) == (repr @[
|
||||||
|
LifecycleEvent(event: started, time: 0),
|
||||||
|
LifecycleEvent(event: down, time: 10)
|
||||||
|
]))
|
||||||
|
|||||||
@ -9,6 +9,11 @@ withTypeId:
|
|||||||
Inbox* = ref object of Protocol
|
Inbox* = ref object of Protocol
|
||||||
protocolId*: string
|
protocolId*: string
|
||||||
messages*: seq[Message]
|
messages*: seq[Message]
|
||||||
|
events*: seq[LifecycleEvent]
|
||||||
|
|
||||||
|
LifecycleEvent* = ref object of RootObj
|
||||||
|
event*: LifecycleEventType
|
||||||
|
time*: uint64
|
||||||
|
|
||||||
method deliver*(
|
method deliver*(
|
||||||
self: Inbox,
|
self: Inbox,
|
||||||
@ -20,7 +25,16 @@ method deliver*(
|
|||||||
|
|
||||||
method `protocolId`*(self: Inbox): string = self.protocolId
|
method `protocolId`*(self: Inbox): string = self.protocolId
|
||||||
|
|
||||||
|
method onLifecycleEventType*(
|
||||||
|
self: Inbox,
|
||||||
|
peer: Peer,
|
||||||
|
event: LifecycleEventType,
|
||||||
|
network: Network
|
||||||
|
) =
|
||||||
|
self.events.add(LifecycleEvent(event: event, time: network.engine.currentTime))
|
||||||
|
|
||||||
export Message
|
export Message
|
||||||
|
export LifecycleEvent
|
||||||
export peer
|
export peer
|
||||||
export protocol
|
export protocol
|
||||||
export network
|
export network
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user