diff --git a/operations/claims.py b/operations/claims.py index bcd79927..53681221 100644 --- a/operations/claims.py +++ b/operations/claims.py @@ -3,10 +3,10 @@ or a job writes its claims in one short transaction, only if no active claim conflicts; otherwise it is refused at once (Busy).""" import os -import socket import threading from contextlib import contextmanager from datetime import timedelta +from functools import cache from django.db import connection, transaction from django.db.models import Q @@ -57,8 +57,34 @@ def quota_key(user_id): return f"quota:user:{user_id}" +@cache +def _node(): + """The installation's id (worker.host_identity): the same after a container + is recreated on its volume, unlike its hostname.""" + from .worker import data_dir, host_identity # the worker imports this module + + return host_identity(data_dir()) + + def holder_id(): - return f"{socket.gethostname()}:{os.getpid()}:{threading.get_ident()}" + return f"{_node()}:{os.getpid()}:{threading.get_ident()}" + + +def _gone(holder): + """Whether a transient claim's process ended: one of this installation + that no longer runs, killed before its request's finally removed the + claim (a gunicorn worker past its timeout, OOM, a restart). A PID used + again keeps the claim until it expires.""" + node, _sep, rest = holder.partition(":") + if node != _node(): + return False + try: + os.kill(int(rest.partition(":")[0]), 0) + except ProcessLookupError: + return True + except PermissionError: # another user's process + pass + return False def serialize(keys): @@ -80,6 +106,9 @@ def _check(wanted, now): ) for claim in active: if wanted[claim.resource] == Claim.EXCLUSIVE or claim.mode == Claim.EXCLUSIVE: + if claim.operation is None and _gone(claim.holder): + claim.delete() + continue raise Busy(claim.resource, claim.operation) diff --git a/operations/tests/test_claims.py b/operations/tests/test_claims.py index ae68a8b9..095541ff 100644 --- a/operations/tests/test_claims.py +++ b/operations/tests/test_claims.py @@ -1,6 +1,7 @@ """Admission writes claims only when nothing active conflicts, all or nothing; exclusive conflicts with any claim, shared only with exclusive.""" +import subprocess from datetime import timedelta from django.contrib.auth import get_user_model @@ -62,6 +63,26 @@ class AdmitTestCase(TestCase): claims.admit({"instance:u1": X}, holder="b") self.assertEqual(list(Claim.objects.values_list("holder", flat=True)), ["b"]) + def test_a_claim_of_a_process_that_ended_does_not_block(self): + ended = subprocess.Popen(["true"]) + ended.wait() + node = claims._node() + Claim.objects.create(resource="instance:u1", mode=X, holder=f"{node}:{ended.pid}:1", + expires_at=timezone.now() + claims.TRANSIENT_TTL) + claims.admit({"instance:u1": X}, holder="b") + self.assertEqual(list(Claim.objects.values_list("holder", flat=True)), ["b"]) + + def test_a_claim_of_a_running_process_or_another_installation_blocks(self): + ended = subprocess.Popen(["true"]) + ended.wait() + for holder in (claims.holder_id(), f"other-node:{ended.pid}:1"): + with self.subTest(holder=holder): + Claim.objects.create(resource="instance:u1", mode=X, holder=holder, + expires_at=timezone.now() + claims.TRANSIENT_TTL) + with self.assertRaises(claims.Busy): + claims.admit({"instance:u1": X}, holder="b") + Claim.objects.all().delete() + def test_keys_are_canonical(self): self.assertEqual(claims.compute_key(3), "compute:3") self.assertEqual(claims.instance_key("u-1"), "instance:u-1")