forked from conda/conda
-
Notifications
You must be signed in to change notification settings - Fork 0
/
lock.py
80 lines (67 loc) · 2.42 KB
/
lock.py
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
# (c) 2012-2013 Continuum Analytics, Inc. / http://continuum.io
# All Rights Reserved
#
# conda is distributed under the terms of the BSD 3-clause license.
# Consult LICENSE.txt or http://opensource.org/licenses/BSD-3-Clause.
"""
Tools for working with locks
A lock is just an empty directory. We use directories because this lets us use
the race condition-proof os.makedirs.
For now, there is one global lock for all of conda, because some things happen
globally (such as downloading packages).
We don't raise an error if the lock is named with the current PID
"""
import os
import logging
from os.path import join
import glob
from time import sleep
LOCKFN = '.conda_lock'
stdoutlog = logging.getLogger('stdoutlog')
class Locked(object):
"""
Context manager to handle locks.
"""
def __init__(self, path):
self.path = path
self.end = "-" + str(os.getpid())
self.lock_path = join(self.path, LOCKFN + self.end)
self.pattern = join(self.path, LOCKFN + '-*')
self.remove = True
def __enter__(self):
retries = 10
# Keep the string "LOCKERROR" in this string so that external
# programs can look for it.
lockstr = ("""\
LOCKERROR: It looks like conda is already doing something.
The lock %s was found. Wait for it to finish before continuing.
If you are sure that conda is not running, remove it and try again.
You can also use: $ conda clean --lock\n""" % self.lock_path)
sleeptime = 1
while retries:
files = glob.glob(self.pattern)
if files and not files[0].endswith(self.end):
stdoutlog.info(lockstr)
stdoutlog.info("Sleeping for %s seconds\n" % sleeptime)
sleep(sleeptime)
sleeptime *= 2
retries -= 1
else:
break
else:
stdoutlog.error("Exceeded max retries, giving up")
raise RuntimeError(lockstr)
if not files:
try:
os.makedirs(self.lock_path)
except OSError:
pass
else: # PID lock already here --- someone else will remove it.
self.remove = False
def __exit__(self, exc_type, exc_value, traceback):
if self.remove:
for path in self.lock_path, self.path:
try:
os.rmdir(path)
except OSError:
pass