nimbus-eth1/fluffy/network/state/state_network.nim

118 lines
3.6 KiB
Nim
Raw Normal View History

# Nimbus
# Copyright (c) 2021-2022 Status Research & Development GmbH
# Licensed and distributed under either of
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed except according to those terms.
import
std/options,
stew/results, chronos, chronicles,
eth/p2p/discoveryv5/[protocol, enr],
../../content_db,
../wire/[portal_protocol, portal_stream, portal_protocol_config],
./state_content,
./state_distance
logScope:
topics = "portal_state"
const
stateProtocolId* = [byte 0x50, 0x0A]
type StateNetwork* = ref object
portalProtocol*: PortalProtocol
contentDB*: ContentDB
2022-08-17 07:32:06 +00:00
contentQueue*: AsyncQueue[(ContentKeysList, seq[seq[byte]])]
processContentLoop: Future[void]
func toContentIdHandler(contentKey: ByteList): Option[ContentId] =
toContentId(contentKey)
proc dbGetHandler(db: ContentDB, contentKey: ByteList):
(Option[ContentId], Option[seq[byte]]) =
let contentIdOpt = contentKey.toContentId()
if contentIdOpt.isSome():
(contentIdOpt, db.get(contentIdOpt.get()))
else:
(contentIdOpt, none(seq[byte]))
proc getContent*(n: StateNetwork, key: ContentKey):
2021-09-13 13:56:44 +00:00
Future[Option[seq[byte]]] {.async.} =
let
keyEncoded = encode(key)
contentId = toContentId(key)
contentInRange = n.portalProtocol.inRange(contentId)
# When the content id is in the radius range, try to look it up in the db.
if contentInRange:
let contentFromDB = n.contentDB.get(contentId)
if contentFromDB.isSome():
return contentFromDB
let content = await n.portalProtocol.contentLookup(keyEncoded, contentId)
if content.isNone():
return none[seq[byte]]()
let contentResult = content.get()
# When content is found on the network and is in the radius range, store it.
if content.isSome() and contentInRange:
# TODO Add poke when working on state network
# TODO When working on state network, make it possible to pass different
# distance functions to store content
n.portalProtocol.storeContent(contentId, contentResult.content)
# TODO: for now returning bytes, ultimately it would be nice to return proper
# domain types.
return some(contentResult.content)
proc validateContent(content: openArray[byte], contentKey: ByteList): bool =
true
proc new*(
T: type StateNetwork,
baseProtocol: protocol.Protocol,
contentDB: ContentDB,
2022-08-17 07:32:06 +00:00
streamManager: StreamManager,
bootstrapRecords: openArray[Record] = [],
portalConfig: PortalProtocolConfig = defaultPortalProtocolConfig): T =
2022-08-17 07:32:06 +00:00
let cq = newAsyncQueue[(ContentKeysList, seq[seq[byte]])](50)
let s = streamManager.registerNewStream(cq)
let portalProtocol = PortalProtocol.new(
baseProtocol, stateProtocolId, contentDB,
2022-08-17 07:32:06 +00:00
toContentIdHandler, dbGetHandler, s,
bootstrapRecords, stateDistanceCalculator,
config = portalConfig)
2021-09-13 13:56:44 +00:00
2022-08-17 07:32:06 +00:00
return StateNetwork(
portalProtocol: portalProtocol,
contentDB: contentDB,
contentQueue: cq
)
proc processContentLoop(n: StateNetwork) {.async.} =
try:
while true:
# Just dropping state date for now
2022-08-22 10:23:26 +00:00
discard await n.contentQueue.popFirst()
except CancelledError:
trace "processContentLoop canceled"
proc start*(n: StateNetwork) =
info "Starting Portal execution state network",
protocolId = n.portalProtocol.protocolId
n.portalProtocol.start()
n.processContentLoop = processContentLoop(n)
proc stop*(n: StateNetwork) =
n.portalProtocol.stop()
if not n.processContentLoop.isNil:
n.processContentLoop.cancel()