##// END OF EJS Templates
revlog: avoid possible collision between directory and temporary index...
revlog: avoid possible collision between directory and temporary index Since 6.4, we create a temporary index file to write the split data without overwriting the inline version too early. However, the store encoding does not prevent these new `.i.s` file to collide with a directory with the same name. While the odds for such a collision to happens are fairly low, the collision would prevent Mercurial from working. The store encoding have a mitigation solution in place to prevent such collisions from happening for `.i` and `.d` files, but not for other extensions. We cannot update this encoding scheme to solve the issue since it would diverge from older version of Mercurial. Instead, we create an alternative directory tree dedicated to such files. The use of the `.i` extension combined with store encoding will prevent collisions there.

File last commit:

r51109:8e0d823e stable
r51555:12f13b13 stable
Show More
__init__.py
53 lines | 1.5 KiB | text/x-python | PythonLexer
import os
import time
# work around check-code complains
#
# This is a simple log level module doing simple test related work, we can't
# import more things, and we do not need it.
environ = getattr(os, 'environ')
def wait_on_cfg(ui, cfg, timeout=10):
"""synchronize on the `cfg` config path
Use this to synchronize commands during race tests.
"""
full_config = b'sync.' + cfg
wait_config = full_config + b'-timeout'
sync_path = ui.config(b'devel', full_config)
if sync_path is not None:
timeout = ui.config(b'devel', wait_config)
ready_path = sync_path + b'.waiting'
write_file(ready_path)
wait_file(sync_path, timeout=timeout)
def _timeout_factor():
"""return the current modification to timeout"""
default = int(environ.get('HGTEST_TIMEOUT_DEFAULT', 360))
current = int(environ.get('HGTEST_TIMEOUT', default))
if current == 0:
return 1
return current / float(default)
def wait_file(path, timeout=10):
timeout *= _timeout_factor()
start = time.time()
while not os.path.exists(path):
if timeout and time.time() - start > timeout:
raise RuntimeError(b"timed out waiting for file: %s" % path)
time.sleep(0.01)
def write_file(path, content=b''):
if content:
write_path = b'%s.tmp' % path
else:
write_path = path
with open(write_path, 'wb') as f:
f.write(content)
if path != write_path:
os.rename(write_path, path)