#!/usr/bin/env python3
# hammer.py -- JSON-RPC robustness/load harness for a LOCAL regtest dragonxd.
#
# Scope: connects only to the --host/--port you pass (meant to be 127.0.0.1 and a
# regtest RPC port). Its job is to reproduce a known file-descriptor bug in the
# libevent-2.1.8 HTTP RPC server (src/httpserver.cpp) by exercising the same
# request patterns the production pool box sees: many concurrent clients, clients
# that disconnect mid-request/mid-response (the explorer indexer's 120 s client
# timeout), work-queue overflow, keep-alive pipelining, and malformed input.
# stdlib only.
#
# Correctness oracle: every request carries a unique JSON-RPC id; if a reply comes
# back with a different id on the same socket, that is cross-talk (the server wrote
# one client's reply onto another client's fd) and is logged + counted.
#
# See README.txt for the exact invocation.

import argparse
import base64
import collections
import json
import os
import random
import socket
import struct
import threading
import time

# ------------------------------------------------------------------ config/state

class C:
    host = "127.0.0.1"
    port = 19461
    auth = ""
    ctimeout = 15.0

CFG = C()
STOP = threading.Event()
SHARED = {"hashes": [], "lpid": None, "zaddrs": [], "rawblock": None}
SHARED_LOCK = threading.Lock()
_ID = [random.SystemRandom().randrange(1 << 40)]
_ID_LOCK = threading.Lock()
LOGLOCK = threading.Lock()
LOGF = [None]

def next_id():
    with _ID_LOCK:
        _ID[0] += 1
        return _ID[0]

def log(msg):
    line = "%s %s" % (time.strftime("%H:%M:%S"), msg)
    with LOGLOCK:
        print(line, flush=True)
        if LOGF[0]:
            LOGF[0].write(line + "\n"); LOGF[0].flush()

# ------------------------------------------------------------------ stats

KEYS = ["sent", "ok", "rpcerr", "wq500", "http401", "http4xx", "httpother",
        "early_rst", "early_fin", "srvreset", "srveof", "ctimeout", "connfail",
        "idmismatch", "malformed", "mined", "half_open", "half_srvclose"]

class Stats:
    def __init__(self):
        self.lk = threading.Lock()
        self.total = collections.Counter()
        self.win = collections.Counter()
        self.anom = []
    def inc(self, k, n=1):
        with self.lk:
            self.total[k] += n; self.win[k] += n
    def anomaly(self, kind, detail):
        with self.lk:
            if len(self.anom) < 300:
                self.anom.append((time.time(), kind, detail))
        log("ANOMALY %s %s" % (kind, detail))
    def swap(self):
        with self.lk:
            w = self.win; self.win = collections.Counter(); return w

ST = Stats()

# ------------------------------------------------------------------ socket helpers

def auth_header(u, p):
    return "Basic " + base64.b64encode(("%s:%s" % (u, p)).encode()).decode()

def mkconn(timeout=None, rcvbuf=None):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    if rcvbuf:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, rcvbuf)
    s.settimeout(timeout if timeout is not None else CFG.ctimeout)
    s.connect((CFG.host, CFG.port))
    return s

def end_rst(s):
    # linger 0 -> TCP RST on close (models a client that dies / resets abruptly)
    try:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
    except OSError:
        pass
    try:
        s.close()
    except OSError:
        pass

def end_fin(s):
    try:
        s.shutdown(socket.SHUT_RDWR)
    except OSError:
        pass
    try:
        s.close()
    except OSError:
        pass

def end_early(s, how=None):
    how = how or random.choice(("rst", "fin"))
    if how == "rst":
        end_rst(s); ST.inc("early_rst")
    else:
        end_fin(s); ST.inc("early_fin")

def req_bytes(method, params, rid, keepalive=True, auth=None, uri="/",
              httpmethod="POST", body=None, clen=None):
    if body is None:
        body = json.dumps({"jsonrpc": "1.0", "id": rid,
                           "method": method, "params": params}).encode()
    head = [
        "%s %s HTTP/1.1" % (httpmethod, uri),
        "Host: %s:%d" % (CFG.host, CFG.port),
        "Authorization: %s" % (auth if auth is not None else CFG.auth),
        "Content-Type: application/json",
        "Content-Length: %d" % (clen if clen is not None else len(body)),
        "Connection: %s" % ("keep-alive" if keepalive else "close"),
        "", "",
    ]
    return "\r\n".join(head).encode() + body

