"""Original LockedIn Labs concurrency-isolation lab. Python 3.11+ standard library. Real concurrent loopback GETs; a gate deliberately holds admitted slow work. Semaphores bound the work section, not TCP connections or HTTP handler threads. No external dependency, queue, business write, throughput or latency benchmark. """ import argparse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import sys import threading from urllib.error import HTTPError from urllib.request import build_opener, ProxyHandler, Request MODES = ("shared", "partitioned") def get_json(url, request_id): opener = build_opener(ProxyHandler({})) try: response = opener.open(Request(url, headers={"X-Fixture-Request": request_id}), timeout=20) except HTTPError as error: response = error with response: body = response.read(4097) if len(body) > 4096: raise RuntimeError("unexpected_fixture_body_size") return {"status": response.code, "body": json.loads(body)} def run_case(mode, slots=2, held=2): if mode not in MODES: raise ValueError("unsupported mode") if type(slots) is not int or not 2 <= slots <= 4: raise ValueError("slots must be an integer from 2 through 4") if type(held) is not int or not 1 <= held <= 4: raise ValueError("held must be an integer from 1 through 4") capacities = {"shared": slots} if mode == "shared" else {"slow": slots - 1, "fast": 1} pools = {name: threading.BoundedSemaphore(value) for name, value in capacities.items()} active = {"slow": 0, "fast": 0} lock, release = threading.Lock(), threading.Event() ids = [f"S{i}" for i in range(1, held + 1)] ready = {name: threading.Event() for name in [*ids, "F1", "S-after", "F-after"]} done = {name: threading.Event() for name in ready} admissions, results, events, errors = {}, {}, [], [] clients = [] peak_total = 0 def snapshot(): return {"active_by_class": dict(active), "active_total": sum(active.values())} class Handler(BaseHTTPRequestHandler): def log_message(self, *_args): pass def send_json(self, status, value): body = json.dumps(value).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): nonlocal peak_total request_id = self.headers.get("X-Fixture-Request") group = {"/slow": "slow", "/fast": "fast"}.get(self.path) if request_id not in ready or group is None: self.send_json(400, {"reason": "unsupported_fixture_request"}) return pool_name = "shared" if mode == "shared" else group pool = pools[pool_name] with lock: accepted = pool.acquire(blocking=False) if accepted: active[group] += 1 peak_total = max(peak_total, sum(active.values())) admissions[request_id] = {"class": group, "admitted": accepted, "pool": pool_name} events.append({"request": request_id, "phase": "admitted" if accepted else "rejected", "pool": pool_name, **snapshot()}) ready[request_id].set() # Control observer knows admission; caller has no success yet. if not accepted: self.send_json(503, {"reason": "work_capacity_full", "request": request_id, "class": group}) return freed = False try: if group == "slow" and not release.wait(15): raise RuntimeError("fixture_release_not_received") with lock: active[group] -= 1 pool.release() freed = True events.append({"request": request_id, "phase": "finished", "pool": pool_name, **snapshot()}) # The protected work section has finished before response bytes are sent. self.send_json(200, {"reason": "work_finished", "request": request_id, "class": group}) except Exception as error: with lock: errors.append(type(error).__name__) self.send_json(500, {"reason": "unexpected_fixture_failure"}) finally: if not freed: with lock: active[group] -= 1 pool.release() events.append({"request": request_id, "phase": "released_after_error", "pool": pool_name, **snapshot()}) server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) server.daemon_threads = False port = server.server_address[1] server_thread = threading.Thread(target=lambda: server.serve_forever(poll_interval=0.01), name="fde-isolation-server") server_thread.start() url = f"http://127.0.0.1:{port}" def request(request_id, path): try: value = get_json(url + path, request_id) with lock: results[request_id] = value except Exception as error: with lock: errors.append(type(error).__name__) finally: done[request_id].set() try: for request_id in ids: client = threading.Thread(target=request, args=(request_id, "/slow"), name="fde-isolation-client-" + request_id) clients.append(client) client.start() if not ready[request_id].wait(5): raise RuntimeError("request_admission_not_observed") if not admissions[request_id]["admitted"] and not done[request_id].wait(5): raise RuntimeError("refused_response_not_observed") with lock: before_fast = snapshot() pending = [name for name in ids if admissions[name]["admitted"] and not done[name].is_set()] request("F1", "/fast") with lock: if "F1" not in results: raise RuntimeError("fast_probe_response_unavailable") probe = results["F1"] after_fast = snapshot() release.set() for client in clients: client.join(5) if client.is_alive(): raise RuntimeError("owned_client_not_finished") request("S-after", "/slow") request("F-after", "/fast") finally: release.set() for client in clients: client.join(5) server.shutdown() server.server_close() server_thread.join(2) if errors or server_thread.is_alive() or any(client.is_alive() for client in clients): raise RuntimeError("fixture_cleanup_or_request_failure:" + ",".join(errors)) return {"mode": mode, "configured_work_slots": slots, "held_slow_requests": held, "pool_capacities": capacities, "admissions": admissions, "observer_before_fast": before_fast, "slow_callers_pending_before_release": pending, "fast_response_before_release": probe, "observer_after_fast_before_release": after_fast, "responses": results, "work_events": events, "final_state": snapshot(), "peak_active_work": peak_total, "server": {"host": "127.0.0.1", "port": port, "thread_stopped": not server_thread.is_alive(), "listener_closed": server.socket.fileno() == -1}, "client_threads_stopped": all(not client.is_alive() for client in clients), "errors": errors} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--mode", choices=MODES, help="One comparison; default runs both") parser.add_argument("--slots", type=int, default=2, help="Total work permits, from 2 through 4; partitioned reserves one for fast work") parser.add_argument("--held", type=int, default=2, help="Slow requests sent before the fast probe, from 1 through 4") args = parser.parse_args() if not 2 <= args.slots <= 4 or not 1 <= args.held <= 4: parser.error("--slots must be 2..4 and --held must be 1..4") print(json.dumps({"schema": "lockedin.concurrency-isolation/v1", "python_version": sys.version.split()[0], "claim": "Actual concurrent loopback requests and work-section permits. Held work is an explicit gate, not an external dependency or measured service duration. No connection/thread-pool, queue, replica or performance guarantee.", "cases": [run_case(mode, args.slots, args.held) for mode in ([args.mode] if args.mode else MODES)]}, indent=2)) if __name__ == "__main__": main()