3.0.ipv8.peerdiscovery.discovery

Attributes

Classes

Module Contents

3.0.ipv8.peerdiscovery.discovery._OT

class 3.0.ipv8.peerdiscovery.discovery.DiscoveryStrategy(overlay: _OT)

Bases: Generic[_OT]

Strategy for discovering peers in a network.

overlay: _OT

walk_lock

abstract take_step() → None: Callback for when an IPv8 tick occurs (defaults to roughly every 0.5 seconds).

get_peer_count() → int

Determine the current number of peers. This is used by IPv8 to determine whether to call this strategy.

If self.overlay is not None, we expect it to have a get_peers() -> list[Peer] method.

class 3.0.ipv8.peerdiscovery.discovery.RandomWalk(overlay: 3.0.ipv8.types.Overlay, timeout: float = 3.0, window_size: int = 5, reset_chance: int = 50, target_interval: int = 0)

Bases: DiscoveryStrategy[3.0.ipv8.types.Overlay]

Walk randomly through the network.

intro_timeouts: dict[3.0.ipv8.types.Address, float]

node_timeout

window_size

reset_chance

target_interval

last_step: float = 0

take_step() → None: Walk to random walkable peer.

class 3.0.ipv8.peerdiscovery.discovery.EdgeWalk(overlay: 3.0.ipv8.types.Overlay, edge_length: int = 4, neighborhood_size: int = 6, edge_timeout: float = 3.0)

Bases: DiscoveryStrategy[3.0.ipv8.types.Overlay]

Walk through the network by using edges.

This will perform a depth-first search in the network starting from your direct neighborhood. When a certain depth is reached, we teleport home and start again from our neighborhood.

_neighborhood: list[3.0.ipv8.types.Peer] = []

complete_edges: list[list[3.0.ipv8.types.Peer]] = []

under_construction: dict[3.0.ipv8.types.Peer, list[3.0.ipv8.types.Peer]]

last_edge_responses: dict[3.0.ipv8.types.Peer, float]

edge_length

neighborhood_size

edge_timeout

get_available_root() → 3.0.ipv8.types.Peer | None: Get a root, if it exists, which is not busy constructing an edge for us.

take_step() → None: Attempt to grow an edge.