Distributed Hash Table (DHT)

The distributed hash table (DHT) is a lightweight distributed key-value store, that can store record across nodes.

The DHT Service

Python

from hyveos_sdk import Connection
import asyncio


async with Connection() as connection:
    dht = connection.get_dht_service()

    # continue with usage of dht
    # TODO

Rust

use hyveos_sdk::Connection;


#[tokio::main]
async fn main() {
    let connection = Connection::new().await.unwrap();
    let mut dht = connection.dht();

    // continue with usage of dht
    // todo!()
}

Typescript/JS

import { Client } from 'hyveos-sdk'
import { Connection } from "hyveos-web"


async function main() {
    const transport = new Connection('http://localhost:8080')
    const client = new Client(transport)
    const dhtService = client.dht

    // continue with usage of dhtService
    // TODO

    await transport.close();
}

Example 1. Obtain the DHT service handler. Note the async environment.

Put & Retrieve a record

The DHT is grouped into topics. You store a record into the DHT under a specific topic.

Python

node_charging.py

from hyveos_sdk import Connection
import asyncio


async def main():
    async with Connection() as connection:
        dht = connection.get_dht_service()
        discovery = connection.get_discovery_service()

        id = discovery.get_own_id()

        await dht.put_record("charging_pause", id, "1h")
        # TODO

asyncio.run(main())

Rust

node_charging.rs

use hyveos_sdk::Connection;


#[tokio::main]
async fn main() {
    let connection = Connection::new().await.unwrap();
    let mut dht = connection.dht();
    let mut discovery = connection.discovery();

    let id = discovery.get_own_id().await.unwrap();

    dht.put_record("charging_pause", id, "1h").await.unwrap();
    // todo!()
}

Typescript/JS

node_charging.ts

import { Client } from 'hyveos-sdk'
import { Connection } from "hyveos-web"


async function main() {
    const transport = new Connection('http://localhost:8080')
    const client = new Client(transport)
    const dhtService = client.dht
    const discoveryService = client.discovery

    const id = await discoveryService.getOwnId()

    await dhtService.putRecord('charging_pause', id, '1h')
    // TODO

    await transport.close()
}

main().catch(console.error)

Example 2. Put a record in the DHT. The runtime uses the discovery service to find its own peer id as a string, and then use it as a key. The value of "1h" is the duration of charging.

Rust: Data Encoding Formats

The Rust SDK allows the user to control in which format the data is encoded internally during transport.
The choice depends on your use case: cbor is more encoding efficient, json is human-readable.

JSON

dht.put_record_json("topic", "key", &value).await.unwrap();

CBOR

dht.put_record_cbor("topic", "key", &value).await.unwrap();

Next, retreiving a record works with get_record(topic, key).

Python

get_charging_information.py

from hyveos_sdk import Connection
import asyncio


async def main():
    async with Connection() as connection:
        dht = connection.get_dht_service()

        record = await dht.get_record("charging_pause", "012345678ABCDEF")

        if record is not None and record.data is not None:
            try:
                value = record.data.decode("utf-8")
                print(f"012345678ABCDEF is charging for: {value}")
            except UnicodeDecodeError:
                print("Record data is not valid UTF-8")
        else:
            print("Record not found")

asyncio.run(main())

Rust

get_charging_information.rs

use hyveos_sdk::Connection;


#[tokio::main]
async fn main() {
    let connection = Connection::new().await.unwrap();
    let mut dht = connection.dht();

    let value = dht.get_record("charging_pause", "012345678ABCDEF").await.unwrap();

    // convert the returned raw Vec<u8> into a UTF-8 string
    if let Some(value) = value.and_then(|value| String::from_utf8(value).ok()) {
        println!("012345678ABCDEF is charging for: {value}");
    } else {
        println!("Record not found");
    }
}

Typescript/JS

get_charging_information.ts

import { Client } from 'hyveos-sdk';
import { Connection } from "hyveos-web";


