ravenclaw/modules
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Unify supercomputer mesh into one virtual machine via cluster-fabric

Browse Source 
Add hyper-p2p-cluster-fabric: aggregates every peer's CPU, RAM, GPU, disk,
and bandwidth into a single logical supercomputer (clusterId, asOneMachine).

- getVirtualMachine() — total vs available capacity across all nodes
- publishNode() — join the giant computer; syncs attached pool modules
- reserveCluster() — greedy multi-peer allocation for one workload
- runClusterJob() — reserve + fan-out jobs across slices
- Gossip cluster-node / cluster-reserve / cluster-run on Protomux

Enhance hyper-p2p-capacity-registry with clusterTotals() for the same
aggregate view at the registry layer.

Update SUPERCOMPUTER_LAYERS, category README, demo (3 racks → 44 cores,
155 GB RAM, 5 GPUs as one machine). Registry now 165 modules.

Co-authored-by: Cursor <[email protected]>
This commit is contained in:
Raven Scott
2026-05-21 00:41:57 -04:00
co-authored by Cursor
parent 693385bd7c
commit 03151a4cfc
17 changed files with 2442 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MODULE_CATEGORIES.md
+3 -2
View File
@@ -1,6 +1,6 @@
# Module categories (v2) # Module categories (v2)
164 modules in **30** category directories. All **production**. 165 modules in **30** category directories. All **production**.
Source of truth: [`MODULE_REGISTRY.yaml`](MODULE_REGISTRY.yaml). Source of truth: [`MODULE_REGISTRY.yaml`](MODULE_REGISTRY.yaml).
@@ -182,8 +182,9 @@ Source of truth: [`MODULE_REGISTRY.yaml`](MODULE_REGISTRY.yaml).
- 🟢 **hyper-p2p-peer-scheduler** — `peer-scheduler/v1` (production) - 🟢 **hyper-p2p-peer-scheduler** — `peer-scheduler/v1` (production)
- 🟢 **hyper-p2p-topic-lease** — `topic-lease/v1` (production) - 🟢 **hyper-p2p-topic-lease** — `topic-lease/v1` (production)
## Supercomputer (`supercomputer`) — 12 modules ## Supercomputer (`supercomputer`) — 13 modules
- 🟢 **hyper-p2p-cluster-fabric** — `cluster-fabric/v1` (production)
- 🟢 **hyper-p2p-bandwidth-share** — `bandwidth-share/v1` (production) - 🟢 **hyper-p2p-bandwidth-share** — `bandwidth-share/v1` (production)
- 🟢 **hyper-p2p-cache-farm** — `cache-farm/v1` (production) - 🟢 **hyper-p2p-cache-farm** — `cache-farm/v1` (production)
- 🟢 **hyper-p2p-capacity-registry** — `capacity-registry/v1` (production) - 🟢 **hyper-p2p-capacity-registry** — `capacity-registry/v1` (production)
MODULE_REGISTRY.yaml
+17 -1
View File
@@ -1,4 +1,4 @@
# Module registry — source of truth (164 modules) # Module registry — source of truth (165 modules)
# tiers: production | scaffold # tiers: production | scaffold
version: 1 version: 1
categories: { categories: {
@@ -1834,3 +1834,19 @@ modules:
composes_with: composes_with:
- hyper-p2p-job-dispatcher - hyper-p2p-job-dispatcher
- hyper-p2p-peer-scheduler - hyper-p2p-peer-scheduler
- name: hyper-p2p-cluster-fabric
category: supercomputer
tier: production
status: implemented
protocol: cluster-fabric/v1
class: HyperP2PClusterFabric
summary: Unifies all peer resources into one virtual supercomputer with cluster-wide reserve and run.
holepunch_refs: []
composes_with:
- hyper-p2p-capacity-registry
- hyper-p2p-cpu-share
- hyper-p2p-ram-pool
- hyper-p2p-gpu-slot
- hyper-p2p-job-dispatcher
- hyper-p2p-thermal-guard
- hyper-p2p-cluster-affinity
README.md
+3 -3
View File
@@ -1,6 +1,6 @@
# Hyper-P2P module workspace # Hyper-P2P module workspace
This directory is the **module library** for the repository: **164** independent npm packages for [Bare](https://github.com/holepunchto/bare) and Pear. Each package is a small, composable peer-to-peer primitive—discovery, messaging, storage, coordination, security, or application logic—built on shared Hyperswarm and Protomux conventions. This directory is the **module library** for the repository: **165** independent npm packages for [Bare](https://github.com/holepunchto/bare) and Pear. Each package is a small, composable peer-to-peer primitive—discovery, messaging, storage, coordination, security, or application logic—built on shared Hyperswarm and Protomux conventions.
The parent repository adds cross-cutting guides under `docs/`, composition examples under `examples/`, and the unified test runner under `real_tests/`. Start at the [repository README](../README.md) for clone, CI, and top-level commands. The parent repository adds cross-cutting guides under `docs/`, composition examples under `examples/`, and the unified test runner under `real_tests/`. Start at the [repository README](../README.md) for clone, CI, and top-level commands.
@@ -44,7 +44,7 @@ Packages live under **31 functional categories** (one folder per category). Cate
| Consensus & coordination | 5 | Locks, leases, quorum, Raft-lite, causal consensus | | Consensus & coordination | 5 | Locks, leases, quorum, Raft-lite, causal consensus |
| Scheduling & queues | 5 | Activity queue, peer scheduler, topic lease, cron, deadlines | | Scheduling & queues | 5 | Activity queue, peer scheduler, topic lease, cron, deadlines |
| Measurement & rate control | 4 | Token buckets, histograms, sketches, SLA budgets | | Measurement & rate control | 4 | Token buckets, histograms, sketches, SLA budgets |
| Supercomputer | 12 | CPU/RAM/GPU/bandwidth/disk pools, jobs, cache, gateway, affinity | | Supercomputer | 13 | Cluster fabric (one VM), CPU/RAM/GPU/bandwidth/disk pools, jobs |
| Indexes & search | 8 | Bloom, fulltext, graph, inverted, vector, LSH, trie, spatial | | Indexes & search | 8 | Bloom, fulltext, graph, inverted, vector, LSH, trie, spatial |
| Storage (Hypercore) | 7 | Replication, fetch, forks, merkle, audit, bitfield, seed policy | | Storage (Hypercore) | 7 | Replication, fetch, forks, merkle, audit, bitfield, seed policy |
| Storage (Hyperbee) | 5 | Batch write, diff follow, range watch, secondary index, tombstones | | Storage (Hyperbee) | 5 | Batch write, diff follow, range watch, secondary index, tombstones |
@@ -59,7 +59,7 @@ Packages live under **31 functional categories** (one folder per category). Cate
| Oracles | 1 | Decentralized oracle gossip | | Oracles | 1 | Decentralized oracle gossip |
| Time & ordering | 1 | Temporal index | | Time & ordering | 1 | Temporal index |
**164** modules total. The authoritative list with protocol ids and paths is in **`MODULE_REGISTRY.yaml`**; **`module-paths.json`** maps package name → category path for tooling and tests. **165** modules total. The authoritative list with protocol ids and paths is in **`MODULE_REGISTRY.yaml`**; **`module-paths.json`** maps package name → category path for tooling and tests.
For a per-module index with protocol ids, use **`MODULE_CATEGORIES.md`** in this directory. For a per-module index with protocol ids, use **`MODULE_CATEGORIES.md`** in this directory.
_shared/SUPERCOMPUTER_LAYERS.md
+15 -2
View File
@@ -1,8 +1,13 @@
# Supercomputer resource layers # Supercomputer resource layers
P2P modules for pooling compute, memory, network, and storage across a Hyperswarm topic. P2P modules for pooling compute, memory, network, and storage across a Hyperswarm topic — **combined into one virtual machine** via `hyper-p2p-cluster-fabric`.
```text ```text
┌─────────────────────────────────────────────────────┐
│ hyper-p2p-cluster-fabric (ONE giant computer) │
│ getVirtualMachine() · reserveCluster() · runClusterJob() │
└──────────────────────────┬──────────────────────────┘
│
Job / workload (job-dispatcher, work-stealer) Job / workload (job-dispatcher, work-stealer)
↓ ↓
Placement (cluster-affinity, capacity-registry, thermal-guard) Placement (cluster-affinity, capacity-registry, thermal-guard)
@@ -16,7 +21,8 @@ P2P modules for pooling compute, memory, network, and storage across a Hyperswar
| Layer | Modules | | Layer | Modules |
|-------|---------| |-------|---------|
| Registry | `hyper-p2p-capacity-registry` | | **Unified fabric** | `hyper-p2p-cluster-fabric` |
| Registry | `hyper-p2p-capacity-registry` (`clusterTotals()`) |
| Compute | `hyper-p2p-cpu-share`, `hyper-p2p-gpu-slot` | | Compute | `hyper-p2p-cpu-share`, `hyper-p2p-gpu-slot` |
| Memory | `hyper-p2p-ram-pool`, `hyper-p2p-cache-farm` | | Memory | `hyper-p2p-ram-pool`, `hyper-p2p-cache-farm` |
| Network | `hyper-p2p-bandwidth-share`, `hyper-p2p-net-gateway` | | Network | `hyper-p2p-bandwidth-share`, `hyper-p2p-net-gateway` |
@@ -25,4 +31,11 @@ P2P modules for pooling compute, memory, network, and storage across a Hyperswar
| Safety | `hyper-p2p-thermal-guard` | | Safety | `hyper-p2p-thermal-guard` |
| Placement | `hyper-p2p-cluster-affinity` | | Placement | `hyper-p2p-cluster-affinity` |
## One machine mental model
1. Each peer calls `fabric.publishNode({ cpuCores, ramMb, … })` to join the cluster.
2. `fabric.getVirtualMachine()` returns **summed** totals — e.g. 44 cores, 156 GB RAM, 5 GPUs from 3 racks.
3. `fabric.runClusterJob({ cpuMs, ramMb, gpuSlots })` **splits** work across nodes automatically.
4. `capacity-registry.clusterTotals()` provides the same aggregate view at the registry layer.
Pairs with `hyper-p2p-bandwidth-broker`, `hyper-p2p-topic-lease`, and `hyper-p2p-quorum-pool`. Pairs with `hyper-p2p-bandwidth-broker`, `hyper-p2p-topic-lease`, and `hyper-p2p-quorum-pool`.
module-paths.json
+1
View File
@@ -30,6 +30,7 @@
"hyper-p2p-circuit-breaker": "routing-advanced/hyper-p2p-circuit-breaker", "hyper-p2p-circuit-breaker": "routing-advanced/hyper-p2p-circuit-breaker",
"hyper-p2p-circuit-loom": "network-stack/hyper-p2p-circuit-loom", "hyper-p2p-circuit-loom": "network-stack/hyper-p2p-circuit-loom",
"hyper-p2p-cluster-affinity": "supercomputer/hyper-p2p-cluster-affinity", "hyper-p2p-cluster-affinity": "supercomputer/hyper-p2p-cluster-affinity",
"hyper-p2p-cluster-fabric": "supercomputer/hyper-p2p-cluster-fabric",
"hyper-p2p-collab-room": "applications-collab/hyper-p2p-collab-room", "hyper-p2p-collab-room": "applications-collab/hyper-p2p-collab-room",
"hyper-p2p-compact-codec-bridge": "encoding-wire/hyper-p2p-compact-codec-bridge", "hyper-p2p-compact-codec-bridge": "encoding-wire/hyper-p2p-compact-codec-bridge",
"hyper-p2p-conflict-set": "state-crdts/hyper-p2p-conflict-set", "hyper-p2p-conflict-set": "state-crdts/hyper-p2p-conflict-set",
supercomputer/README.md
+20 -14
View File
@@ -1,8 +1,8 @@
# Supercomputer # Supercomputer
**Path:** `modules/supercomputer/` · **Modules:** 12 (all production) **Path:** `modules/supercomputer/` · **Modules:** 13 (all production)
Distributed **resource pooling** over Hyperswarm: advertise capacity, share CPU/RAM/GPU/bandwidth/disk, dispatch jobs, route internet egress, and place workloads by hardware affinity. Hub: [`../../docs/supercomputer/README.md`](../../docs/supercomputer/README.md). Layering: [`../_shared/SUPERCOMPUTER_LAYERS.md`](../_shared/SUPERCOMPUTER_LAYERS.md). Distributed **resource pooling** over Hyperswarm — combined into **one virtual supercomputer** via [`hyper-p2p-cluster-fabric`](./hyper-p2p-cluster-fabric/). Hub: [`../../docs/supercomputer/README.md`](../../docs/supercomputer/README.md). Layering: [`../_shared/SUPERCOMPUTER_LAYERS.md`](../_shared/SUPERCOMPUTER_LAYERS.md).
## When to use ## When to use
@@ -12,33 +12,39 @@ Building a **volunteer or federated compute mesh** on one topic: peers donate id
Single-machine orchestration (use local process APIs). Untrusted open networks without `trust-security` and rate limits. Single-machine orchestration (use local process APIs). Untrusted open networks without `trust-security` and rate limits.
## Quick start ## Quick start — one giant computer
```bash ```bash
bare ../../examples/demo-supercomputer-mesh/index.js bare ../../examples/demo-supercomputer-mesh/index.js
``` ```
```js ```js
const { HyperP2PCapacityRegistry } = require('hyper-p2p-capacity-registry') const { HyperP2PClusterFabric } = require('hyper-p2p-cluster-fabric')
const { HyperP2PCpuShare } = require('hyper-p2p-cpu-share') const fabric = new HyperP2PClusterFabric({ topic: 'super-mesh' })
const { HyperP2PJobDispatcher } = require('hyper-p2p-job-dispatcher') await fabric.ready()
const reg = new HyperP2PCapacityRegistry({ topic: 'super-mesh' }) fabric.publishNode({ cpuCores: 8, ramMb: 16384, gpuSlots: 1, cpuMsDonated: 3_600_000 })
const cpu = new HyperP2PCpuShare({ topic: 'super-mesh' }) fabric.publishNode({ cpuCores: 32, ramMb: 131072, gpuSlots: 4, cpuMsDonated: 14_400_000 }, 'rack-b')
const jobs = new HyperP2PJobDispatcher({ topic: 'super-mesh' })
await reg.ready() const vm = fabric.getVirtualMachine()
reg.advertise({ cpuCores: 8, ramMb: 16384, gpuSlots: 1, upMbps: 100 }) // vm.total.cpuCores === 40, vm.nodes === 2, vm.asOneMachine === true
cpu.donate(reg.peerId, 60_000)
const job = jobs.submitJob({ kind: 'render', cpuMs: 5000, ramMb: 512 }) fabric.runClusterJob({ kind: 'train', cpuMs: 500_000, ramMb: 8192, gpuSlots: 2 })
``` ```
## Modules by layer ## Modules by layer
### Unified fabric (start here)
| Module | Protocol | Role |
|--------|----------|------|
| [hyper-p2p-cluster-fabric](./hyper-p2p-cluster-fabric/) | `cluster-fabric/v1` | **One virtual supercomputer** — `getVirtualMachine()`, `reserveCluster()`, `runClusterJob()` |
### Registry & placement ### Registry & placement
| Module | Protocol | Role | | Module | Protocol | Role |
|--------|----------|------| |--------|----------|------|
| [hyper-p2p-capacity-registry](./hyper-p2p-capacity-registry/) | `capacity-registry/v1` | Gossip CPU/RAM/GPU/disk/bandwidth ads | | [hyper-p2p-capacity-registry](./hyper-p2p-capacity-registry/) | `capacity-registry/v1` | Gossip ads + `clusterTotals()` aggregate |
| [hyper-p2p-cluster-affinity](./hyper-p2p-cluster-affinity/) | `cluster-affinity/v1` | Tag + latency scoring for placement | | [hyper-p2p-cluster-affinity](./hyper-p2p-cluster-affinity/) | `cluster-affinity/v1` | Tag + latency scoring for placement |
| [hyper-p2p-thermal-guard](./hyper-p2p-thermal-guard/) | `thermal-guard/v1` | CPU/RAM/temp throttle signals | | [hyper-p2p-thermal-guard](./hyper-p2p-thermal-guard/) | `thermal-guard/v1` | CPU/RAM/temp throttle signals |
supercomputer/hyper-p2p-capacity-registry/index.js
+25
View File
@@ -47,6 +47,31 @@ class HyperP2PCapacityRegistry extends EventEmitter {
listPeers () { return [...this._peers.values()] } listPeers () { return [...this._peers.values()] }
/**
* Sum every peer into one logical supercomputer capacity view.
*/
clusterTotals () {
const t = {
nodes: 0,
cpuCores: 0,
ramMb: 0,
diskGb: 0,
gpuSlots: 0,
upMbps: 0,
downMbps: 0
}
for (const rec of this._peers.values()) {
t.nodes++
t.cpuCores += rec.cpuCores
t.ramMb += rec.ramMb
t.diskGb += rec.diskGb
t.gpuSlots += rec.gpuSlots
t.upMbps += rec.upMbps
t.downMbps += rec.downMbps
}
return { ...t, asOneMachine: true }
}
bestMatch (requirements = {}) { bestMatch (requirements = {}) {
let best = null let best = null
let bestScore = -1 let bestScore = -1
supercomputer/hyper-p2p-capacity-registry/test/test.js
+11
View File
@@ -11,6 +11,17 @@ test('capacity-registry: advertise and match', async (t) => {
await r.close() await r.close()
}) })
test('capacity-registry: clusterTotals', async (t) => {
const r = new HyperP2PCapacityRegistry()
r.advertise({ cpuCores: 4, ramMb: 8192 })
r.advertise({ cpuCores: 8, ramMb: 16384 }, 'peer-b')
const t0 = r.clusterTotals()
t.is(t0.nodes, 2)
t.is(t0.cpuCores, 12)
t.ok(t0.asOneMachine)
await r.close()
})
test('capacity-registry: getStats', async (t) => { test('capacity-registry: getStats', async (t) => {
const r = new HyperP2PCapacityRegistry() const r = new HyperP2PCapacityRegistry()
t.is(r.getStats().protocol, 'capacity-registry/v1') t.is(r.getStats().protocol, 'capacity-registry/v1')
supercomputer/hyper-p2p-cluster-fabric/CHANGELOG.md
+5
View File
@@ -0,0 +1,5 @@
# Changelog
## 0.3.0
- Initial release: unified virtual supercomputer (`cluster-fabric/v1`).
supercomputer/hyper-p2p-cluster-fabric/README.md
+54
View File
@@ -0,0 +1,54 @@
# hyper-p2p-cluster-fabric
**The unified supercomputer** — combines every peer on a topic into **one logical machine** with summed CPU, RAM, GPU, disk, and bandwidth.
**Category:** Supercomputer · **Protocol:** `cluster-fabric/v1` · **Export:** `HyperP2PClusterFabric`
## When to use
You want the mesh to behave as a **single giant computer**, not isolated pools. Call `getVirtualMachine()` for total capacity, `reserveCluster()` to allocate across nodes, and `runClusterJob()` to execute on the combined fabric.
## Quick start
```js
const { HyperP2PClusterFabric } = require('hyper-p2p-cluster-fabric')
const { HyperP2PCapacityRegistry } = require('hyper-p2p-capacity-registry')
const { HyperP2PCpuShare } = require('hyper-p2p-cpu-share')
const { HyperP2PJobDispatcher } = require('hyper-p2p-job-dispatcher')
const fabric = new HyperP2PClusterFabric({
topic: 'my-supercomputer',
registry: new HyperP2PCapacityRegistry({ topic: 'my-supercomputer' }),
cpu: new HyperP2PCpuShare({ topic: 'my-supercomputer' }),
jobs: new HyperP2PJobDispatcher({ topic: 'my-supercomputer' })
})
await fabric.ready()
fabric.publishNode({ cpuCores: 8, ramMb: 16384, gpuSlots: 1, cpuMsDonated: 60_000 })
const vm = fabric.getVirtualMachine()
console.log(vm.total.cpuCores, 'cores across', vm.nodes, 'nodes')
fabric.runClusterJob({ kind: 'train', cpuMs: 100_000, ramMb: 8192 })
```
## Key APIs
| Method | Purpose |
|--------|---------|
| `publishNode(resources)` | Join the giant computer; syncs attached pools |
| `getVirtualMachine()` | `{ total, available, nodes, asOneMachine: true }` |
| `clusterTotals()` | Summed capacity of all nodes |
| `availableTotals()` | Free capacity after reservations |
| `reserveCluster(spec)` | Multi-peer greedy allocation |
| `runClusterJob(spec)` | Reserve + fan-out jobs to slices |
## Docs
- [docs/api.md](docs/api.md)
- [docs/architecture.md](docs/architecture.md)
## Test
```bash
npm install && npm test
```
supercomputer/hyper-p2p-cluster-fabric/docs/api.md
+52
View File
@@ -0,0 +1,52 @@
# API: hyper-p2p-cluster-fabric
**Protocol:** `cluster-fabric/v1` · **Export:** `HyperP2PClusterFabric`
## Overview
Top-level **virtual supercomputer** facade. Every peer that `publishNode()` becomes part of one machine identified by `clusterId`. Workloads draw from **combined** CPU/RAM/GPU/disk across the mesh.
## Constructor
```js
const fabric = new HyperP2PClusterFabric({
topic,
clusterId, // optional; derived from topic hash
registry, // optional HyperP2PCapacityRegistry
cpu, ram, gpu, jobs, thermal, affinity, bandwidth, disk // optional attached modules
})
```
## Methods
### `publishNode(resources, peerId?) → nodeRecord`
Registers a node on the fabric; forwards to attached modules (`advertise`, `donate`, `lend`, `offer`, etc.).
### `getVirtualMachine() → { clusterId, name, nodes, total, available, asOneMachine }`
Single view of the **entire** mesh as one computer.
### `clusterTotals()` / `availableTotals()`
Aggregated capacity and remaining free resources.
### `reserveCluster({ cpuMs, ramMb, gpuSlots, diskGb }) → reservation`
Greedy placement across peers; returns `{ reservationId, slices: [{ peerId, cpuMs, ramMb, ... }] }`.
### `releaseReservation(reservationId) → boolean`
### `runClusterJob(spec) → { reservationId, slices, jobIds }`
Calls `reserveCluster` then `jobs.submitJob` per slice when `jobs` is attached.
### `listNodes()` / `mergeCluster(remote)` / `getStats()` / `ready()` / `close()`
## Events
`node`, `reserve`, `release`, `run`, `merge`, `closed`
## Composition
Sits above all other `supercomputer/*` modules. See [`../../_shared/SUPERCOMPUTER_LAYERS.md`](../../_shared/SUPERCOMPUTER_LAYERS.md).
supercomputer/hyper-p2p-cluster-fabric/docs/architecture.md
+29
View File
@@ -0,0 +1,29 @@
# Architecture: hyper-p2p-cluster-fabric
```text
┌─────────────────────────────┐
│ HyperP2PClusterFabric │
│ (one virtual machine) │
└──────────────┬──────────────┘
┌──────────┼──────────┬──────────┬──────────┐
▼ ▼ ▼ ▼ ▼
capacity cpu-share ram-pool gpu-slot job-dispatcher
registry ...
└──────────┴──────────┴──────────┴──────────┘
Hyperswarm topic
```
## Wire messages
| type | Purpose |
|------|---------|
| `cluster-node` | Gossip node capacity into fabric view |
| `cluster-reserve` | Shared reservation across mesh |
| `cluster-release` | Release reservation |
| `cluster-run` | Announced cluster job fan-out |
## State
- `_nodes` — per-peer contributed capacity + usage
- `_reservations` — active multi-peer allocations
- `_used` — cluster-wide reserved totals
supercomputer/hyper-p2p-cluster-fabric/examples/basic.js
+11
View File
@@ -0,0 +1,11 @@
require('bare-process/global')
const { HyperP2PClusterFabric } = require('../index.js')
async function main () {
const fabric = new HyperP2PClusterFabric()
fabric.publishNode({ cpuCores: 8, ramMb: 16384, gpuSlots: 1, cpuMsDonated: 50_000 })
console.log('virtual machine', fabric.getVirtualMachine())
await fabric.close()
console.log('done')
}
main().catch(console.error)
supercomputer/hyper-p2p-cluster-fabric/index.js
+347
View File
@@ -0,0 +1,347 @@
require('bare-process/global')
const EventEmitter = require('bare-events')
const b4a = require('b4a')
const { initModuleSwarm, gossipSend } = require('../../_shared/p2p-bare.js')
const PROTOCOL = 'cluster-fabric/v1'
/**
* Unifies every peer on a topic into one logical supercomputer.
* Aggregates CPU, RAM, GPU, disk, and bandwidth; reserves and runs workloads
* across multiple nodes as a single cluster allocation.
*/
class HyperP2PClusterFabric extends EventEmitter {
constructor (opts = {}) {
super()
this._stats = { publishes: 0, reservations: 0, jobs: 0 }
this.topic = opts.topic || null
this.keyPair = opts.keyPair || require('hypercore-crypto').keyPair()
this.peerId = b4a.toString(this.keyPair.publicKey, 'hex')
this.clusterId = opts.clusterId || b4a.toString(
require('hypercore-crypto').hash(b4a.from(String(this.topic || 'local-cluster'))),
'hex'
).slice(0, 16)
this.registry = opts.registry || null
this.cpu = opts.cpu || null
this.ram = opts.ram || null
this.gpu = opts.gpu || null
this.jobs = opts.jobs || null
this.thermal = opts.thermal || null
this.affinity = opts.affinity || null
this.bandwidth = opts.bandwidth || null
this.disk = opts.disk || null
this._nodes = new Map()
this._reservations = new Map()
this._used = { cpuMs: 0, ramMb: 0, gpuSlots: 0, diskGb: 0 }
this.swarm = null
this._peerMsgs = null
}
_pid (peerId) {
return typeof peerId === 'string' ? peerId : b4a.toString(peerId, 'hex')
}
_nodeRecord (peerId, resources = {}) {
return {
peerId: this._pid(peerId),
cpuCores: Math.max(0, resources.cpuCores ?? 0),
ramMb: Math.max(0, resources.ramMb ?? 0),
diskGb: Math.max(0, resources.diskGb ?? 0),
gpuSlots: Math.max(0, resources.gpuSlots ?? 0),
upMbps: Math.max(0, resources.upMbps ?? 0),
downMbps: Math.max(0, resources.downMbps ?? 0),
cpuMsDonated: Math.max(0, resources.cpuMsDonated ?? 0),
cpuMsUsed: Math.max(0, resources.cpuMsUsed ?? 0),
ramMbLent: Math.max(0, resources.ramMbLent ?? 0),
ramMbUsed: Math.max(0, resources.ramMbUsed ?? 0),
tags: resources.tags || [],
at: Date.now()
}
}
/**
* Join the mesh as one node of the giant computer — advertises capacity
* and optionally syncs into attached pool modules.
*/
publishNode (resources = {}, peerId = null) {
const rec = this._nodeRecord(peerId || this.peerId, resources)
this._nodes.set(rec.peerId, rec)
this._stats.publishes++
if (this.registry) this.registry.advertise(resources, rec.peerId)
if (this.cpu && resources.cpuMsDonated) this.cpu.donate(rec.peerId, resources.cpuMsDonated)
if (this.ram && resources.ramMbLent) this.ram.lend(rec.peerId, resources.ramMbLent * 1024 * 1024)
if (this.gpu && resources.gpuSlots) {
for (let i = 0; i < resources.gpuSlots; i++) {
this.gpu.registerSlot(`${rec.peerId}-gpu${i}`, { vramMb: resources.vramMb ?? 8192, tags: resources.tags })
}
}
if (this.bandwidth && (resources.upMbps || resources.downMbps)) {
this.bandwidth.offer(rec.peerId, resources.upMbps ?? 0, resources.downMbps ?? 0)
}
if (this.affinity) this.affinity.tagPeer(rec.peerId, resources.tags || [], resources.latencyMs ?? null)
if (this._peerMsgs) gossipSend(this, { type: 'cluster-node', clusterId: this.clusterId, node: rec })
this.emit('node', rec)
return rec
}
_syncFromRegistry () {
if (!this.registry) return
for (const rec of this.registry.listPeers()) {
const cur = this._nodes.get(rec.peerId)
if (!cur || rec.at > cur.at) {
this._nodes.set(rec.peerId, this._nodeRecord(rec.peerId, {
cpuCores: rec.cpuCores,
ramMb: rec.ramMb,
diskGb: rec.diskGb,
gpuSlots: rec.gpuSlots,
upMbps: rec.upMbps,
downMbps: rec.downMbps,
tags: rec.tags,
at: rec.at
}))
}
}
}
/**
* One virtual machine: sum of every node's contributed resources.
*/
clusterTotals () {
this._syncFromRegistry()
const t = {
clusterId: this.clusterId,
nodes: 0,
cpuCores: 0,
ramMb: 0,
diskGb: 0,
gpuSlots: 0,
upMbps: 0,
downMbps: 0,
cpuMsPool: 0,
cpuMsUsed: 0,
ramMbPool: 0,
ramMbUsed: 0
}
for (const n of this._nodes.values()) {
if (this.thermal && this.thermal.shouldThrottle(n.peerId)) continue
t.nodes++
t.cpuCores += n.cpuCores
t.ramMb += n.ramMb
t.diskGb += n.diskGb
t.gpuSlots += n.gpuSlots
t.upMbps += n.upMbps
t.downMbps += n.downMbps
t.cpuMsPool += n.cpuMsDonated
t.cpuMsUsed += n.cpuMsUsed
t.ramMbPool += n.ramMbLent
t.ramMbUsed += n.ramMbUsed
}
if (this.cpu) {
let pool = 0
for (const id of this.cpu.listPeerIds()) pool += this.cpu.balance(id)
t.cpuMsPool = Math.max(t.cpuMsPool, pool)
}
return t
}
/**
* What the unified supercomputer still has free (cluster minus reservations).
*/
availableTotals () {
const t = this.clusterTotals()
return {
...t,
cpuCores: Math.max(0, t.cpuCores - Math.ceil(this._used.cpuMs / 1000)),
ramMb: Math.max(0, t.ramMb - this._used.ramMb),
gpuSlots: Math.max(0, t.gpuSlots - this._used.gpuSlots),
diskGb: Math.max(0, t.diskGb - this._used.diskGb),
cpuMs: Math.max(0, t.cpuMsPool - t.cpuMsUsed - this._used.cpuMs),
ramMbFree: Math.max(0, t.ramMbPool - t.ramMbUsed - this._used.ramMb)
}
}
/**
* Greedy multi-peer reservation — one workload drawn from the combined cluster.
*/
reserveCluster (spec = {}) {
const need = {
cpuMs: Math.max(0, spec.cpuMs ?? 0),
ramMb: Math.max(0, spec.ramMb ?? 0),
gpuSlots: Math.max(0, spec.gpuSlots ?? 0),
diskGb: Math.max(0, spec.diskGb ?? 0)
}
const avail = this.availableTotals()
if (need.cpuMs > avail.cpuMs) throw new Error('cluster insufficient cpuMs')
if (need.ramMb > avail.ramMb && need.ramMb > avail.ramMbFree) throw new Error('cluster insufficient ramMb')
if (need.gpuSlots > avail.gpuSlots) throw new Error('cluster insufficient gpuSlots')
if (need.diskGb > avail.diskGb) throw new Error('cluster insufficient diskGb')
const ranked = [...this._nodes.values()]
.filter((n) => !this.thermal || !this.thermal.shouldThrottle(n.peerId))
.sort((a, b) => (b.cpuCores + b.ramMb / 512) - (a.cpuCores + a.ramMb / 512))
const slices = []
let left = { ...need }
for (const n of ranked) {
if (left.cpuMs <= 0 && left.ramMb <= 0 && left.gpuSlots <= 0 && left.diskGb <= 0) break
const slice = {
peerId: n.peerId,
cpuMs: Math.min(left.cpuMs, n.cpuMsDonated || n.cpuCores * 60_000),
ramMb: Math.min(left.ramMb, n.ramMb - n.ramMbUsed),
gpuSlots: Math.min(left.gpuSlots, n.gpuSlots),
diskGb: Math.min(left.diskGb, n.diskGb)
}
if (slice.cpuMs <= 0 && slice.ramMb <= 0 && slice.gpuSlots <= 0 && slice.diskGb <= 0) continue
if (this.cpu && slice.cpuMs) this.cpu.consume(slice.peerId, slice.cpuMs)
left.cpuMs -= slice.cpuMs
left.ramMb -= slice.ramMb
left.gpuSlots -= slice.gpuSlots
left.diskGb -= slice.diskGb
n.cpuMsUsed += slice.cpuMs
n.ramMbUsed += slice.ramMb
slices.push(slice)
}
if (left.cpuMs > 0 || left.ramMb > 0 || left.gpuSlots > 0 || left.diskGb > 0) {
throw new Error('could not place reservation across cluster')
}
const reservationId = b4a.toString(
require('hypercore-crypto').hash(b4a.from(JSON.stringify(need) + Date.now())),
'hex'
).slice(0, 16)
const reservation = { reservationId, clusterId: this.clusterId, spec: need, slices, at: Date.now() }
this._reservations.set(reservationId, reservation)
this._used.cpuMs += need.cpuMs
this._used.ramMb += need.ramMb
this._used.gpuSlots += need.gpuSlots
this._used.diskGb += need.diskGb
this._stats.reservations++
if (this._peerMsgs) gossipSend(this, { type: 'cluster-reserve', reservation })
this.emit('reserve', reservation)
return reservation
}
releaseReservation (reservationId) {
const r = this._reservations.get(reservationId)
if (!r) return false
for (const s of r.slices) {
const n = this._nodes.get(s.peerId)
if (n) {
n.cpuMsUsed = Math.max(0, n.cpuMsUsed - s.cpuMs)
n.ramMbUsed = Math.max(0, n.ramMbUsed - s.ramMb)
}
}
this._used.cpuMs = Math.max(0, this._used.cpuMs - r.spec.cpuMs)
this._used.ramMb = Math.max(0, this._used.ramMb - r.spec.ramMb)
this._used.gpuSlots = Math.max(0, this._used.gpuSlots - r.spec.gpuSlots)
this._used.diskGb = Math.max(0, this._used.diskGb - r.spec.diskGb)
this._reservations.delete(reservationId)
if (this._peerMsgs) gossipSend(this, { type: 'cluster-release', reservationId })
this.emit('release', { reservationId })
return true
}
/**
* Run one logical job on the combined supercomputer (may fan out to N peers).
*/
runClusterJob (spec = {}) {
const reservation = this.reserveCluster(spec)
const jobIds = []
if (this.jobs) {
for (const slice of reservation.slices) {
const job = this.jobs.submitJob({
...spec,
kind: spec.kind || 'cluster',
targetPeer: slice.peerId,
cpuMs: slice.cpuMs,
ramMb: slice.ramMb,
gpuSlots: slice.gpuSlots,
clusterId: this.clusterId,
reservationId: reservation.reservationId
})
jobIds.push(job.jobId)
}
}
this._stats.jobs++
const run = { reservationId: reservation.reservationId, clusterId: this.clusterId, slices: reservation.slices, jobIds }
if (this._peerMsgs) gossipSend(this, { type: 'cluster-run', run })
this.emit('run', run)
return run
}
getVirtualMachine () {
const t = this.clusterTotals()
const a = this.availableTotals()
return {
clusterId: this.clusterId,
name: `HyperP2P-Cluster-${this.clusterId}`,
nodes: t.nodes,
total: t,
available: a,
asOneMachine: true
}
}
listNodes () { return [...this._nodes.values()] }
mergeCluster (remote) {
if (!remote || !remote.nodes) return 0
let n = 0
for (const node of remote.nodes) {
const cur = this._nodes.get(node.peerId)
if (!cur || node.at > cur.at) {
this._nodes.set(node.peerId, node)
n++
}
}
if (n) this.emit('merge', { updated: n })
return n
}
async ready () {
if (this.swarm || !this.topic) return this
await initModuleSwarm(this, {
keyPair: this.keyPair,
topic: this.topic,
protocol: PROTOCOL,
onmessage: (data) => {
if (data?.type === 'cluster-node' && data.node) {
if (data.clusterId === this.clusterId) this._nodes.set(data.node.peerId, data.node)
} else if (data?.type === 'cluster-reserve' && data.reservation) {
this._reservations.set(data.reservation.reservationId, data.reservation)
} else if (data?.type === 'cluster-release') {
this._reservations.delete(data.reservationId)
}
}
})
return this
}
getStats () {
const t = this.clusterTotals()
return {
...this._stats,
clusterId: this.clusterId,
nodes: t.nodes,
reservations: this._reservations.size,
protocol: PROTOCOL
}
}
async close () {
if (this.swarm) await this.swarm.destroy().catch(() => {})
this.swarm = null
this._nodes.clear()
this._reservations.clear()
this.emit('closed')
}
}
module.exports = { HyperP2PClusterFabric, PROTOCOL }
supercomputer/hyper-p2p-cluster-fabric/package-lock.json
Generated
+1774
View File
File diff suppressed because it is too large Load Diff
supercomputer/hyper-p2p-cluster-fabric/package.json
+23
View File
@@ -0,0 +1,23 @@
{
"name": "hyper-p2p-cluster-fabric",
"version": "0.3.0",
"description": "Unified virtual supercomputer registry for Bare/Pear P2P supercomputer mesh.",
"main": "index.js",
"type": "commonjs",
"license": "Apache-2.0",
"scripts": { "test": "brittle-bare test/test.js" },
"dependencies": {
"bare-events": "^2.8.0",
"bare-process": "^4.4.0",
"b4a": "^1.6.7",
"hypercore-crypto": "^3.0.0",
"protomux": "^3.0.0",
"compact-encoding": "^2.0.0"
},
"peerDependencies": { "hyperswarm": "^4.0.0", "bare": ">=1.0.0" },
"devDependencies": { "brittle": "^3.0.0" },
"imports": {
"process": { "bare": "bare-process", "default": "process" },
"events": { "bare": "bare-events", "default": "events" }
}
}
supercomputer/hyper-p2p-cluster-fabric/test/test.js
+52
View File
@@ -0,0 +1,52 @@
require('bare-process/global')
const test = require('brittle')
const { HyperP2PClusterFabric } = require('../index.js')
const { HyperP2PCapacityRegistry } = require('../../hyper-p2p-capacity-registry/index.js')
const { HyperP2PCpuShare } = require('../../hyper-p2p-cpu-share/index.js')
const { HyperP2PJobDispatcher } = require('../../hyper-p2p-job-dispatcher/index.js')
test('cluster-fabric: combines nodes into one VM', async (t) => {
const reg = new HyperP2PCapacityRegistry()
const cpu = new HyperP2PCpuShare()
const jobs = new HyperP2PJobDispatcher()
const fabric = new HyperP2PClusterFabric({ registry: reg, cpu, jobs })
fabric.publishNode({ cpuCores: 8, ramMb: 16384, gpuSlots: 1, cpuMsDonated: 60_000 }, 'node-a')
fabric.publishNode({ cpuCores: 16, ramMb: 32768, gpuSlots: 2, cpuMsDonated: 120_000 }, 'node-b')
const vm = fabric.getVirtualMachine()
t.is(vm.asOneMachine, true)
t.is(vm.nodes, 2)
t.is(vm.total.cpuCores, 24)
t.is(vm.total.ramMb, 49152)
t.is(vm.total.gpuSlots, 3)
const run = fabric.runClusterJob({ kind: 'render', cpuMs: 5000, ramMb: 4096 })
t.ok(run.jobIds.length >= 1)
t.ok(run.slices.length >= 1)
await fabric.close()
await reg.close()
await cpu.close()
await jobs.close()
})
test('cluster-fabric: clusterTotals standalone', async (t) => {
const f = new HyperP2PClusterFabric()
f.publishNode({ cpuCores: 4, ramMb: 8192 })
f.publishNode({ cpuCores: 4, ramMb: 8192 }, 'peer-2')
t.is(f.clusterTotals().cpuCores, 8)
await f.close()
})
test('cluster-fabric: validation', async (t) => {
const f = new HyperP2PClusterFabric()
f.publishNode({ cpuCores: 1, ramMb: 512, cpuMsDonated: 100 })
try {
f.reserveCluster({ cpuMs: 999_999_999 })
t.fail('expected throw')
} catch (e) {
t.ok(e instanceof Error)
}
await f.close()
})