mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 11:16:47 +00:00
fix(claims): a request's claim no longer blocks once its process has ended
A request removes its transient claims in a finally. A gunicorn worker that exits past its timeout ends its other threads' requests too, and an OOM kill or a container restart ends all of them, so their claims stayed until they expired 11 minutes later. A VM delete killed this way held its compute exclusively: every VM action, job and list refresh on that compute was refused as busy, and nothing in the panel could clear it. The holder now names the installation by its node id (as the worker does), which stays the same when the container is recreated, instead of the hostname. A conflicting claim whose holder is a process of this installation that no longer runs is removed during admission. A claim of another installation, or of a PID that is in use again, still waits for its expiry.
This commit is contained in:
parent
4c0efe9db0
commit
2991afc9b9
2 changed files with 52 additions and 2 deletions
|
|
@ -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)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue