Improve application responsiveness and background processing
This commit is contained in:
+132
-11
@@ -30,7 +30,7 @@ import urllib.parse
|
||||
import urllib.request
|
||||
import uuid
|
||||
|
||||
VERSION = "1.3.0"
|
||||
VERSION = "1.4.0"
|
||||
LABEL = "io.electric-cherry"
|
||||
MAX_FILE = 32 * 1024 * 1024
|
||||
LIBRARY_DOWNLOAD_WORKERS = 4
|
||||
@@ -329,10 +329,47 @@ def managed_write(root, value, content, uid=1000, gid=1000):
|
||||
os.unlink(temp_name, dir_fd=parent)
|
||||
|
||||
|
||||
class ReadWriteGate:
|
||||
"""Let observations run together; global mutations wait for existing readers."""
|
||||
def __init__(self):
|
||||
self.condition = threading.Condition()
|
||||
self.readers = 0
|
||||
self.writers = 0
|
||||
|
||||
@contextlib.contextmanager
|
||||
def read(self):
|
||||
with self.condition:
|
||||
while self.writers:
|
||||
self.condition.wait()
|
||||
self.readers += 1
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
with self.condition:
|
||||
self.readers -= 1
|
||||
self.condition.notify_all()
|
||||
|
||||
@contextlib.contextmanager
|
||||
def write(self):
|
||||
# The mutation lane is serial, so there is only one writer.
|
||||
with self.condition:
|
||||
self.writers += 1
|
||||
while self.readers:
|
||||
self.condition.wait()
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
with self.condition:
|
||||
self.writers -= 1
|
||||
self.condition.notify_all()
|
||||
|
||||
|
||||
class Journal:
|
||||
def __init__(self, path):
|
||||
def __init__(self, path, recover=True):
|
||||
self.db = sqlite3.connect(path)
|
||||
self.db.row_factory = sqlite3.Row
|
||||
if not recover:
|
||||
return
|
||||
self.db.execute("PRAGMA journal_mode=WAL")
|
||||
self.db.execute("PRAGMA synchronous=FULL")
|
||||
self.db.execute("CREATE TABLE IF NOT EXISTS operations (id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, request TEXT NOT NULL, state TEXT NOT NULL, result TEXT, error TEXT, updated REAL NOT NULL)")
|
||||
@@ -672,6 +709,10 @@ class Agent:
|
||||
self.docker = docker or Docker()
|
||||
self.stopping = threading.Event()
|
||||
self.current_operation = None
|
||||
self.instance_locks = {}
|
||||
self.read_gate = ReadWriteGate()
|
||||
self.read_claim_lock = threading.Lock()
|
||||
self.read_claims = set()
|
||||
self.cancelled_creations = frozenset()
|
||||
self.library_progress = {}
|
||||
self.library_transfers = {}
|
||||
@@ -732,17 +773,25 @@ class Agent:
|
||||
return Path(self.config_path) == AGENT_CONFIG and (UPDATE_JOURNAL.exists() or UPDATE_JOURNAL.is_symlink())
|
||||
|
||||
@contextlib.contextmanager
|
||||
def activity(self):
|
||||
def activity(self, read_only=False):
|
||||
path = self.root / "agent" / "activity.lock"
|
||||
descriptor = os.open(path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
|
||||
mutation_descriptor = None
|
||||
try:
|
||||
try:
|
||||
fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
# All work excludes the updater. Mutations additionally exclude
|
||||
# other mutation batches while read lanes may continue.
|
||||
fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB)
|
||||
if not read_only:
|
||||
mutation_descriptor = os.open(self.root / "agent" / "mutation.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
|
||||
fcntl.flock(mutation_descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
except BlockingIOError:
|
||||
yield False
|
||||
return
|
||||
yield True
|
||||
finally:
|
||||
if mutation_descriptor is not None:
|
||||
os.close(mutation_descriptor)
|
||||
os.close(descriptor)
|
||||
|
||||
def record_health(self):
|
||||
@@ -953,7 +1002,7 @@ class Agent:
|
||||
if line.startswith("PRETTY_NAME="):
|
||||
os_name = line.split("=", 1)[1].strip('"')
|
||||
update = self.update_inventory()
|
||||
capabilities = ["instance.cancel-create"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
|
||||
capabilities = ["instance.cancel-create", "operations.read-lane"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
|
||||
return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
|
||||
|
||||
def safe_to_disrupt(self, instance_id, generation):
|
||||
@@ -1860,23 +1909,89 @@ class Agent:
|
||||
except (AgentError, OSError, ValueError):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def read_operation(operation):
|
||||
return operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "logs.read") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version"))
|
||||
|
||||
def instance_lock(self, operation):
|
||||
key = operation.get("instanceId")
|
||||
return self.instance_locks.setdefault(key, threading.RLock()) if key else contextlib.nullcontext()
|
||||
|
||||
def read_loop(self):
|
||||
# Each lane owns its SQLite connection and never writes executing_id,
|
||||
# checkpoints, or current_operation used by the mutation state machine.
|
||||
journal = Journal(self.root / "agent" / "operations.sqlite3", recover=False)
|
||||
try:
|
||||
while not self.stopping.is_set():
|
||||
try:
|
||||
if self.update_pending():
|
||||
self.stopping.wait(3)
|
||||
continue
|
||||
response = self.request("GET", "/api/agent/operations?lane=read&wait=20")
|
||||
for operation in response.get("operations", []):
|
||||
if self.stopping.is_set() or not self.read_operation(operation):
|
||||
continue
|
||||
with self.read_claim_lock:
|
||||
if operation["id"] in self.read_claims:
|
||||
continue
|
||||
self.read_claims.add(operation["id"])
|
||||
try:
|
||||
with self.activity(read_only=True) as acquired:
|
||||
if not acquired or self.update_pending():
|
||||
continue
|
||||
with self.read_gate.read(), self.instance_lock(operation):
|
||||
row, _ = journal.accept(operation)
|
||||
if row["state"] in ("accepted", "unknown", "running"):
|
||||
self.report(journal.update(operation["id"], "running"))
|
||||
try:
|
||||
result = self.dispatch(operation)
|
||||
row = journal.update(operation["id"], "completed", result)
|
||||
except Exception as err:
|
||||
row = journal.update(operation["id"], "failed", error=str(err) if isinstance(err, AgentError) else "Observation unavailable")
|
||||
self.report(row)
|
||||
finally:
|
||||
with self.read_claim_lock:
|
||||
self.read_claims.discard(operation["id"])
|
||||
except (AgentError, OSError, ValueError, KeyError, sqlite3.Error):
|
||||
self.stopping.wait(3)
|
||||
finally:
|
||||
journal.close()
|
||||
|
||||
def poll_operations(self):
|
||||
if self.stopping.is_set() or self.update_pending():
|
||||
return
|
||||
# Check both update and mutation locks before delivery, release them
|
||||
# during the long poll, and recheck immediately before accepting work.
|
||||
with self.activity() as acquired:
|
||||
if not acquired:
|
||||
return
|
||||
# Never hold the updater lock while waiting for delivery.
|
||||
response = self.request("GET", "/api/agent/operations?lane=write&wait=20")
|
||||
with self.activity() as acquired:
|
||||
if not acquired or self.stopping.is_set() or self.update_pending():
|
||||
return
|
||||
if self.config.get("pendingCredential"):
|
||||
self.recover_credential()
|
||||
response = self.request("GET", "/api/agent/operations")
|
||||
completed = True
|
||||
for operation in response.get("operations", []):
|
||||
if self.stopping.is_set():
|
||||
break
|
||||
if self.read_operation(operation):
|
||||
continue
|
||||
self.current_operation = operation["id"]
|
||||
# Global changes wait for observations; instance changes only
|
||||
# block observations for their own immutable instance identity.
|
||||
guard = self.read_gate.write() if operation["kind"] in ("host.decommission", "credentials.rotate") else contextlib.nullcontext()
|
||||
try:
|
||||
accepted, _ = self.journal.accept(operation)
|
||||
self.report(accepted)
|
||||
self.report(self.execute(operation, report_running=True))
|
||||
with guard, self.instance_lock(operation):
|
||||
accepted, _ = self.journal.accept(operation)
|
||||
self.report(accepted)
|
||||
result = self.execute(operation, report_running=True)
|
||||
self.report(result)
|
||||
completed = completed and result["state"] in ("completed", "failed")
|
||||
finally:
|
||||
self.current_operation = None
|
||||
return completed
|
||||
|
||||
def run(self):
|
||||
# Obtain the configured host port range before accepting any deployment.
|
||||
@@ -1895,13 +2010,19 @@ class Agent:
|
||||
self.stopping.wait(3)
|
||||
heartbeats = threading.Thread(target=self.heartbeat_loop, daemon=True)
|
||||
heartbeats.start()
|
||||
readers = [threading.Thread(target=self.read_loop, daemon=True) for _ in range(2)]
|
||||
for reader in readers:
|
||||
reader.start()
|
||||
while not self.stopping.is_set():
|
||||
completed = False
|
||||
try:
|
||||
self.poll_operations()
|
||||
completed = self.poll_operations()
|
||||
except (AgentError, OSError, ValueError, KeyError):
|
||||
# Logs deliberately omit URLs, tokens, payloads and raw errors.
|
||||
print("Host controller contact unavailable; existing containers continue running", file=sys.stderr)
|
||||
self.stopping.wait(3)
|
||||
self.stopping.wait(0.25 if completed else 3)
|
||||
for reader in readers:
|
||||
reader.join(timeout=2)
|
||||
heartbeats.join(timeout=2)
|
||||
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
@@ -43,11 +44,15 @@ class UpdateIntegrationTests(unittest.TestCase):
|
||||
with self.assertRaises(BlockingIOError):
|
||||
fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
|
||||
def test_operation_delivery_execution_and_reports_exclude_updates(self):
|
||||
operation = {"id": "one", "kind": "host.inspect", "payload": {}}
|
||||
def test_execution_and_reports_exclude_updates_but_long_poll_does_not(self):
|
||||
operation = {"id": "one", "kind": "runtime.pull", "payload": {}}
|
||||
calls = []
|
||||
def request(method, path, body=None):
|
||||
self.assert_activity_held()
|
||||
if method != "GET":
|
||||
self.assert_activity_held()
|
||||
else:
|
||||
with (self.agent.root / "agent" / "activity.lock").open("a") as stream:
|
||||
fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
calls.append((method, path))
|
||||
return {"operations": [operation]} if method == "GET" else {}
|
||||
def dispatch(value):
|
||||
@@ -82,7 +87,61 @@ class UpdateIntegrationTests(unittest.TestCase):
|
||||
module.UPDATE_JOURNAL.unlink()
|
||||
with patch.object(self.agent, "request", return_value={"operations": []}) as request:
|
||||
self.agent.poll_operations()
|
||||
request.assert_called_once_with("GET", "/api/agent/operations")
|
||||
request.assert_called_once_with("GET", "/api/agent/operations?lane=write&wait=20")
|
||||
|
||||
def test_read_lane_completes_while_mutation_is_busy_and_excludes_updater(self):
|
||||
read_done = threading.Event()
|
||||
operation = {"id": "read-one", "kind": "host.inspect", "payload": {}}
|
||||
self.agent.request = lambda method, path, body=None: {"operations": [operation]} if method == "GET" else {}
|
||||
def dispatch(value):
|
||||
self.assert_activity_held()
|
||||
read_done.set()
|
||||
self.agent.stopping.set()
|
||||
return {"ok": True}
|
||||
self.agent.dispatch = dispatch
|
||||
with self.agent.activity() as acquired:
|
||||
self.assertTrue(acquired)
|
||||
reader = threading.Thread(target=self.agent.read_loop)
|
||||
reader.start()
|
||||
self.assertTrue(read_done.wait(2), "Read waited behind unrelated mutation activity")
|
||||
reader.join(2)
|
||||
self.assertFalse(reader.is_alive())
|
||||
self.assertEqual(self.agent.journal.get("read-one")["state"], "completed")
|
||||
self.assertIsNone(self.agent.executing_id)
|
||||
|
||||
def test_read_lane_never_dispatches_mutations(self):
|
||||
def request(method, path, body=None):
|
||||
self.agent.stopping.set()
|
||||
return {"operations": [{"id": "bad-read", "kind": "instance.remove", "payload": {}}]}
|
||||
self.agent.request = request
|
||||
with patch.object(self.agent, "dispatch") as dispatch:
|
||||
self.agent.read_loop()
|
||||
dispatch.assert_not_called()
|
||||
|
||||
def test_instance_mutations_and_global_mutations_exclude_observations(self):
|
||||
operation = {"instanceId": "one"}
|
||||
entered = threading.Event()
|
||||
release = threading.Event()
|
||||
def read():
|
||||
with self.agent.read_gate.read(), self.agent.instance_lock(operation):
|
||||
entered.set()
|
||||
release.wait(2)
|
||||
with self.agent.instance_lock(operation):
|
||||
reader = threading.Thread(target=read)
|
||||
reader.start()
|
||||
self.assertFalse(entered.wait(0.05))
|
||||
self.assertTrue(entered.wait(2))
|
||||
writer_entered = threading.Event()
|
||||
def write():
|
||||
with self.agent.read_gate.write():
|
||||
writer_entered.set()
|
||||
writer = threading.Thread(target=write)
|
||||
writer.start()
|
||||
self.assertFalse(writer_entered.wait(0.05))
|
||||
release.set()
|
||||
reader.join(2)
|
||||
writer.join(2)
|
||||
self.assertTrue(writer_entered.is_set())
|
||||
|
||||
def test_failed_controller_contact_never_writes_healthy_receipt(self):
|
||||
with patch.object(self.agent, "request", side_effect=module.AgentError("offline")):
|
||||
|
||||
Reference in New Issue
Block a user