class Reader:
    """Incremental HTTP/1.1 reply reader over one socket."""
    def __init__(self, s):
        self.s = s; self.buf = b""
    def _fill(self):
        d = self.s.recv(65536)
        if not d:
            raise EOFError()
        self.buf += d
    def read(self):
        while b"\r\n\r\n" not in self.buf:
            self._fill()
            if len(self.buf) > (1 << 20):
                raise ValueError("oversize header")
        head, self.buf = self.buf.split(b"\r\n\r\n", 1)
        lines = head.split(b"\r\n")
        p = lines[0].split(b" ", 2)
        if len(p) < 2 or not p[0].startswith(b"HTTP/1."):
            raise ValueError("bad status line %r" % lines[0][:80])
        status = int(p[1])
        hdr = {}
        for ln in lines[1:]:
            if b":" in ln:
                k, v = ln.split(b":", 1)
                hdr[k.strip().lower()] = v.strip()
        if hdr.get(b"transfer-encoding", b"").lower() == b"chunked":
            body = b""
            while True:
                while b"\r\n" not in self.buf:
                    self._fill()
                ln, self.buf = self.buf.split(b"\r\n", 1)
                n = int(ln.split(b";")[0], 16)
                while len(self.buf) < n + 2:
                    self._fill()
                body += self.buf[:n]; self.buf = self.buf[n + 2:]
                if n == 0:
                    break
        elif b"content-length" in hdr:
            n = int(hdr[b"content-length"])
            while len(self.buf) < n:
                self._fill()
            body, self.buf = self.buf[:n], self.buf[n:]
        else:
            try:
                while True:
                    self._fill()
            except EOFError:
                pass
            body, self.buf = self.buf, b""
        return status, hdr, body

def account(status, body, rid, method, expect401=False):
    if status == 401:
        ST.inc("http401")
        if not expect401:
            ST.anomaly("unexpected401", (method,))
        return None
    if status == 500 and b"Work queue depth exceeded" in body:
        ST.inc("wq500"); return None
    j = None
    if body[:1] in (b"{", b"["):
        try:
            j = json.loads(body)
        except ValueError:
            j = None
    if j is None:
        if 400 <= status < 500:
            ST.inc("http4xx")
        else:
            ST.inc("httpother")
            if status == 200:
                ST.anomaly("non-json-200", (method, body[:120].decode("latin1")))
        return None
    if isinstance(j, dict) and rid is not None and j.get("id") != rid:
        ST.inc("idmismatch")
        ST.anomaly("idmismatch", {"method": method, "sent": rid, "got": j.get("id"),
                                  "body": body[:200].decode("latin1")})
    if isinstance(j, dict) and j.get("error"):
        ST.inc("rpcerr")
    else:
        ST.inc("ok")
    return j

def call(s, rd, method, params, keepalive=True, expect401=False, auth=None):
    rid = next_id()
    try:
        s.sendall(req_bytes(method, params, rid, keepalive=keepalive, auth=auth))
        ST.inc("sent")
        status, hdr, body = rd.read()
    except socket.timeout:
        ST.inc("ctimeout"); return None, False
    except (ConnectionResetError, BrokenPipeError):
        ST.inc("srvreset"); return None, False
    except EOFError:
        ST.inc("srveof"); return None, False
    except ValueError as e:
        ST.inc("malformed"); ST.anomaly("malformed", (method, str(e))); return None, False
    except OSError:
        ST.inc("srvreset"); return None, False
    j = account(status, body, rid, method, expect401)
    alive = keepalive and hdr.get(b"connection", b"").lower() != b"close"
    return j, alive

# ------------------------------------------------------------------ menus

def a_hash():
    with SHARED_LOCK:
        return random.choice(SHARED["hashes"]) if SHARED["hashes"] else None

def a_zaddr():
    with SHARED_LOCK:
        return random.choice(SHARED["zaddrs"]) if SHARED["zaddrs"] else None

