/usr/lib/python2.7/dist-packages/obnamlib/lockmgr.py is in obnam 1.19.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | # Copyright 2012-2015 Lars Wirzenius
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import time
import obnamlib
class LockManager(object):
'''Lock and unlock sets of directories at once.'''
def __init__(self, fs, timeout, client):
self._fs = fs
self.timeout = timeout
self._got_locks = []
def _time(self): # pragma: no cover
return time.time()
def _sleep(self): # pragma: no cover
time.sleep(1)
def sort(self, dirnames):
def bytelist(s):
return [ord(s) for s in str(s)]
return sorted(dirnames, key=bytelist)
def get_lock_name(self, dirname):
return os.path.join(dirname, 'lock')
def _lock_one(self, dirname):
started = self._time()
while True:
lock_name = self.get_lock_name(dirname)
try:
self._fs.lock(lock_name)
except obnamlib.LockFail:
if self._time() - started >= self.timeout:
raise obnamlib.LockFail(
lock_name=lock_name,
reason='timeout')
else:
self._got_locks.append(dirname)
return
self._sleep()
def _unlock_one(self, dirname):
self._fs.unlock(self.get_lock_name(dirname))
if dirname in self._got_locks:
self._got_locks.remove(dirname)
def is_locked(self, dirname):
'''Is the given directory locked?
Note the usual race conditions between testing and locking.
'''
return self._fs.exists(self.get_lock_name(dirname))
def got_lock(self, dirname):
'''Does this lock manager hold the lock for a directory?'''
return dirname in self._got_locks
def lock(self, dirnames):
'''Lock ALL the directories.'''
we_locked = []
for dirname in self.sort(dirnames):
try:
self._lock_one(dirname)
except obnamlib.LockFail:
self.unlock(we_locked)
raise
else:
we_locked.append(dirname)
def unlock(self, dirnames):
'''Unlock ALL the directories.'''
for dirname in self.sort(dirnames):
self._unlock_one(dirname)
def force(self, dirnames):
'''Force all the directories to be unlocked.
This works even if they weren't locked already.
'''
for dirname in self.sort(dirnames):
self._force_one(dirname)
def _force_one(self, dirname):
lock_name = self.get_lock_name(dirname)
if self._fs.exists(lock_name):
self._fs.remove(lock_name)
if dirname in self._got_locks:
self._got_locks.remove(dirname)
|