zc2 0.0.28

P2P compute broker with credit-based billing, WAL, and broker mesh support
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

#!/usr/bin/env python3
"""
Minimal Fibonacci worker for Zakuro broker benchmarks.

Listens for POST /execute with JSON body:
  - {"n": <int>} (default n=4) — compute fibonacci(n)
  - {"sleep": <secs>} (optional) — sleep this many seconds before responding (for long-run tests; max 300)
Returns a dictionary with:
  - result: fibonacci(n)
  - n: input n
  - pid: process ID of this worker
  - ip: this host's primary IP address

Compatible with the broker /execute flow: the broker forwards the request
body to the worker and returns the worker's response to the client.

Usage:
  python scripts/fibonacci_worker.py [--port PORT] [--bind ADDRESS]
  Default: bind 0.0.0.0:3960 (Zakuro worker port). Use --port 0 to pick any port.
"""

from __future__ import annotations

import argparse
import json
import os
import socket
import sys
from http.server import BaseHTTPRequestHandler, HTTPServer
from typing import Optional


def fibonacci(n: int) -> int:
    """Compute fibonacci(n) for non-negative n. fib(0)=0, fib(1)=1."""
    if n <= 0:
        return 0
    if n == 1:
        return 1
    a, b = 0, 1
    for _ in range(2, n + 1):
        a, b = b, a + b
    return b


def get_local_ip() -> str:
    """Best-effort primary IPv4 address for this host."""
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
            s.connect(("8.8.8.8", 80))
            return s.getsockname()[0]
    except Exception:
        pass
    try:
        return socket.gethostbyname(socket.gethostname())
    except Exception:
        return "127.0.0.1"


class FibonacciHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        if self.path.rstrip("/") != "/execute":
            self.send_response(404)
            self.end_headers()
            return
        content_length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_length) if content_length else b"{}"
        try:
            data = json.loads(body.decode("utf-8"))
            n = int(data.get("n", 4))
            sleep_secs = data.get("sleep")
            if sleep_secs is not None:
                sleep_secs = max(0, min(float(sleep_secs), 300))  # cap 0–300s for tests
            else:
                sleep_secs = 0
        except (ValueError, TypeError):
            n = 4
            sleep_secs = 0
        if sleep_secs > 0:
            import time
            time.sleep(sleep_secs)
        n = max(0, min(n, 50))  # clamp to avoid abuse
        result = fibonacci(n)
        payload = {
            "result": result,
            "n": n,
            "pid": os.getpid(),
            "ip": get_local_ip(),
        }
        if sleep_secs > 0:
            payload["sleep_secs"] = sleep_secs
        raw = json.dumps(payload).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.send_header("X-Zakuro-Pid", str(payload["pid"]))
        self.send_header("X-Zakuro-IP", str(payload["ip"]))
        self.end_headers()
        self.wfile.write(raw)

    def log_message(self, format: str, *args) -> None:
        # Optional: reduce noise
        pass


def main() -> None:
    ap = argparse.ArgumentParser(description="Fibonacci worker for Zakuro benchmarks")
    ap.add_argument("--port", "-p", type=int, default=3960, help="Port (0 = pick any)")
    ap.add_argument("--bind", "-b", default="0.0.0.0", help="Bind address")
    args = ap.parse_args()
    server = HTTPServer((args.bind, args.port), FibonacciHandler)
    port = server.server_port
    print(f"Fibonacci worker listening on {args.bind}:{port}  pid={os.getpid()}  ip={get_local_ip()}", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    server.server_close()


if __name__ == "__main__":
    main()