Server

The server is where you deploy all of the peers' logic from. The subnet template comes with a base class for building the application logic on top of. The base class exposes the application logic to all of the P2P logic if peers need to access it.

The ServerBase is a class that handles all of the peer discovery and connection logic, enabling developers to build the application-level logic, such as the peers or different roles business logic, how peers score one another, and setting up other communication templates.

Server Base

Application

Application Base

To develop the application logic of the peers or roles, you have to inherit the ApplicationBase class and fill out the required functions.

__init__: Initialize the class with all of your required arguments.

setup: Initialize local logic, such as the telemetry.

start_application: Initialize all of the logic required for the peer to run, such as the GossipSub, DAG, or protocols. Everything the peer requires to do their job.

run:

cleanup: This is a hook to clean up any processes if needed.

class ApplicationBase:
    def __init__(self, *args: object, **kwargs: object) -> None:
        self.args = args
        self.kwargs = kwargs

    async def setup(self, context: P2PNetworkContext) -> None:
        """Called once after host, DHT, and optional pubsub are ready."""

    async def start_application(self, context: P2PNetworkContext) -> None:
        """Called after bootstrap and optional connection maintenance startup."""

    async def run(self, context: P2PNetworkContext) -> None:
        """Block until the app or server should stop."""
        await context.termination_event.wait()

    async def cleanup(self, context: P2PNetworkContext) -> None:
        """Called once while the network context is still available."""

Application

The application class inherits the ApplicationBase class and is filled out.

from subnet.telemetry.telemetry import Telemetry


class MyApplication(ApplicationBase):
    def __init__(self, telemetry: Telemetry | None = None):
        self.telemetry = telemetry

    async def setup(self, context: P2PNetworkContext) -> None:
        if self.telemetry is not None:
            context.nursery.start_soon(self.telemetry.run)

        self.mock_protocol = MockProtocolV2(
            host=context.host,
            telemetry=self.telemetry,
        )

        if context.peer_multiaddr is not None:
            logger.info("Running peer on %s", context.peer_multiaddr)

Server

The server deploys the P2P logic and the application logic.

class MyServer(ServerBase):
    def __init__(
        self,
        *,
        ip: str | None = None,
        port: int,
        bootstrap_addrs: Sequence[str] | None = None,
        key_pair: KeyPair,
        db: RocksDB,
        subnet_id: int,
        subnet_slot: int = 3,
        subnet_node_id: int,
        hypertensor: Hypertensor | LocalMockHypertensor,
        is_bootstrap: bool = False,
        telemetry: Telemetry | None = None,
        enable_mDNS: bool = False,
        enable_upnp: bool = False,
        enable_autotls: bool = False,
        resource_manager: ResourceManager | None = None,
        psk: str | None = None,
    ) -> None:
        application = MyApplication(
            key_pair=key_pair,
            db=db,
            subnet_id=subnet_id,
            subnet_node_id=subnet_node_id,
            is_bootstrap=is_bootstrap,
            telemetry=telemetry,
        )

        super().__init__(
            ip=ip or "0.0.0.0",
            port=port,
            application=application,
            key_pair=key_pair,
            bootstrap_addrs=bootstrap_addrs,
            use_available_interfaces=True,
            enable_pubsub=True,
            enable_random_walk=True,
            enable_mDNS=enable_mDNS,
            enable_upnp=enable_upnp,
            enable_autotls=enable_autotls,
            resource_manager=resource_manager,
            psk=psk,
            max_connections_per_peer=6,
            enable_ping=True,
            enable_proof_of_stake=True,
            db=db,
            subnet_id=subnet_id,
            subnet_slot=subnet_slot,
            subnet_node_id=subnet_node_id,
            hypertensor=hypertensor,
            is_bootstrap=is_bootstrap,
            enable_subnet_info_tracker=True,
            enable_connection_maintenance=True,
            strict_maintain_connections=True,
            connection_maintenance_telemetry=telemetry,
            maintain_connections_log_level=logging.DEBUG,
        )