def normal_menu():
    # Mostly fast calls (sub-ms) so a well-behaved client interleaves with mining.
    # Heavy wallet calls (z_gettotalbalance ~2.3s on the golden wallet) appear but
    # are rare, so they don't permanently peg the small worker pool on their own.
    fast = [("getinfo", []), ("getblockcount", []), ("getbestblockhash", []),
            ("getrawmempool", []), ("getmininginfo", []), ("getdifficulty", []),
            ("getblockchaininfo", []), ("getpeerinfo", []),
            ("getblocktemplate", []), ("getnetworkinfo", [])]
    h = a_hash()
    if h:
        fast.append(("getblock", [h, 2]))
        fast.append(("getblockheader", [h]))
    # "slow" here is only mildly-larger fast calls; the genuinely heavy wallet calls
    # live in the paced slow-fire class so a normal dashboard poller never pegs the
    # worker pool.
    slow = [("coinsupply", []), ("listtransactions", ["*", 10]),
            ("listunspent", [0]), ("getblock", [a_hash() or "", 2])]
    return random.choice(slow) if random.random() < 0.1 else random.choice(fast)

def medium_menu():
    # Fast to compute (~4 ms on this node) but a non-trivial reply to write, so they
    # are good targets for a mid-response disconnect. The fd-reuse race fires on the
    # disconnect, not on call duration, so firing these at a HIGH rate drives the
    # race hard while the worker frees in milliseconds (mining keeps its slots).
    m = [("listtransactions", ["*", 10000]), ("getrawmempool", []),
         ("getrawmempool", [True]), ("getblockchaininfo", []),
         ("getblocktemplate", []), ("getchaintips", []), ("getinfo", []),
         ("coinsupply", []), ("listunspent", [0])]
    h = a_hash()
    if h:
        m.append(("getblock", [h, 2]))
        m.append(("getblock", [h, 1]))
    return random.choice(m)

def heavy_menu():
    # Genuinely slow calls (z_gettotalbalance ~1.7 s, z_listunspent ~5.7 s on the
    # golden wallet). Fired only by the small, PACED slow-fire class so a worker
    # keeps running a multi-second call long after the client disconnected (the
    # production indexer/coinsupply scenario that widens the fd-reuse window),
    # without the fire-and-disconnect backlog pinning the whole pool.
    m = [("z_gettotalbalance", [0]), ("z_listunspent", [0])]
    z = a_zaddr()
    if z:
        m.append(("z_listreceivedbyaddress", [z, 0]))
    return random.choice(m)

# ------------------------------------------------------------------ clients

def client_normal(cid):
    while not STOP.is_set():
        keepalive = random.random() < 0.7
        try:
            s = mkconn()
        except OSError:
            ST.inc("connfail"); time.sleep(0.05); continue
        rd = Reader(s)
        n = random.randint(1, 8) if keepalive else 1
        try:
            for _ in range(n):
                if STOP.is_set():
                    break
                method, params = normal_menu()
                j, alive = call(s, rd, method, params, keepalive=keepalive)
                if not alive:
                    break
        finally:
            (end_fin if random.random() < 0.8 else end_rst)(s)
        time.sleep(random.uniform(0.01, 0.15))

def client_impatient(cid):
    # High-rate fast-disconnect churn: send a (medium) request, then disconnect
    # after 1-500 ms at a random phase. This is the main driver of the fd-reuse
    # race -- a disconnect concurrent with the server writing the reply / re-arming
    # the socket -- and the worker frees in ms so mining is unaffected.
    while not STOP.is_set():
        method, params = medium_menu()
        rid = next_id()
        body = json.dumps({"jsonrpc": "1.0", "id": rid, "method": method,
                           "params": params}).encode()
        phase = random.choice(("mid_request", "after_send", "mid_response"))
        try:
            s = mkconn(timeout=2.0)
        except OSError:
            ST.inc("connfail"); continue
        try:
            if phase == "mid_request":
                full = req_bytes(method, params, rid, keepalive=False, body=body)
                cut = random.randint(len(full) - len(body), len(full) - 1)
                s.sendall(full[:cut])
                time.sleep(random.uniform(0.001, 0.5))
                end_early(s); continue
            s.sendall(req_bytes(method, params, rid, keepalive=False, body=body))
            ST.inc("sent")
            time.sleep(random.uniform(0.001, 0.5))
            if phase == "mid_response":
                try:
                    s.recv(random.randint(0, 64))  # read a few bytes, then bail
                except OSError:
                    pass
            end_early(s)
        except (OSError, socket.timeout):
            ST.inc("srvreset")
            end_rst(s)
        time.sleep(random.uniform(0.05, 0.4))

