Socket 激活把监听器交给兼容的服务

from pathlib import Path
from tempfile import TemporaryDirectory
import json,socket,subprocess,time,uuid

name='xbcatm-socket-'+uuid.uuid4().hex[:8];service=name+'.service';unit=name+'.socket'
paths={n:Path('/run/systemd/system')/n for n in (service,unit)}
assert all(not p.exists() for p in paths.values())
def run(*args):
    result=subprocess.run(args,capture_output=True,text=True,timeout=10)
    assert result.returncode==0,(args,result.stderr)
    return result.stdout.strip()
def value(key):return run('systemctl','show',service,'-p',key,'--value')
def wait_inactive():
    deadline=time.monotonic()+5
    while value('ActiveState')!='inactive':
        assert time.monotonic()<deadline,'service did not exit';time.sleep(0.02)
with TemporaryDirectory(prefix=name+'-',dir='/run') as directory:
    root=Path(directory);endpoint=root/'listener.sock';worker=root/'worker.py'
    worker.write_text(r'''from pathlib import Path
import errno,json,os,socket
root=Path(__file__).parent
assert int(os.environ['LISTEN_PID'])==os.getpid()
assert int(os.environ['LISTEN_FDS'])==1
with socket.socket(fileno=3) as listener:
    assert listener.getsockopt(socket.SOL_SOCKET,socket.SO_ACCEPTCONN)==1
    with socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as duplicate:
        try:duplicate.bind(str(root/'listener.sock'))
        except OSError as error:assert error.errno==errno.EADDRINUSE
        else:raise AssertionError('unexpected duplicate bind success')
    connection,_=listener.accept()
    with connection,connection.makefile('rb') as incoming:
        message=incoming.readline(256);assert message.endswith(b'\n')
        connection.sendall(b'ack '+message)
    (root/(str(os.getpid())+'.json')).write_text(json.dumps({'pid':os.getpid(),'fd':3,'message':message.decode().strip()}))
''')
    paths[unit].write_text(f'[Socket]\nListenStream={endpoint}\nAccept=no\nSocketMode=0600\nRemoveOnStop=yes\n')
    paths[service].write_text(f'[Service]\nType=exec\nExecStart=/usr/bin/python3 {worker}\n')
    try:
        run('systemd-analyze','verify',*[str(p) for p in paths.values()]);run('systemctl','daemon-reload')
        run('systemctl','start',unit)
        assert endpoint.is_socket() and value('ActiveState')=='inactive' and value('MainPID')=='0'
        print('socket listening while service has no process')
        for message in ('first','second'):
            with socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as client:
                client.settimeout(5);client.connect(str(endpoint));client.sendall((message+'\n').encode())
                with client.makefile('rb') as response:assert response.readline()==('ack '+message+'\n').encode()
            wait_inactive();assert endpoint.is_socket() and value('Result')=='success'
        records=[json.loads(p.read_text()) for p in root.glob('*.json')]
        assert len(records)==2 and len({row['pid'] for row in records})==2
        assert {row['message'] for row in records}=={'first','second'}
        assert all(row['fd']==3 for row in records)
        print('two requests: two deliberate one-request workers; each uses inherited listening fd 3')
        print('rebinding the occupied socket fails; manager retains listener between workers')
        run('systemctl','stop',unit);assert not endpoint.exists()
        print('stopping socket removes its endpoint with RemoveOnStop=yes')
    finally:
        run('systemctl','stop',unit,service)
        for name,path in paths.items():
            if run('systemctl','show',name,'-p','ActiveState','--value')=='failed':run('systemctl','reset-failed',name)
            path.unlink()
        run('systemctl','daemon-reload')