Mercurial > repos > guerler > springsuite

comparison planemo/lib/python3.7/site-packages/lockfile/linklockfile.py @ 1:56ad4e20f292 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:32:28 -0400
parents
children
comparison
equal deleted inserted replaced
0:d30785e31577 1:56ad4e20f292
1 from __future__ import absolute_import
2
3 import time
4 import os
5
6 from . import (LockBase, LockFailed, NotLocked, NotMyLock, LockTimeout,
7 AlreadyLocked)
8
9
10 class 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)