def client_slowfire(cid, pace):
    # Paced indexer/coinsupply client: fire ONE genuinely-slow request, disconnect
    # after 1-500 ms (so the worker runs the multi-second call with no client left
    # == the production fd-reuse window), then wait `pace` s before the next so a
    # fire-and-disconnect backlog cannot pin the whole worker pool. Phase: cut the
    # request mid-body, or send it whole and abort while the worker is busy, or read
    # a few bytes of the reply then abort.
    while not STOP.is_set():
        method, params = heavy_menu()
        rid = next_id()
        body = json.dumps({"jsonrpc": "1.0", "id": rid, "method": method,
                           "params": params}).encode()
        phase = random.choice(("mid_request", "after_send", "mid_response"))
        try:
            s = mkconn(timeout=2.0)
        except OSError:
            ST.inc("connfail"); time.sleep(pace); continue
        try:
            if phase == "mid_request":
                full = req_bytes(method, params, rid, keepalive=False, body=body)
                cut = random.randint(len(full) - len(body), len(full) - 1)
                s.sendall(full[:cut])
                time.sleep(random.uniform(0.001, 0.5)); end_early(s)
            else:
                s.sendall(req_bytes(method, params, rid, keepalive=False, body=body))
                ST.inc("sent")
                time.sleep(random.uniform(0.001, 0.5))
                if phase == "mid_response":
                    try:
                        s.recv(random.randint(0, 64))
                    except OSError:
                        pass
                end_early(s)
        except (OSError, socket.timeout):
            ST.inc("srvreset"); end_rst(s)
        time.sleep(random.uniform(pace * 0.5, pace * 1.5))

def client_half_open(cid):
    # connect, send nothing / partial headers, trickle, then vanish
    while not STOP.is_set():
        try:
            s = mkconn(timeout=1.0)
        except OSError:
            ST.inc("connfail"); time.sleep(0.1); continue
        ST.inc("half_open")
        style = random.choice(("silent", "partial_headers", "full_headers_no_body"))
        try:
            if style == "partial_headers":
                s.sendall(b"POST / HTTP/1.1\r\nHost: x\r\n")
            elif style == "full_headers_no_body":
                s.sendall(b"POST / HTTP/1.1\r\nHost: x\r\nAuthorization: " +
                          CFG.auth.encode() + b"\r\nContent-Length: 1000\r\n\r\n")
            deadline = time.monotonic() + random.uniform(0.1, 20.0)
            while time.monotonic() < deadline and not STOP.is_set():
                time.sleep(random.uniform(0.2, 1.5))
                try:
                    s.sendall(b" ")  # one byte, keep it half-alive
                except OSError:
                    ST.inc("half_srvclose"); break
                try:
                    s.settimeout(0.01)
                    if s.recv(1) == b"":
                        ST.inc("half_srvclose"); break
                except socket.timeout:
                    pass
                except OSError:
                    ST.inc("half_srvclose"); break
        finally:
            end_early(s)

