79 lines
3.5 KiB
Python
79 lines
3.5 KiB
Python
"""Independent bounded network-role observer; HostWatch remains unchanged."""
|
|
import array
|
|
import socket
|
|
import struct
|
|
from c01_network_settings_owner import PROGRESS, now
|
|
from c01_network_settings_nm import require
|
|
|
|
|
|
class Watch:
|
|
def __init__(self, control, application, pid):
|
|
self.control, self.application, self.pid = control, application, pid
|
|
control.setblocking(False)
|
|
self.state, self.sequence, self.epoch = 'starting', 0, None
|
|
self.deadline, self.heartbeat = None, now() + 10_000_000_000
|
|
|
|
def revoke(self, reason):
|
|
if self.state != 'failed':
|
|
self.state = 'failed'
|
|
try:
|
|
self.application.shutdown(socket.SHUT_RDWR)
|
|
except OSError:
|
|
pass
|
|
self.control.close()
|
|
|
|
def release_reaped(self):
|
|
self.revoke('reaped')
|
|
self.application.close()
|
|
|
|
def receive(self):
|
|
if self.state == 'failed':
|
|
return False
|
|
stamp = now()
|
|
if self.heartbeat is not None and stamp >= self.heartbeat or self.deadline is not None and stamp >= self.deadline:
|
|
self.revoke('deadline')
|
|
return False
|
|
try:
|
|
raw, anc, flags, _ = self.control.recvmsg(PROGRESS.size + 1, socket.CMSG_SPACE(12) + socket.CMSG_SPACE(64), socket.MSG_DONTWAIT | socket.MSG_CMSG_CLOEXEC)
|
|
except BlockingIOError:
|
|
return True
|
|
except OSError:
|
|
self.revoke('closed')
|
|
return False
|
|
credentials, unexpected = [], False
|
|
for level, kind, data in anc:
|
|
if (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS):
|
|
rights = array.array('i')
|
|
rights.frombytes(data[:len(data) - len(data) % rights.itemsize])
|
|
for fd in rights:
|
|
import os
|
|
os.close(fd)
|
|
unexpected = True
|
|
elif (level, kind) == (socket.SOL_SOCKET, socket.SCM_CREDENTIALS) and len(data) == 12:
|
|
credentials.append(struct.unpack('3i', data))
|
|
else:
|
|
unexpected = True
|
|
try:
|
|
require(not unexpected and credentials == [(self.pid, 0, 0)] and flags & ~socket.MSG_CMSG_CLOEXEC == 0 and len(raw) == PROGRESS.size)
|
|
version, event, sequence, deadline, epoch = PROGRESS.unpack(raw)
|
|
require(version == 1 and epoch != bytes(16))
|
|
if event == 1:
|
|
require(self.state == 'starting' and sequence == deadline == 0)
|
|
self.state, self.epoch, self.heartbeat = 'idle', epoch, stamp + 2_000_000_000
|
|
elif event == 5:
|
|
require(self.state == 'idle' and epoch == self.epoch and sequence == deadline == 0)
|
|
self.heartbeat = stamp + 2_000_000_000
|
|
elif event == 2:
|
|
require(self.state == 'idle' and epoch == self.epoch and sequence > self.sequence and stamp < deadline <= stamp + 45_000_000_000)
|
|
self.state, self.sequence, self.deadline, self.heartbeat = 'busy', sequence, deadline, stamp + 2_000_000_000
|
|
else:
|
|
require(self.state == 'busy' and epoch == self.epoch and sequence == self.sequence and deadline == self.deadline)
|
|
require(event in (3, 4))
|
|
if event == 4:
|
|
self.state, self.deadline, self.heartbeat = 'idle', None, stamp + 2_000_000_000
|
|
else:
|
|
self.heartbeat = stamp + 2_000_000_000
|
|
except (AssertionError, ValueError):
|
|
self.revoke('protocol')
|
|
return self.state != 'failed'
|