"""tpc_volunteer 0.1.0 (DRAFT, pending the pause of landing round 8): the
VOLUNTEER channel's law of D126 as PMP-TPM 0.12.0 words it (VOLUNTEER
LAW, GOOD STANDING, the tuning options EARLY CALL LIMIT through
VOLUNTEER SHARE) with D155's FALCON answer - GOOD STANDING, the token
bucket, RETRY AFTER with full jitter, INTAKE's EWMA and the random early
throttle p_nw, the p_v epoch, the draw itself, and the VOLUNTEER SHARE
cap - as pure functions of recorded state and a supplied random value.
Reference implementation, Python standard library only; imports
tpc_probe (uniform, udiscrete) and tpc_beta (canon_float). DATA for
landing round 8 of branch tpm_project_closeout; not a declared TPC-PY
version and not TPX code. Every draw derives its uniform values from r
(32 lowercase hex) by the D97 digest of [tag, "1", r, counter]. Declared
parameters and tuning options are read from params under PMP-TPM
0.12.0's names: VOLUNTEER TIER; VOLUNTEER RATE (tokens per hour,
wall-clock); VOLUNTEER BURST; BACKOFF BASE and BACKOFF CAP (seconds,
wall-clock); INTAKE w_q; INTAKE MIN; INTAKE MAX; INTAKE MAX_P; P_V
BOUNDS; P_V ROTATION; EARLY CALL LIMIT; VOLUNTEER COOL-OFF; VOLUNTEER
SHARE; CONSISTENCY SHARE. Times are integer seconds of server time;
blocks are landed blocks. Every refused argument raises ValueError with
a short fixed message; render_*() give the canonical text the PHP
residence (tpc_volunteer.php) reproduces byte for byte.
good_standing(facts, params) -> dict (D126.1)
"GOOD STANDING: certificate current, D(IS) matching, state GREEN, no
reset in force, not SUSPENDED; VOLUNTEER TIER may narrow it to the
TRUST TIER". facts: {certificate_current, dis_match, state, reset_in_
force, suspended, in_tier}. GS1 The first failing clause, in that
order, is the reason; VOLUNTEER TIER = TRUST TIER adds in_tier last.
bucket_refill(bucket, now, params) -> dict (D126.1; RFC 2697 srTCM)
"a token bucket on the call, tokens at VOLUNTEER RATE up to VOLUNTEER
BURST". bucket: {tokens, updated}. BR1 tokens' = min(BURST, tokens +
RATE x (now - updated) / 3600), updated' = now; now >= updated.
bucket_take(bucket) -> dict
BT1 A token is taken when tokens >= 1; else the call has no token.
retry_after(attempt, r, params) -> float (D126.1; Brooker 2015)
"RETRY AFTER = random(0, min(BACKOFF CAP, BACKOFF BASE x 2^attempt)),
attempt the consecutive NO WORK answers, reset by WORK". RA1 u from
TPC-VOLUNTEER-RETRY counter 0; the bound is CAP once 2^attempt
overflows a double (attempt >= 1024); seconds.
intake_update(intake, volunteer_blocks, landed_blocks, params) -> float
"INTAKE the EWMA (w_q) of the share of the account's landed blocks in
the last RATING PERIOD that were volunteer blocks" (D126.1; RED).
IU1 share = volunteer / landed (0 when landed is 0, PD-SC-19);
intake' = (1 - w_q) x intake + w_q x share.
p_nw(intake, params) -> float
"p_nw 0 below INTAKE MIN, linear to INTAKE MAX_P at INTAKE MAX, 1
above": 0 when intake <= MIN; (intake - MIN) / (MAX - MIN) x MAX_P
when MIN < intake <= MAX; 1 when intake > MAX. RED's spacing
correction is not used.
pv_step(state, r, params) -> dict (D126.1)
"p_v is drawn per epoch from P_V BOUNDS with interior endpoints, the
epoch ending with P_V ROTATION per volunteer call fleet-wide
(Geometric)". state: {p_v: null or the value in force}. PV1 With p_v
null, or when u0 < P_V ROTATION (TPC-PV-ROTATE 0), a new epoch starts
on this call: p_v = lo + u1 x (hi - lo) (TPC-PV-DRAW 0). PV2 The step
runs on every call TPX answers (PD-SC-20; alternative: on every call,
refused ones included).
volunteer_share(window, g, v_max) -> dict (D126.1)
"FOR weight from volunteer work beyond v_max of the account's FOR
weight in the half-life window carries zero, the diversity quota's
shape applied to the channel". window: [{volunteer: bool, weight}];
g the match's base weight. VS1 q = clip((v_max x (W + g) - W_v) / g,
0, 1), the same shape as diversity_quota (PD-SC-1).
volunteer_draw(state, call, r, params) -> dict (D126.1; D155.1)
"THE DRAW, per call, in order: no token or inside RETRY AFTER,
REFUSED; u < p_nw(INTAKE), NO WORK; v < p_v, PROBE (canary or parrot
by PARROT RATIO); a legitimate item available, that item; otherwise
NO WORK"; "A too-early call is REFUSED, its own answer with RETRY
AFTER, never NO WORK; ... past EARLY CALL LIMIT refusals in a RATING
PERIOD the channel closes for VOLUNTEER COOL-OFF"; D155.1 "VOLUNTEER
as a fourth answer kind FALCON, outside the p_v law". state: {bucket,
retry_until (server time or null), attempt, refusals, period (the
RATING PERIOD index the refusals belong to), closed_until (block or
null), intake, p_v (pv_step's state)}; call: {now, block, period,
has_item, has_falcon, has_own_record, ratio_k, standing (good_
standing's facts)}.
VD1 Not in GOOD STANDING: REFUSED (STANDING), not counted (it is not
a too-early call). The channel closed (block < closed_until):
REFUSED (CLOSED), not counted (PD-SC-21).
VD2 The bucket is refilled to now. Inside RETRY AFTER (now < retry_
until): REFUSED (EARLY); no token: REFUSED (TOKEN). A refused call
is counted in its RATING PERIOD (a new period restarts the count);
past EARLY CALL LIMIT the channel closes until block + VOLUNTEER
COOL-OFF and the count restarts. Its RETRY AFTER is the jitter law
at the current attempt, and retry_until becomes the later of the
one in force and now + RETRY AFTER (PD-SC-22).
VD3 A token is taken; pv_step runs. An assigned falcon in flight:
FALCON, outside the p_v law, the attempt unchanged (PD-SC-23: the
falcon is neither a WORK BLOCK nor a PROBE, so INTAKE's throttle
does not apply to it).
VD4 u < p_nw(intake) (TPC-VOLUNTEER-NW 0): NO WORK, attempt + 1.
VD5 v < p_v (TPC-VOLUNTEER-PV 0): WORK, a PROBE; a parrot when w < k /
(1 + k) (TPC-VOLUNTEER-KIND 0), a CONSISTENCY PARROT when x <
CONSISTENCY SHARE (TPC-VOLUNTEER-KIND 1) and the account has an
own record, else a CANARY; attempt 0.
VD6 has_item: WORK, the ITEM; attempt 0. Else NO WORK, attempt + 1;
when the queue is empty every WORK answer is therefore a probe.
VD7 Every answer carries RETRY AFTER = retry_after(attempt', r) and
retry_until' = now + RETRY AFTER; the record carries the answer,
its kind, p_nw, p_v, intake and the attempt.
"""
import math
import os
import sys
_here = os.path.dirname(os.path.abspath(__file__))
for _p in (_here,):
if _p not in sys.path:
sys.path.insert(0, _p)
from tpc_beta import canon_float # noqa: E402
from tpc_probe import uniform # noqa: E402
VERSION = "0.1.0"
TIME_MAX = 253402300799
COUNT_MAX = 1000000000
STATES = ("GREEN", "YELLOW", "ORANGE", "SUSPENDED")
TIERS = ("ANY GREEN", "TRUST TIER")
_SKEYS = ("certificate_current", "dis_match", "state", "reset_in_force",
"suspended", "in_tier")
# ---------------------------------------------------------------- common
def _is_int(x, lo, hi):
return type(x) is int and lo <= x <= hi
def _keys_are(o, required, optional=()):
if type(o) is not dict:
return False
for k in required:
if k not in o:
return False
for k in o:
if k not in required and k not in optional:
return False
return True
def _num(x):
if type(x) is bool or not isinstance(x, (int, float)):
return None
x = float(x)
if not math.isfinite(x):
return None
return x
def _is_hex(s, n):
if type(s) is not str or len(s) != n:
return False
for ch in s:
if ch not in "0123456789abcdef":
return False
return True
def _pf(params, key, lo, hi, lo_open=False, hi_open=False):
if type(params) is not dict or key not in params:
raise ValueError("params:" + key)
v = _num(params[key])
if v is None:
raise ValueError("params:" + key)
if v < lo or v > hi or (lo_open and v == lo) or (hi_open and v == hi):
raise ValueError("params:" + key)
return v
def _pi(params, key, lo, hi):
if type(params) is not dict or key not in params:
raise ValueError("params:" + key)
v = params[key]
if not _is_int(v, lo, hi):
raise ValueError("params:" + key)
return v
def _clip(x, lo, hi):
if x < lo:
return lo
if x > hi:
return hi
return x
def _f(x):
return canon_float(x)
def _b(x):
return "true" if x else "false"
# --------------------------------------------------------- good_standing
def good_standing(facts, params):
if not _keys_are(facts, _SKEYS):
raise ValueError("facts")
for k in ("certificate_current", "dis_match", "reset_in_force", "suspended", "in_tier"):
if type(facts[k]) is not bool:
raise ValueError("facts")
if facts["state"] not in STATES:
raise ValueError("facts")
if type(params) is not dict or params.get("VOLUNTEER TIER") not in TIERS:
raise ValueError("params:VOLUNTEER TIER")
if not facts["certificate_current"]:
return {"ok": False, "reason": "CERTIFICATE"}
if not facts["dis_match"]:
return {"ok": False, "reason": "D(IS)"}
if facts["state"] != "GREEN":
return {"ok": False, "reason": "STATE"}
if facts["reset_in_force"]:
return {"ok": False, "reason": "RESET"}
if facts["suspended"]:
return {"ok": False, "reason": "SUSPENDED"}
if params["VOLUNTEER TIER"] == "TRUST TIER" and not facts["in_tier"]:
return {"ok": False, "reason": "TIER"}
return {"ok": True, "reason": "GOOD STANDING"}
def render_standing(d):
return _b(d["ok"]) + "|" + d["reason"]
# ---------------------------------------------------------------- bucket
def _bucket(bucket):
if not _keys_are(bucket, ("tokens", "updated")):
raise ValueError("bucket")
t = _num(bucket["tokens"])
if t is None or t < 0.0 or not _is_int(bucket["updated"], 0, TIME_MAX):
raise ValueError("bucket")
return t, bucket["updated"]
def bucket_refill(bucket, now, params):
tokens, updated = _bucket(bucket)
if not _is_int(now, 0, TIME_MAX) or now < updated:
raise ValueError("now")
rate = _pf(params, "VOLUNTEER RATE", 0.0, 1e9, True)
burst = _pi(params, "VOLUNTEER BURST", 1, COUNT_MAX)
t = tokens + rate * float(now - updated) / 3600.0
if t > float(burst):
t = float(burst)
return {"tokens": t, "updated": now}
def bucket_take(bucket):
tokens, updated = _bucket(bucket)
if tokens >= 1.0:
return {"ok": True, "bucket": {"tokens": tokens - 1.0, "updated": updated}}
return {"ok": False, "bucket": {"tokens": tokens, "updated": updated}}
def render_bucket(b):
return "%s|%d" % (_f(b["tokens"]), b["updated"])
def render_take(d):
return _b(d["ok"]) + "|" + render_bucket(d["bucket"])
# ----------------------------------------------------------- retry_after
def retry_after(attempt, r, params):
if not _is_int(attempt, 0, COUNT_MAX):
raise ValueError("attempt")
if not _is_hex(r, 32):
raise ValueError("r")
base = _pf(params, "BACKOFF BASE", 1.0, float(TIME_MAX))
cap = _pf(params, "BACKOFF CAP", 1.0, float(TIME_MAX))
if cap < base:
raise ValueError("params:BACKOFF CAP")
if attempt >= 1024:
bound = cap
else:
bound = base * math.pow(2.0, float(attempt))
if bound > cap:
bound = cap
return uniform("TPC-VOLUNTEER-RETRY", r, 0) * bound
# ---------------------------------------------------------------- intake
def intake_update(intake, volunteer_blocks, landed_blocks, params):
i = _num(intake)
if i is None or i < 0.0 or i > 1.0:
raise ValueError("intake")
if not _is_int(volunteer_blocks, 0, COUNT_MAX) or not _is_int(landed_blocks, 0, COUNT_MAX):
raise ValueError("blocks")
if volunteer_blocks > landed_blocks:
raise ValueError("blocks")
w = _pf(params, "INTAKE w_q", 0.0, 1.0)
share = 0.0 if landed_blocks == 0 else float(volunteer_blocks) / float(landed_blocks)
return (1.0 - w) * i + w * share
def p_nw(intake, params):
i = _num(intake)
if i is None or i < 0.0 or i > 1.0:
raise ValueError("intake")
lo = _pf(params, "INTAKE MIN", 0.0, 1.0)
hi = _pf(params, "INTAKE MAX", 0.0, 1.0)
mp = _pf(params, "INTAKE MAX_P", 0.0, 1.0)
if hi <= lo:
raise ValueError("params:INTAKE MAX")
if i <= lo:
return 0.0
if i > hi:
return 1.0
return (i - lo) / (hi - lo) * mp
# ----------------------------------------------------------------- p_v
def pv_step(state, r, params):
if not _keys_are(state, ("p_v",)):
raise ValueError("state")
pv = state["p_v"]
if pv is not None:
pv = _num(pv)
if pv is None or pv < 0.0 or pv > 1.0:
raise ValueError("state")
if not _is_hex(r, 32):
raise ValueError("r")
if type(params) is not dict or "P_V BOUNDS" not in params:
raise ValueError("params:P_V BOUNDS")
b = params["P_V BOUNDS"]
if type(b) is not list or len(b) != 2:
raise ValueError("params:P_V BOUNDS")
lo = _num(b[0])
hi = _num(b[1])
if lo is None or hi is None or lo <= 0.0 or hi >= 1.0 or lo > hi:
raise ValueError("params:P_V BOUNDS")
rot = _pf(params, "P_V ROTATION", 0.0, 1.0)
started = False
if pv is None or uniform("TPC-PV-ROTATE", r, 0) < rot:
pv = lo + uniform("TPC-PV-DRAW", r, 0) * (hi - lo)
started = True
return {"p_v": pv, "started": started}
def render_pv(d):
return _f(d["p_v"]) + "|" + _b(d["started"])
# ------------------------------------------------------- volunteer_share
def volunteer_share(window, g, v_max):
if type(window) is not list:
raise ValueError("window")
gf = _num(g)
if gf is None or gf <= 0.0:
raise ValueError("g")
vm = _num(v_max)
if vm is None or vm <= 0.0 or vm >= 1.0:
raise ValueError("v_max")
w_all = 0.0
w_v = 0.0
for e in window:
if not _keys_are(e, ("volunteer", "weight")) or type(e["volunteer"]) is not bool:
raise ValueError("window")
w = _num(e["weight"])
if w is None or w < 0.0:
raise ValueError("window")
w_all = w_all + w
if e["volunteer"]:
w_v = w_v + w
q = _clip((vm * (w_all + gf) - w_v) / gf, 0.0, 1.0)
return {"q": q, "W": w_all, "W_v": w_v, "admitted": gf * q}
def render_share(d):
return "%s|%s|%s|%s" % (_f(d["q"]), _f(d["W"]), _f(d["W_v"]), _f(d["admitted"]))
# -------------------------------------------------------- volunteer_draw
_VKEYS = ("bucket", "retry_until", "attempt", "refusals", "period",
"closed_until", "intake", "p_v")
_CKEYS = ("now", "block", "period", "has_item", "has_falcon",
"has_own_record", "ratio_k", "standing")
def _state_out(st, bucket, retry_until, attempt, refusals, period, closed, intake, pv):
return {"bucket": bucket, "retry_until": retry_until, "attempt": attempt,
"refusals": refusals, "period": period, "closed_until": closed,
"intake": intake, "p_v": pv}
def volunteer_draw(state, call, r, params):
if not _keys_are(state, _VKEYS) or not _keys_are(call, _CKEYS):
raise ValueError("state")
_bucket(state["bucket"])
ru = state["retry_until"]
if ru is not None:
ru = _num(ru)
if ru is None or ru < 0.0:
raise ValueError("state")
if not _is_int(state["attempt"], 0, COUNT_MAX) or not _is_int(state["refusals"], 0, COUNT_MAX):
raise ValueError("state")
if not _is_int(state["period"], 0, COUNT_MAX):
raise ValueError("state")
closed = state["closed_until"]
if closed is not None and not _is_int(closed, 0, COUNT_MAX):
raise ValueError("state")
intake = _num(state["intake"])
if intake is None or intake < 0.0 or intake > 1.0:
raise ValueError("state")
pvs = state["p_v"]
if not _keys_are(pvs, ("p_v",)):
raise ValueError("state")
if not _is_int(call["now"], 0, TIME_MAX) or not _is_int(call["block"], 0, COUNT_MAX):
raise ValueError("call")
if not _is_int(call["period"], 0, COUNT_MAX) or call["period"] < state["period"]:
raise ValueError("call")
for k in ("has_item", "has_falcon", "has_own_record"):
if type(call[k]) is not bool:
raise ValueError("call")
kf = _num(call["ratio_k"])
if kf is None or kf <= 0.0:
raise ValueError("call")
if not _is_hex(r, 32):
raise ValueError("r")
limit = _pi(params, "EARLY CALL LIMIT", 0, COUNT_MAX)
cool = _pi(params, "VOLUNTEER COOL-OFF", 1, COUNT_MAX)
share = _pf(params, "CONSISTENCY SHARE", 0.0, 1.0)
gs = good_standing(call["standing"], params)
now = call["now"]
block = call["block"]
refusals = state["refusals"] if call["period"] == state["period"] else 0
period = call["period"]
attempt = state["attempt"]
rec = {"answer": None, "kind": None, "reason": None, "retry_after": None,
"p_nw": None, "p_v": None, "intake": intake, "attempt": attempt}
def refused(reason, bucket, count):
ra = retry_after(attempt, r, params)
until = float(now) + ra
if ru is not None and ru > until:
until = ru
nref = refusals
nclosed = closed
if count:
nref = refusals + 1
if nref > limit:
nclosed = block + cool
nref = 0
rec.update(answer="REFUSED", reason=reason, retry_after=ra)
return {"record": rec, "state": _state_out(state, bucket, until, attempt, nref,
period, nclosed, intake, pvs)}
if not gs["ok"]:
return refused("STANDING", state["bucket"], False)
if closed is not None and block < closed:
return refused("CLOSED", state["bucket"], False)
bucket = bucket_refill(state["bucket"], now, params)
if ru is not None and float(now) < ru:
return refused("EARLY", bucket, True)
take = bucket_take(bucket)
if not take["ok"]:
return refused("TOKEN", bucket, True)
bucket = take["bucket"]
pv = pv_step(pvs, r, params)
pnw = p_nw(intake, params)
rec["p_nw"] = pnw
rec["p_v"] = pv["p_v"]
if call["has_falcon"]:
rec["answer"] = "FALCON"
rec["kind"] = "FALCON"
elif uniform("TPC-VOLUNTEER-NW", r, 0) < pnw:
rec["answer"] = "NO WORK"
rec["reason"] = "THROTTLE"
attempt = attempt + 1
elif uniform("TPC-VOLUNTEER-PV", r, 0) < pv["p_v"]:
rec["answer"] = "WORK"
if uniform("TPC-VOLUNTEER-KIND", r, 0) < kf / (1.0 + kf):
if uniform("TPC-VOLUNTEER-KIND", r, 1) < share and call["has_own_record"]:
rec["kind"] = "CONSISTENCY PARROT"
else:
rec["kind"] = "CORROBORATION PARROT"
else:
rec["kind"] = "CANARY"
attempt = 0
elif call["has_item"]:
rec["answer"] = "WORK"
rec["kind"] = "ITEM"
attempt = 0
else:
rec["answer"] = "NO WORK"
rec["reason"] = "EMPTY"
attempt = attempt + 1
ra = retry_after(attempt, r, params)
rec["retry_after"] = ra
rec["attempt"] = attempt
return {"record": rec, "state": _state_out(state, bucket, float(now) + ra, attempt, refusals,
period, closed, intake, {"p_v": pv["p_v"]})}
def render_draw(d):
rec = d["record"]
st = d["state"]
lines = ["answer|%s|%s|%s|%s" % (rec["answer"], "-" if rec["kind"] is None else rec["kind"],
"-" if rec["reason"] is None else rec["reason"],
_f(rec["retry_after"]))]
lines.append("draw|%s|%s|%s|%d" % ("-" if rec["p_nw"] is None else _f(rec["p_nw"]),
"-" if rec["p_v"] is None else _f(rec["p_v"]),
_f(rec["intake"]), rec["attempt"]))
lines.append("state|%s|%s|%d|%d|%d|%s|%s" % (
render_bucket(st["bucket"]), _f(st["retry_until"]), st["attempt"], st["refusals"],
st["period"], "-" if st["closed_until"] is None else str(st["closed_until"]),
"-" if st["p_v"]["p_v"] is None else _f(st["p_v"]["p_v"])))
return "\n".join(lines)
# ---------------------------------------------------------------- corpus
FNS = ("standing", "refill", "take", "retry", "intake", "p_nw", "pv",
"share", "draw")
def run_case(fn, args):
"""One corpus case: the render, or "ERROR:" and the message."""
if fn not in FNS:
raise ValueError("fn")
try:
p = args.get("params")
if fn == "standing":
return render_standing(good_standing(args["facts"], p))
if fn == "refill":
return render_bucket(bucket_refill(args["bucket"], args["now"], p))
if fn == "take":
return render_take(bucket_take(args["bucket"]))
if fn == "retry":
return _f(retry_after(args["attempt"], args["r"], p))
if fn == "intake":
return _f(intake_update(args["intake"], args["volunteer_blocks"],
args["landed_blocks"], p))
if fn == "p_nw":
return _f(p_nw(args["intake"], p))
if fn == "pv":
return render_pv(pv_step(args["state"], args["r"], p))
if fn == "share":
return render_share(volunteer_share(args["window"], args["g"], args["v_max"]))
return render_draw(volunteer_draw(args["state"], args["call"], args["r"], p))
except ValueError as ex:
return "ERROR:" + str(ex)
