Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/lockfile/linklockfile.py: 24%

1from __future__ import absolute_import 

2 

3import time 

4import os 

5 

6from . import (LockBase, LockFailed, NotLocked, NotMyLock, LockTimeout, 

7 AlreadyLocked) 

8 

9 

10class LinkLockFile(LockBase): 

11 """Lock access to a file using atomic property of link(2). 

12 

13 >>> lock = LinkLockFile('somefile') 

14 >>> lock = LinkLockFile('somefile', threaded=False) 

15 """ 

16 

17 def acquire(self, timeout=None): 

18 try: 

19 open(self.unique_name, "wb").close() 

20 except IOError: 

21 raise LockFailed("failed to create %s" % self.unique_name) 

22 

23 timeout = timeout if timeout is not None else self.timeout 

24 end_time = time.time() 

25 if timeout is not None and timeout > 0: 

26 end_time += timeout 

27 

28 while True: 

29 # Try and create a hard link to it. 

30 try: 

31 os.link(self.unique_name, self.lock_file) 

32 except OSError: 

33 # Link creation failed. Maybe we've double-locked? 

34 nlinks = os.stat(self.unique_name).st_nlink 

35 if nlinks == 2: 

36 # The original link plus the one I created == 2. We're 

37 # good to go. 

38 return 

39 else: 

40 # Otherwise the lock creation failed. 

41 if timeout is not None and time.time() > end_time: 

42 os.unlink(self.unique_name) 

43 if timeout > 0: 

44 raise LockTimeout("Timeout waiting to acquire" 

45 " lock for %s" % 

46 self.path) 

47 else: 

48 raise AlreadyLocked("%s is already locked" % 

49 self.path) 

50 time.sleep(timeout is not None and timeout / 10 or 0.1) 

51 else: 

52 # Link creation succeeded. We're good to go. 

53 return 

54 

55 def release(self): 

56 if not self.is_locked(): 

57 raise NotLocked("%s is not locked" % self.path) 

58 elif not os.path.exists(self.unique_name): 

59 raise NotMyLock("%s is locked, but not by me" % self.path) 

60 os.unlink(self.unique_name) 

61 os.unlink(self.lock_file) 

62 

63 def is_locked(self): 

64 return os.path.exists(self.lock_file) 

65 

66 def i_am_locking(self): 

67 return (self.is_locked() and 

68 os.path.exists(self.unique_name) and 

69 os.stat(self.unique_name).st_nlink == 2) 

70 

71 def break_lock(self): 

72 if os.path.exists(self.lock_file): 

73 os.unlink(self.lock_file)