def client_burst(cid, size, every):
    while not STOP.is_set():
        batch = []
        for _ in range(size):
            try:
                s = mkconn(timeout=3.0)
            except OSError:
                ST.inc("connfail"); continue
            batch.append(s)
        for s in batch:
            try:
                method, params = medium_menu()
                s.sendall(req_bytes(method, params, next_id(), keepalive=False))
                ST.inc("sent")
            except OSError:
                pass
        # most close immediately (before the 500/reply is written); a few read it
        keep = random.sample(batch, max(0, len(batch) // 4)) if batch else []
        for s in batch:
            if s in keep:
                continue
            end_early(s)
        for s in keep:
            try:
                rd = Reader(s); st, hd, bd = rd.read(); account(st, bd, None, "burst")
            except Exception:
                ST.inc("srvreset")
            finally:
                end_fin(s)
        for _ in range(int(every * 10)):
            if STOP.is_set():
                break
            time.sleep(0.1)

def client_pipeline(cid):
    while not STOP.is_set():
        try:
            s = mkconn()
        except OSError:
            ST.inc("connfail"); time.sleep(0.05); continue
        n = random.randint(2, 8)
        ids = []
        blob = b""
        for _ in range(n):
            rid = next_id(); ids.append(rid)
            method, params = normal_menu()
            blob += req_bytes(method, params, rid, keepalive=True)
        try:
            s.sendall(blob); ST.inc("sent", n)
            rd = Reader(s)
            toread = random.randint(0, n)  # sometimes leave replies unread, then close
            for k in range(toread):
                st, hd, bd = rd.read()
                account(st, bd, ids[k], "pipeline")
        except (OSError, EOFError, ValueError):
            ST.inc("srvreset")
        finally:
            end_early(s)
        time.sleep(random.uniform(0, 0.03))

def client_garbage(cid):
    probes = [
        ("submitblock", lambda: ["ab" * random.randint(1, 4000)]),
        ("submitblock", lambda: [os.urandom(random.randint(1, 200)).hex()]),
        ("submitblock", lambda: ["zzzz"]),
        ("submitblock", lambda: [""]),
    ]
    while not STOP.is_set():
        pick = random.random()
        try:
            s = mkconn(timeout=5.0)
        except OSError:
            ST.inc("connfail"); continue
        try:
            if pick < 0.5:
                method, mk = random.choice(probes)
                j, _ = call(s, Reader(s), method, mk(), keepalive=False)
            elif pick < 0.65:
                # malformed JSON body
                s.sendall(req_bytes("x", [], next_id(), keepalive=False,
                                    body=b'{"jsonrpc":"1.0","method":'))
                ST.inc("sent")
                try:
                    account(*((lambda r: (r[0], r[2]))(Reader(s).read())), rid=None, method="badjson")
                except Exception:
                    ST.inc("srvreset")
            elif pick < 0.8:
                # unknown method / wrong URI
                s.sendall(req_bytes("getinfo", [], next_id(), keepalive=False, uri="/nope"))
                ST.inc("sent")
                try:
                    st, hd, bd = Reader(s).read(); account(st, bd, None, "wronguri")
                except Exception:
                    ST.inc("srvreset")
            else:
                # Content-Length larger than the body we send, then bail
                body = b'{"id":1}'
                s.sendall(req_bytes("getinfo", [], 1, keepalive=False, body=body, clen=len(body) + 5000))
                ST.inc("sent")
                time.sleep(random.uniform(0.001, 0.3))
                end_early(s)
                continue
        except (OSError, socket.timeout):
            ST.inc("srvreset")
        finally:
            end_early(s)
        time.sleep(random.uniform(0.02, 0.2))

def client_slowread(cid):
    # ask for a big reply with a tiny recv buffer and do not read it, stalling the
    # server write side, then close abruptly
    while not STOP.is_set():
        try:
            s = mkconn(timeout=3.0, rcvbuf=512)
        except OSError:
            ST.inc("connfail"); continue
        try:
            s.sendall(req_bytes("listtransactions", ["*", 10000], next_id(), keepalive=False))
            ST.inc("sent")
            time.sleep(random.uniform(0.05, 1.0))
        except OSError:
            ST.inc("srvreset")
        finally:
            end_early(s)
        time.sleep(random.uniform(0, 0.05))

def _stale_lpid():
    # A longpollid whose hash != the current tip makes getblocktemplate return at
    # once (the server's wait loop condition is immediately false), so it exercises
    # the long-poll request path (lpval parsing, cs_main leave/enter) and the
    # mid-response disconnect WITHOUT parking a worker. On a tiny -rpcthreads pool a
    # genuinely parking long-poll + RPC-driven mining self-deadlocks (both workers
    # wait for a block that only a third worker could mine); that is a pool-size
    # artifact, not the fd bug, so the harness avoids it by default.
    with SHARED_LOCK:
        lpid = SHARED["lpid"]
    if lpid and len(lpid) > 64:
        h = lpid[:64]
        flip = "0" if h[0] != "0" else "1"
        return flip + h[1:] + lpid[64:]
    return ("0" * 64) + "0"

def client_longpoll(cid):
    while not STOP.is_set():
        try:
            s = mkconn(timeout=10.0)
        except OSError:
            ST.inc("connfail"); time.sleep(0.2); continue
        rd = Reader(s)
        params = [{"longpollid": _stale_lpid()}]
        rid = next_id()
        try:
            s.sendall(req_bytes("getblocktemplate", params, rid, keepalive=False))
            ST.inc("sent")
            if random.random() < 0.5:
                # impatient pool worker: abandon the request after 1-500 ms
                time.sleep(random.uniform(0.001, 0.5))
                end_early(s); continue
            st, hd, bd = rd.read()
            j = account(st, bd, rid, "getblocktemplate(lp)")
            if j and isinstance(j, dict) and j.get("result"):
                with SHARED_LOCK:
                    SHARED["lpid"] = j["result"].get("longpollid")
        except (OSError, socket.timeout, EOFError, ValueError):
            ST.inc("srvreset")
        finally:
            end_fin(s)
        time.sleep(random.uniform(0.02, 0.2))

def miner(interval):
    # Mine a block every `interval` seconds. Retry through work-queue rejections so
    # blocks still land under heavy load (the production box mines steadily too).
    while not STOP.is_set():
        got = False
        deadline = time.monotonic() + interval + 30
        while not got and not STOP.is_set() and time.monotonic() < deadline:
            try:
                s = mkconn(timeout=60.0)
                j, _ = call(s, Reader(s), "generate", [1], keepalive=False)
                end_fin(s)
                if j and not j.get("error") and j.get("result"):
                    ST.inc("mined"); got = True
                    h = j["result"][0] if isinstance(j["result"], list) else None
                    if h:
                        with SHARED_LOCK:
                            SHARED["hashes"].append(h)
                            SHARED["hashes"] = SHARED["hashes"][-200:]
                else:
                    time.sleep(0.2)  # likely work-queue 500; back off and retry
            except Exception:
                time.sleep(0.2)
        for _ in range(int(interval * 10)):
            if STOP.is_set():
                break
            time.sleep(0.1)

# ------------------------------------------------------------------ bootstrap shared state

def bootstrap():
    try:
        s = mkconn(timeout=30.0); rd = Reader(s)
        j, _ = call(s, rd, "getbestblockhash", [], keepalive=True)
        if j and j.get("result"):
            with SHARED_LOCK:
                SHARED["hashes"].append(j["result"])
        j, _ = call(s, rd, "getblockcount", [], keepalive=True)
        cnt = j.get("result") if j else None
        if isinstance(cnt, int):
            for hh in range(max(0, cnt - 30), cnt + 1):
                jj, _ = call(s, rd, "getblockhash", [hh], keepalive=True)
                if jj and jj.get("result"):
                    with SHARED_LOCK:
                        SHARED["hashes"].append(jj["result"])
        j, _ = call(s, rd, "listaddresses", [], keepalive=True)
        j, _ = call(s, rd, "z_listaddresses", [], keepalive=True)
        if j and isinstance(j.get("result"), list):
            with SHARED_LOCK:
                SHARED["zaddrs"] = j["result"][:16]
        j, _ = call(s, rd, "getblocktemplate", [{}], keepalive=True)
        if j and isinstance(j.get("result"), dict):
            with SHARED_LOCK:
                SHARED["lpid"] = j["result"].get("longpollid")
        end_fin(s)
        log("bootstrap: %d hashes, %d zaddrs, lpid=%s" %
            (len(SHARED["hashes"]), len(SHARED["zaddrs"]), bool(SHARED["lpid"])))
    except Exception as e:
        log("bootstrap failed: %s" % e)

# ------------------------------------------------------------------ fd accounting

def count_fds(pid):
    if not pid:
        return (None, None, None)
    base = "/proc/%d/fd" % pid
    try:
        fds = os.listdir(base)
    except OSError:
        return (None, None, None)
    sock = files = 0
    for fd in fds:
        try:
            tgt = os.readlink(os.path.join(base, fd))
        except OSError:
            continue
        if tgt.startswith("socket:"):
            sock += 1
        elif tgt.startswith("/"):
            files += 1
    return (len(fds), sock, files)

def pid_alive(pid):
    if not pid:
        return True
    try:
        os.kill(pid, 0); return True
    except OSError:
        return False

# ------------------------------------------------------------------ main

def reporter(pid):
    t0 = time.monotonic()
    while not STOP.is_set():
        for _ in range(10):
            if STOP.is_set():
                break
            time.sleep(0.1)
        w = ST.swap()
        tot, sk, fl = count_fds(pid)
        fdstr = ""
        if tot is not None:
            fdstr = " | fds=%d sock=%d file=%d" % (tot, sk, fl)
        log("t=%4ds ok=%d rpcerr=%d wq500=%d 4xx=%d other=%d | "
            "earlyRST=%d earlyFIN=%d srvRST=%d srvEOF=%d ctmo=%d connfail=%d | "
            "idmis=%d malf=%d | mined=%d halfopen=%d%s" % (
                int(time.monotonic() - t0),
                w["ok"], w["rpcerr"], w["wq500"], w["http4xx"], w["httpother"],
                w["early_rst"], w["early_fin"], w["srvreset"], w["srveof"],
                w["ctimeout"], w["connfail"], w["idmismatch"], w["malformed"],
                w["mined"], w["half_open"], fdstr))
        if pid and not pid_alive(pid):
            log("!!! NODE PROCESS %d IS GONE -- stopping harness" % pid)
            STOP.set()
            break

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=19461)
    ap.add_argument("--user", default="ab")
    ap.add_argument("--password", default="abpass")
    ap.add_argument("--duration", type=int, default=300)
    ap.add_argument("--ctimeout", type=float, default=15.0)
    ap.add_argument("--normal", type=int, default=24)
    ap.add_argument("--impatient", type=int, default=10)
    ap.add_argument("--slowfire", type=int, default=3,
                    help="paced clients that fire a genuinely-slow call then disconnect")
    ap.add_argument("--slowfire-pace", type=float, default=2.0)
    ap.add_argument("--halfopen", type=int, default=6)
    ap.add_argument("--pipeline", type=int, default=6)
    ap.add_argument("--garbage", type=int, default=4)
    ap.add_argument("--slowread", type=int, default=2)
    ap.add_argument("--longpoll", type=int, default=6)
    # burst-size must exceed (rpcthreads + rpcworkqueue) to overflow the queue even
    # when calls are fast; with -rpcthreads=8 -rpcworkqueue=4 that threshold is 12.
    ap.add_argument("--burst-threads", type=int, default=1)
    ap.add_argument("--burst-size", type=int, default=24)
    ap.add_argument("--burst-every", type=float, default=1.5)
    ap.add_argument("--mine", action="store_true", help="run a miner thread")
    ap.add_argument("--mine-interval", type=float, default=4.0)
    ap.add_argument("--pid", type=int, default=0, help="node pid, for /proc fd + liveness")
    ap.add_argument("--logfile", default="")
    args = ap.parse_args()

    CFG.host = args.host; CFG.port = args.port
    CFG.auth = auth_header(args.user, args.password); CFG.ctimeout = args.ctimeout
    if args.logfile:
        LOGF[0] = open(args.logfile, "a")

    log("hammer start host=%s port=%d duration=%ds pid=%s" %
        (CFG.host, CFG.port, args.duration, args.pid or "-"))
    bootstrap()

    threads = []
    def spawn(fn, n, *a):
        for i in range(n):
            t = threading.Thread(target=fn, args=(i,) + a, daemon=True)
            t.start(); threads.append(t)

    spawn(client_normal, args.normal)
    spawn(client_impatient, args.impatient)
    spawn(client_slowfire, args.slowfire, args.slowfire_pace)
    spawn(client_half_open, args.halfopen)
    spawn(client_pipeline, args.pipeline)
    spawn(client_garbage, args.garbage)
    spawn(client_slowread, args.slowread)
    spawn(client_longpoll, args.longpoll)
    for i in range(args.burst_threads):
        t = threading.Thread(target=client_burst, args=(i, args.burst_size, args.burst_every), daemon=True)
        t.start(); threads.append(t)
    if args.mine:
        t = threading.Thread(target=miner, args=(args.mine_interval,), daemon=True)
        t.start(); threads.append(t)

    rt = threading.Thread(target=reporter, args=(args.pid,), daemon=True)
    rt.start()

    end = time.monotonic() + args.duration
    try:
        while time.monotonic() < end and not STOP.is_set():
            time.sleep(0.2)
    except KeyboardInterrupt:
        pass
    STOP.set()
    time.sleep(1.0)

    log("=== TOTALS ===")
    for k in KEYS:
        if ST.total[k]:
            log("  %-14s %d" % (k, ST.total[k]))
    log("anomalies: %d" % len(ST.anom))
    for t, kind, detail in ST.anom[:40]:
        log("  %s %s %s" % (time.strftime("%H:%M:%S", time.localtime(t)), kind, detail))
    if args.pid:
        log("node alive at end: %s" % pid_alive(args.pid))

if __name__ == "__main__":
    main()