Coverage for misc/fairlock/fairlock.py: 95%

1import os 

2import socket 

3import inspect 

4import time 

5 

6SOCKDIR = "/run/fairlock" 

7START_SERVICE_TIMEOUT_SECS = 2 

8 

9class SingletonWithArgs(type): 

10 _instances = {} 

11 _init = {} 

12 

13 def __init__(cls, name, bases, dct): 

14 cls._init[cls] = dct.get('__init__', None) 

15 

16 def __call__(cls, *args, **kwargs): 

17 init = cls._init[cls] 

18 if init is not None: 18 ↛ 22line 18 didn't jump to line 22, because the condition on line 18 was never false

19 key = (cls, frozenset( 

20 inspect.getcallargs(init, None, *args, **kwargs).items())) 

21 else: 

22 key = cls 

23 

24 if key not in cls._instances: 

25 cls._instances[key] = super(SingletonWithArgs, cls).__call__(*args, **kwargs) 

26 return cls._instances[key] 

27 

28class FairlockDeadlock(Exception): 

29 pass 

30 

31class FairlockServiceTimeout(Exception): 

32 pass 

33 

34class Fairlock(metaclass=SingletonWithArgs): 

35 def __init__(self, name): 

36 self.name = name 

37 self.sockname = os.path.join(SOCKDIR, name) 

38 self.connected = False 

39 self.sock = None 

40 

41 def _ensure_service(self): 

42 service=f"fairlock@{self.name}.service" 

43 os.system(f"/usr/bin/systemctl start {service}") 

44 timeout = time.time() + START_SERVICE_TIMEOUT_SECS 

45 time.sleep(0.1) 

46 while os.system(f"/usr/bin/systemctl --quiet is-active {service}") != 0: 

47 time.sleep(0.1) 

48 if time.time() > timeout: 

49 raise FairlockServiceTimeout(f"Timed out starting service {service}") 

50 

51 def _connect_and_recv(self): 

52 while True: 

53 self.sock.connect(self.sockname) 

54 # Merely being connected is not enough. Read a small blob of data. 

55 b = self.sock.recv(10) 

56 if len(b) > 0: 56 ↛ 61line 56 didn't jump to line 61, because the condition on line 56 was never false

57 return True 

58 # If we got a zero-length return, it means the service exited while we 

59 # were waiting. Any timeout we put here would be a max wait time to acquire 

60 # the lock, which is dangerous. 

61 self._ensure_service() 

62 

63 def __enter__(self): 

64 if self.connected: 

65 raise FairlockDeadlock(f"Deadlock on Fairlock resource '{self.name}'") 

66 

67 self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) 

68 self.sock.setblocking(True) 

69 try: 

70 self._connect_and_recv() 

71 except (FileNotFoundError, ConnectionRefusedError): 

72 self._ensure_service() 

73 self._connect_and_recv() 

74 

75 self.sock.send(f'{os.getpid()} - {time.monotonic()}'.encode()) 

76 self.connected = True 

77 return self 

78 

79 def __exit__(self, type, value, traceback): 

80 self.sock.close() 

81 self.sock = None 

82 self.connected = False 

83 return False 

84