logos-storage/logos-storage-network-crawler
1
0
Fork 0
mirror of https://github.com/logos-storage/logos-storage-network-crawler.git synced 2026-01-28 10:13:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

working example of event

Browse Source
This commit is contained in:
Ben 2025-02-10 15:34:41 +01:00
parent 50962d9a91
commit 218443ebe4
No known key found for this signature in database
GPG Key ID: 0F16E812E736C24B
12 changed files with 152 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
codexcrawler/application.nim
View File

@ -11,9 +11,11 @@ import ./utils/logging
import ./metrics
import ./list
import ./utils/datastoreutils
import ./utils/asyncdataevent
import ./installer
import ./state
import ./component
import ./types
declareGauge(todoNodesGauge, "DHT nodes to be visited")
declareGauge(okNodesGauge, "DHT nodes successfully contacted")
@ -83,13 +85,24 @@ proc initializeApp(app: Application): Future[?!void] {.async.} =
# todo move this
let state = State(
config: app.config
config: app.config,
events: Events(
nodesFound: newAsyncDataEvent[seq[Nid]](),
newNodesDiscovered: newAsyncDataEvent[seq[Nid]](),
dhtNodeCheck: newAsyncDataEvent[DhtNodeCheckEventData](),
nodesExpired: newAsyncDataEvent[seq[Nid]](),
),
)
for c in components:
if err =? (await c.start(state)).errorOption:
error "Failed to start component", err = err.msg
# test raise newnodes
let nodes: seq[Nid] = newSeq[Nid]()
if err =? (await state.events.nodesFound.fire(nodes)).errorOption:
return failure(err)
return success()
proc stop*(app: Application) =

8
codexcrawler/component.nim
View File

@ -3,12 +3,10 @@ import pkg/questionable/results
import ./state
type
Component* = ref object of RootObj
type Component* = ref object of RootObj
method start*(c: Component, state: State): Future[?!void] {.async, base.} =
raiseAssert("call to abstract method")
raiseAssert("call to abstract method: component.start")
method stop*(c: Component): Future[?!void] {.async, base.} =
raiseAssert("call to abstract method")
raiseAssert("call to abstract method: component.stop")

38
codexcrawler/components/crawler.nim
View File