async function main() {
    const transport = new Connection('http://localhost:8080');
    const client = new Client(transport);
    const dhtService = client.dht;

    const record: Uint8Array | null = await dhtService.getRecord("charging_pause", "012345678ABCDEF");

    if (record !== null) {
        try {
            const decoded = new TextDecoder("utf-8").decode(record);
            console.log(`012345678ABCDEF is charging for: ${decoded}`);
        } catch (error) {
            console.error("Record data is not valid UTF-8", error);
        }
    } else {
        console.log("Record not found");
    }
}

main().catch(console.error);

Example 3. Retreive records from the DHT. The runtime looks for how long the peer with exemplary peer id "0123456789ABCDEF" will be charging, if even.

Correct peer id

In the example for get_record(topic, key) above, the peer id "0123456789ABCDEF" of the peer is an example string. If you use get_record(topic, key), with key being a peer id, you should first retreive the actual peer id. For this, use the Discovery Service, or reuse the peer id if you have already stored it within the current runtime.

Provide

Apart from storing and retrieving records, the DHT also exposes the interface to mark nodes as “providers” of a key.

This pattern can be useful in assigning and advertising nodes to specific semantic roles

Again, the provide(topic, key) interface marks the node as a provider for a key in the DHT, under a specified topic.

Python

camera_node.py

from hyveos_sdk import Connection
import asyncio


async with Connection() as connection:
    dht = connection.get_dht_service()

    await dht.provide('capabilities', 'camera')
    # TODO

Rust

camera_node.rs

use hyveos_sdk::Connection;


#[tokio::main]
async fn main() {
    let connection = Connection::new().await.unwrap();
    let mut dht_service = connection.dht();

    dht_service.provide("capabilities", "camera").await.unwrap();
    // todo!()
}

Typescript/JS

camera_node.ts

import { Client } from 'hyveos-sdk';
import { Connection } from "hyveos-web";


async function main() {
    const transport = new Connection('http://localhost:8080');
    const client = new Client(transport);
    const dhtService = client.dht;

    const key = new TextEncoder().encode("camera");
    await dhtService.provide("capabilities", key);
    // TODO

    // await transport.close();
}

main().catch(console.error);

Example 4. The node stores its role as a potential camera provider into the DHT.

The particular node that runs this code takes on the role of a robot or drone with visual capabilities.

If another node wants to retrieve which node provides the camera capability, it can search for those providers in the DHT.

get_providers(topic, key) returns an asynchronous iterator representing the providers of a record under a specific topic.

Python

other_node.py

from hyveos_sdk import Connection
import asyncio


async def main():
    async with Connection() as connection:
        dht = connection.get_dht_service()

        async with dht.get_providers("capabilities", "camera") as providers:
            async for provider in providers:
                print(f'Found a node with a camera: {provider}')

asyncio.run(main())

Rust

other_node.rs

use hyveos_sdk::Connection;
use futures::TryStreamExt as _;


#[tokio::main]
async fn main() {
    let connection = Connection::new().await.unwrap();
    let dht_service = connection.dht();

    let mut providers = dht_service.get_providers("capabilities", "camera").await.unwrap();

    while let Some(provider) = providers.try_next().await.unwrap() {
        println!("Found a node with a camera: {provider}");
    }
}

Typescript/JS

other_node.ts

import { Client } from 'hyveos-sdk';
import { Connection } from "hyveos-web";


async function main() {
    const transport = new Connection('http://localhost:8080');
    const client = new Client(transport);
    const dhtService = client.dht;

    const providers = await dhtService.getProviders('capabilities', 'camera')

    for await (const provider of providers) {
        console.log('Found a node with a camera: ${provider}');
    }

    // await transport.close()
}

main().catch(console.error);

Example 5. The node looks for nodes with a camera and prints them out.

Further Information

The DHT optimizes key-value storage and retrieval across large networks such as the ones you want to build. Query, put, get, and provide are standard operations supported. The DHT can be a leightweight data store or role discovery mechanism. The latter is needed when you want to give different nodes different roles and hence functionalities in the network, per scenario. The hyveOS DHT is based on the Kademlia DHT.