@ -8,6 +8,9 @@ import ../list
import ../nodeentry
import ../config
import ../component
import ../types
import ../state
import ../utils/asyncdataevent
import std/sequtils
@ -72,26 +75,33 @@ proc step(c: Crawler) {.async.} =
proc worker(c: Crawler) {.async.} =
try:
while true:
await c.step()
# await c.step()
await sleepAsync(c.config.stepDelayMs.millis)
except Exception as exc:
error "Exception in crawler worker", msg = exc.msg
quit QuitFailure
proc start*(c: Crawler): Future[?!void] {.async.} =
if c.todoNodes.len < 1:
let nodeIds = c.dht.getRoutingTableNodeIds()
info "Loading routing-table nodes to todo-list...", nodes = nodeIds.len
for id in nodeIds:
if err =? (await c.addNewTodoNode(id)).errorOption:
error "Failed to add routing-table node to todo-list", err = err.msg
return failure(err)
method start*(c: Crawler, state: State): Future[?!void] {.async.} =
# if c.todoNodes.len < 1:
# let nodeIds = c.dht.getRoutingTableNodeIds()
# info "Loading routing-table nodes to todo-list...", nodes = nodeIds.len
# for id in nodeIds:
# if err =? (await c.addNewTodoNode(id)).errorOption:
# error "Failed to add routing-table node to todo-list", err = err.msg
# return failure(err)
info "Starting crawler...", stepDelayMs = $c.config.stepDelayMs
asyncSpawn c.worker()
proc onNodesFound(nids: seq[Nid]): Future[?!void] {.async.} =
info "Crawler sees nodes found!", num = nids.len
return success()
let handle = state.events.nodesFound.subscribe(onNodesFound)
return success()
proc stop*(c: Crawler): Future[?!void] {.async.} =
method stop*(c: Crawler): Future[?!void] {.async.} =
return success()
proc new*(
@ -102,7 +112,7 @@ proc new*(
# nokNodes: List,
config: Config,
): Crawler =
raiseAssert("todo")
# Crawler(
# dht: dht, todoNodes: todoNodes, okNodes: okNodes, nokNodes: nokNodes, config: config
# )
Crawler(
dht: dht,
config: config, # todoNodes: todoNodes, okNodes: okNodes, nokNodes: nokNodes,
)

42
codexcrawler/components/dht.nim
View File

@ -1,3 +1,4 @@
import std/os
import std/net
import pkg/chronicles
import pkg/chronos
@ -7,8 +8,12 @@ import pkg/questionable/results
import pkg/codexdht/discv5/[routing_table, protocol as discv5]
from pkg/nimcrypto import keccak256
import ../utils/keyutils
import ../utils/datastoreutils
import ../utils/rng
import ../component
import ../config
import ../state
export discv5
@ -97,16 +102,16 @@ proc updateDhtRecord*(d: Dht, addrs: openArray[MultiAddress]) =
if not d.protocol.isNil:
d.protocol.updateRecord(d.dhtRecord).expect("Should update SPR")
proc start*(d: Dht): Future[?!void] {.async.} =
method start*(d: Dht, state: State): Future[?!void] {.async.} =
d.protocol.open()
await d.protocol.start()
return success()
proc stop*(d: Dht): Future[?!void] {.async.} =
method stop*(d: Dht): Future[?!void] {.async.} =
await d.protocol.closeWait()
return success()
proc new*(
proc new(
T: type Dht,
key: PrivateKey,
bindIp = IPv4_any(),
@ -138,3 +143,34 @@ proc new*(
)
self
proc createDht*(config: Config): Future[?!Dht] {.async.} =
without dhtStore =? createDatastore(config.dataDir / "dht"), err:
return failure(err)
let keyPath = config.dataDir / "privatekey"
without privateKey =? setupKey(keyPath), err:
return failure(err)
var listenAddresses = newSeq[MultiAddress]()
# TODO: when p2p connections are supported:
# let aaa = MultiAddress.init("/ip4/" & config.publicIp & "/tcp/53678").expect("Should init multiaddress")
# listenAddresses.add(aaa)
var discAddresses = newSeq[MultiAddress]()
let bbb = MultiAddress
.init("/ip4/" & config.publicIp & "/udp/" & $config.discPort)
.expect("Should init multiaddress")
discAddresses.add(bbb)
let dht = Dht.new(
privateKey,
bindPort = config.discPort,
announceAddrs = listenAddresses,
bootstrapNodes = config.bootNodes,
store = dhtStore,
)
dht.updateAnnounceRecord(listenAddresses)
dht.updateDhtRecord(discAddresses)
return success(dht)

28
codexcrawler/components/timetracker.nim
View File

@ -8,6 +8,7 @@ import ../list
import ../nodeentry
import ../config
import ../component
import ../state
logScope:
topics = "timetracker"
@ -45,18 +46,18 @@ proc step(t: TimeTracker) {.async.} =
proc worker(t: TimeTracker) {.async.} =
try:
while true:
await t.step()
# await t.step()
await sleepAsync(t.workerDelay.minutes)
except Exception as exc:
error "Exception in timetracker worker", msg = exc.msg
quit QuitFailure
proc start*(t: TimeTracker): Future[?!void] {.async.} =
method start*(t: TimeTracker, state: State): Future[?!void] {.async.} =
info "Starting timetracker...", revisitDelayMins = $t.workerDelay
asyncSpawn t.worker()
return success()
proc stop*(t: TimeTracker): Future[?!void] {.async.} =
method stop*(t: TimeTracker): Future[?!void] {.async.} =
return success()
proc new*(
@ -66,15 +67,14 @@ proc new*(
# nokNodes: List,
config: Config,
): TimeTracker =
raiseAssert("todo")
# var delay = config.revisitDelayMins div 10
# if delay < 1:
# delay = 1
var delay = config.revisitDelayMins div 10
if delay < 1:
delay = 1
# TimeTracker(
# todoNodes: todoNodes,
# okNodes: okNodes,
# nokNodes: nokNodes,
# config: config,
# workerDelay: delay,
# )
TimeTracker(
# todoNodes: todoNodes,
# okNodes: okNodes,
# nokNodes: nokNodes,
config: config,
workerDelay: delay,
)

37
codexcrawler/installer.nim
View File

@ -1,6 +1,4 @@
import std/os
import pkg/chronos
import pkg/chronicles
import pkg/questionable/results
import ./config
@ -8,44 +6,11 @@ import ./component
import ./components/dht
import ./components/crawler
import ./components/timetracker
import ./utils/keyutils
import ./utils/datastoreutils
proc initializeDht(config: Config): Future[?!Dht] {.async.} =
without dhtStore =? createDatastore(config.dataDir / "dht"), err:
return failure(err)
let keyPath = config.dataDir / "privatekey"
without privateKey =? setupKey(keyPath), err:
return failure(err)
var listenAddresses = newSeq[MultiAddress]()
# TODO: when p2p connections are supported:
# let aaa = MultiAddress.init("/ip4/" & config.publicIp & "/tcp/53678").expect("Should init multiaddress")
# listenAddresses.add(aaa)
var discAddresses = newSeq[MultiAddress]()
let bbb = MultiAddress
.init("/ip4/" & config.publicIp & "/udp/" & $config.discPort)
.expect("Should init multiaddress")
discAddresses.add(bbb)
let dht = Dht.new(
privateKey,
bindPort = config.discPort,
announceAddrs = listenAddresses,
bootstrapNodes = config.bootNodes,
store = dhtStore,
)
dht.updateAnnounceRecord(listenAddresses)
dht.updateDhtRecord(discAddresses)
return success(dht)
proc createComponents*(config: Config): Future[?!seq[Component]] {.async.} =
var components: seq[Component] = newSeq[Component]()
without dht =? (await initializeDht(config)), err:
without dht =? (await createDht(config)), err:
return failure(err)
components.add(dht)

20
codexcrawler/list.nim
View File

@ -8,14 +8,13 @@ import pkg/stew/endians2
import pkg/questionable
import pkg/questionable/results
import pkg/stint
import pkg/codexdht
import std/sets
import std/strutils
import std/sequtils
import std/os
import ./nodeentry
import ./types
logScope:
topics = "list"
@ -36,7 +35,7 @@ proc encode(s: NodeEntry): seq[byte] =
proc decode(T: type NodeEntry, bytes: seq[byte]): ?!T =
if bytes.len < 1:
return success(NodeEntry(id: UInt256.fromHex("0"), lastVisit: 0.uint64))
return success(NodeEntry(id: Nid.fromStr("0"), lastVisit: 0.uint64))
return NodeEntry.fromBytes(bytes)
proc saveItem(this: List, item: NodeEntry): Future[?!void] {.async.} =
@ -46,6 +45,10 @@ proc saveItem(this: List, item: NodeEntry): Future[?!void] {.async.} =
return success()
proc load*(this: List): Future[?!void] {.async.} =
let id = Nid.fromStr("0")
let bytes = newSeq[byte]()
let ne = NodeEntry.fromBytes(bytes)
without queryKey =? Key.init(this.name), err:
return failure(err)
without iter =? (await query[NodeEntry](this.store, Query.init(queryKey))), err:
@ -68,7 +71,7 @@ proc new*(
): List =
List(name: name, store: store, onMetric: onMetric)
proc contains*(this: List, nodeId: NodeId): bool =
proc contains*(this: List, nodeId: Nid): bool =
this.items.anyIt(it.id == nodeId)
proc contains*(this: List, item: NodeEntry): bool =
@ -81,9 +84,9 @@ proc add*(this: List, item: NodeEntry): Future[?!void] {.async.} =
this.items.add(item)
this.onMetric(this.items.len.int64)
if isSome(this.emptySignal):
if s =? this.emptySignal:
trace "List no longer empty.", name = this.name
this.emptySignal.get().complete()
s.complete()
this.emptySignal = Future[void].none
if err =? (await this.saveItem(item)).errorOption:
@ -104,8 +107,9 @@ proc remove*(this: List, item: NodeEntry): Future[?!void] {.async.} =
proc pop*(this: List): Future[?!NodeEntry] {.async.} =
if this.items.len < 1:
trace "List is empty. Waiting for new items...", name = this.name
this.emptySignal = some(newFuture[void]("list.emptySignal"))
await this.emptySignal.get().wait(1.hours)
let signal = newFuture[void]("list.emptySignal")
this.emptySignal = some(signal)
await signal.wait(1.hours)
if this.items.len < 1:
return failure(this.name & "List is empty.")

10
codexcrawler/nodeentry.nim
View File

@ -1,13 +1,11 @@
import pkg/stew/byteutils
import pkg/stew/endians2
import pkg/questionable
import pkg/questionable/results
import pkg/codexdht
import pkg/chronos
import pkg/libp2p
import ./types
type NodeEntry* = object
id*: NodeId
id*: Nid
lastVisit*: uint64
proc `$`*(entry: NodeEntry): string =
@ -32,4 +30,4 @@ proc fromBytes*(_: type NodeEntry, data: openArray[byte]): ?!NodeEntry =
if buffer.getField(2, lastVisit).isErr:
return failure("Unable to decode `lastVisit`")
return success(NodeEntry(id: UInt256.fromHex(idStr), lastVisit: lastVisit))
return success(NodeEntry(id: Nid.fromStr(idStr), lastVisit: lastVisit))

16
codexcrawler/state.nim
View File

@ -2,13 +2,25 @@ import pkg/chronos
import pkg/questionable/results
import ./config
import ./utils/asyncdataevent
import ./types
type
OnStep = proc(): Future[?!void] {.async: (raises: []), gcsafe.}
DhtNodeCheckEventData* = object
id*: Nid
isOk*: bool
Events* = ref object
nodesFound*: AsyncDataEvent[seq[Nid]]
newNodesDiscovered*: AsyncDataEvent[seq[Nid]]
dhtNodeCheck*: AsyncDataEvent[DhtNodeCheckEventData]
nodesExpired*: AsyncDataEvent[seq[Nid]]
State* = ref object
config*: Config
# events
# appstate
events*: Events # appstate
proc whileRunning*(this: State, step: OnStep, delay: Duration) =
discard

12
codexcrawler/types.nim Normal file
View File

@ -0,0 +1,12 @@
import pkg/stew/byteutils
import pkg/stew/endians2
import pkg/questionable
import pkg/codexdht
type Nid* = NodeId
proc `$`*(nid: Nid): string =
$(NodeId(nid))
proc fromStr*(T: type Nid, s: string): Nid =
Nid(UInt256.fromHex(s))

13
codexcrawler/utils/asyncdataevent.nim
View File

@ -18,16 +18,17 @@ type
proc newAsyncDataEvent*[T](): AsyncDataEvent[T] =
AsyncDataEvent[T](
queue: newAsyncEventQueue[?T](),
subscriptions: newSeq[AsyncDataEventSubscription]()
queue: newAsyncEventQueue[?T](), subscriptions: newSeq[AsyncDataEventSubscription]()
)
proc subscribe*[T](event: AsyncDataEvent[T], handler: AsyncDataEventHandler[T]): AsyncDataEventSubscription =
proc subscribe*[T](
event: AsyncDataEvent[T], handler: AsyncDataEventHandler[T]
): AsyncDataEventSubscription =
let subscription = AsyncDataEventSubscription(
key: event.queue.register(),
isRunning: true,
fireEvent: newAsyncEvent(),
stopEvent: newAsyncEvent()
stopEvent: newAsyncEvent(),
)
proc listener() {.async.} =
@ -52,7 +53,9 @@ proc fire*[T](event: AsyncDataEvent[T], data: T): Future[?!void] {.async.} =
return failure(err)
success()
proc unsubscribe*[T](event: AsyncDataEvent[T], subscription: AsyncDataEventSubscription) {.async.} =
proc unsubscribe*[T](
event: AsyncDataEvent[T], subscription: AsyncDataEventSubscription
) {.async.} =
subscription.isRunning = false
event.queue.emit(T.none)
await subscription.stopEvent.wait()

19
tests/codexcrawler/utils/testasyncdataevent.nim
View File

@ -5,9 +5,8 @@ import pkg/asynctest/chronos/unittest
import ../../../codexcrawler/utils/asyncdataevent
type
ExampleData = object
s: string
type ExampleData = object
s: string
suite "AsyncDataEvent":
var event: AsyncDataEvent[ExampleData]
@ -28,9 +27,7 @@ suite "AsyncDataEvent":
let s = event.subscribe(eventHandler)
check:
isOK(await event.fire(ExampleData(
s: msg
)))
isOK(await event.fire(ExampleData(s: msg)))
data == msg
await event.unsubscribe(s)
@ -40,9 +37,7 @@ suite "AsyncDataEvent":
failure(msg)
let s = event.subscribe(eventHandler)
let fireResult = await event.fire(ExampleData(
s: "a"
))
let fireResult = await event.fire(ExampleData(s: "a"))
check:
fireResult.isErr
@ -59,9 +54,11 @@ suite "AsyncDataEvent":
proc handler1(e: ExampleData): Future[?!void] {.async.} =
data1 = e.s
success()
proc handler2(e: ExampleData): Future[?!void] {.async.} =
data2 = e.s
success()
proc handler3(e: ExampleData): Future[?!void] {.async.} =
data3 = e.s
success()
@ -71,9 +68,7 @@ suite "AsyncDataEvent":
s2 = event.subscribe(handler2)
s3 = event.subscribe(handler3)
let fireResult = await event.fire(ExampleData(
s: msg
))
let fireResult = await event.fire(ExampleData(s: msg))
check:
fireResult.isOK