Skip to content

[POC] Using fasteners to control parallel execution of Conan #18253

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 21 commits into
base: develop2
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 28 additions & 23 deletions conan/api/subapi/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
from conan.internal.model.settings import Settings
from conan.internal.hook_manager import HookManager
from conan.internal.util.files import load, save, rmdir, remove
from conan.internal.util.semaphore import interprocess_write_lock


class ConfigAPI:
Expand All @@ -41,9 +42,11 @@ def install(self, path_or_url, verify_ssl, config_type=None, args=None,
from conan.internal.api.config.config_installer import configuration_install
cache_folder = self.conan_api.cache_folder
requester = self.conan_api.remotes.requester
configuration_install(cache_folder, requester, path_or_url, verify_ssl, config_type=config_type, args=args,
source_folder=source_folder, target_folder=target_folder)
self.conan_api.reinit()
with interprocess_write_lock(self.conan_api):
configuration_install(cache_folder, requester, path_or_url, verify_ssl, config_type=config_type, args=args,
source_folder=source_folder, target_folder=target_folder)
self.conan_api.reinit()


def install_pkg(self, ref, lockfile=None, force=False, remotes=None, profile=None):
ConanOutput().warning("The 'conan config install-pkg' is experimental",
Expand Down Expand Up @@ -97,14 +100,15 @@ def install_pkg(self, ref, lockfile=None, force=False, remotes=None, profile=Non
from conan.internal.api.config.config_installer import configuration_install
cache_folder = self.conan_api.cache_folder
requester = self.conan_api.remotes.requester
configuration_install(cache_folder, requester, uri=pkg.conanfile.package_folder, verify_ssl=False,
config_type="dir", ignore=["conaninfo.txt", "conanmanifest.txt"])
# We save the current package full reference in the file for future
# And for ``package_id`` computation
config_versions = {ref.split("/", 1)[0]: ref for ref in config_versions}
config_versions[pkg.pref.ref.name] = pkg.pref.repr_notime()
save(config_version_file, json.dumps({"config_version": list(config_versions.values())}))
self.conan_api.reinit()
with interprocess_write_lock(self.conan_api):
configuration_install(cache_folder, requester, uri=pkg.conanfile.package_folder, verify_ssl=False,
config_type="dir", ignore=["conaninfo.txt", "conanmanifest.txt"])
# We save the current package full reference in the file for future
# And for ``package_id`` computation
config_versions = {ref.split("/", 1)[0]: ref for ref in config_versions}
config_versions[pkg.pref.ref.name] = pkg.pref.repr_notime()
save(config_version_file, json.dumps({"config_version": list(config_versions.values())}))
self.conan_api.reinit()
return pkg.pref

def get(self, name, default=None, check_type=None):
Expand Down Expand Up @@ -208,18 +212,19 @@ def appending_recursive_dict_update(d, u):
def clean(self):
contents = os.listdir(self.home())
packages_folder = self.global_conf.get("core.cache:storage_path") or os.path.join(self.home(), "p")
for content in contents:
content_path = os.path.join(self.home(), content)
if content_path == packages_folder:
continue
ConanOutput().debug(f"Removing {content_path}")
if os.path.isdir(content_path):
rmdir(content_path)
else:
remove(content_path)
self.conan_api.reinit()
# CHECK: This also generates a remotes.json that is not there after a conan profile show?
self.conan_api.migrate()
with interprocess_write_lock(self.conan_api):
for content in contents:
content_path = os.path.join(self.home(), content)
if content_path == packages_folder:
continue
ConanOutput().debug(f"Removing {content_path}")
if os.path.isdir(content_path):
rmdir(content_path)
else:
remove(content_path)
self.conan_api.reinit()
# CHECK: This also generates a remotes.json that is not there after a conan profile show?
self.conan_api.migrate()

def set_core_confs(self, core_confs):
confs = ConfDefinition()
Expand Down
6 changes: 6 additions & 0 deletions conan/internal/cache/cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,12 @@ def temp_folder(self):
# TODO: Improve the path definitions, this is very hardcoded
return os.path.join(self._base_folder, "t")

@property
def filelock_folder(self):
""" Folder reserved for file locks on parallel operations
"""
return os.path.join(self._base_folder, "filelock")

@property
def builds_folder(self):
return os.path.join(self._base_folder, "b")
Expand Down
110 changes: 110 additions & 0 deletions conan/internal/util/semaphore.py
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The fastener has context manager support as well, but we would expose it for any part in the code, instead of using this proxy.

Original file line number Diff line number Diff line change
@@ -0,0 +1,110 @@
""" The semaphore module provides inter-process locking mechanisms to ensure Conan commands can
run concurrently without conflicts.

It uses the fasteners library to create and manage locks across multiple processes. Thus, this
module is a proxy in case the project need to use a different library in the future.
"""
import os
from datetime import datetime
from typing import Any

import fasteners

from conan.errors import ConanException
from conan.api.output import ConanOutput
from contextlib import contextmanager
from conan.internal.cache.cache import PkgCache


CONAN_SEMAPHORE_FILELOCK = "conan_semaphore.lock"


def _filelock_path(conan_api: Any) -> str:
""" Get the path to the interprocess file lock.

:param cache_folder: ConanAPI cache folder path
:return: Path to the file lock in Conan cache temporary directory
"""
cache = PkgCache(conan_api.cache_folder, conan_api.config.global_conf)
return os.path.join(cache.filelock_folder, CONAN_SEMAPHORE_FILELOCK)


@contextmanager
def interprocess_lock(conan_api: Any) -> None:
""" Context manager to acquire an interprocess lock.

This method uses the fasteners library to create an interprocess lock, and serves as a proxy
for the library. The lock is acquired using the InterProcessLock class, which allows multiple
processes to safely access shared resources. The lock is released automatically when the
context manager exits.

:param conan_api: ConanAPI instance
:return: None
"""
filelock_path = _filelock_path(conan_api)
lock = fasteners.InterProcessLock(filelock_path)
pid = os.getpid()
try:
ConanOutput().debug(f"{datetime.now()} [{pid}]: Acquiring semaphore lock.")
lock.acquire()
ConanOutput().debug(f"{datetime.now()} [{pid}]: Semaphore has been locked.")
yield
except Exception as error:
raise ConanException(f"Failed to acquire interprocess lock: {error}")
finally:
lock.release()
ConanOutput().debug(f"{datetime.now()} [{pid}]: Semaphore has been released.")


@contextmanager
def interprocess_write_lock(conan_api: Any) -> None:
""" Context manager to acquire an interprocess write lock.

This method uses the fasteners library to create an interprocess write lock, and serves as a
proxy for the library. The lock is acquired using the InterProcessReaderWriterLock class,
which allows multiple processes to safely access shared resources. The lock is released
automatically when the context manager exits.

:param conan_api: ConanAPI instance
:return: None
"""
filelock_path = _filelock_path(conan_api)
lock = fasteners.InterProcessReaderWriterLock(filelock_path)
pid = os.getpid()
try:
ConanOutput().debug(f"{datetime.now()} [{pid}]: Acquiring semaphore write lock.")
lock.acquire_write_lock()
ConanOutput().debug(f"{datetime.now()} [{pid}]: Semaphore write has been locked.")
yield
except Exception as error:
raise ConanException(f"Failed to acquire interprocess write lock: {error}")
finally:
lock.release_write_lock()
ConanOutput().debug(f"{datetime.now()} [{pid}]: Semaphore write has been released.")


@contextmanager
def interprocess_read_lock(conan_api: Any) -> None:
""" Context manager to acquire an interprocess read lock.

This method uses the fasteners library to create an interprocess read lock, and serves as a
proxy for the library. The lock is acquired using the InterProcessReaderWriterLock class,
which allows multiple processes to safely access shared resources. The lock is released
automatically when the context manager exits.

:param conan_api: ConanAPI instance
:return: None
"""
filelock_path = _filelock_path(conan_api)
lock = fasteners.InterProcessReaderWriterLock(filelock_path)
pid = os.getpid()
try:
ConanOutput().debug(f"{datetime.now()} [{pid}]: Acquiring semaphore read lock.")
lock.acquire_read_lock()
ConanOutput().debug(f"{datetime.now()} [{pid}]: Semaphore read has been locked.")
yield
except Exception as error:
raise ConanException(f"Failed to acquire interprocess read lock: {error}")
finally:
lock.release_read_lock()
ConanOutput().debug(f"{datetime.now()} [{pid}]: Semaphore read has been released.")
2 changes: 1 addition & 1 deletion conans/requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ urllib3>=1.26.6, <2.1
colorama>=0.4.3, <0.5.0
PyYAML>=6.0, <7.0
patch-ng>=1.18.0, <1.19
fasteners>=0.15
fasteners>=0.19, <1.0
distro>=1.4.0, <=1.8.0; platform_system == 'Linux' or platform_system == 'FreeBSD'
Jinja2>=3.0, <4.0.0
python-dateutil>=2.8.0, <3
59 changes: 59 additions & 0 deletions test/integration/parallel/test_parallel_cache.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
import os
import subprocess
import threading

from conan.test.utils.test_files import temp_folder
from conan.test.utils.tools import TestClient


def _run_config_install(cmd, env, cwd, results, index) -> None:
"""
Run the command in a subprocess and store the return code in the results list.
"""
completed = subprocess.run(
cmd,
env=env,
cwd=cwd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
results[index] = completed.returncode
if completed.returncode:
print(f"[worker {index}] stderr:\n{completed.stderr}")


def test_parallel_config_subprocess():
"""Validate that subprocesses can run concurrently without issues.

This test starts 30 separate subprocesses, each running the `conan config install` command.
No command should fail, and the cache should be updated correctly.
"""
workers = 30

extra_folder = temp_folder(path_with_spaces=False)
cache_folder = temp_folder(path_with_spaces=False)
env = os.environ.copy()
env["CONAN_HOME"] = cache_folder

test_client = TestClient(cache_folder=cache_folder)
test_client.run("profile detect --force")
test_client.save({os.path.join(extra_folder, "profiles", "foobar"): "include(default)"})
cmd = ["conan", "config", "install", "-vvv", extra_folder, "--type=dir"]

threads = []
return_codes = [None] * workers

for index in range(workers):
thread = threading.Thread(
target=_run_config_install,
args=(cmd, env, os.getcwd(), return_codes, index),
daemon=True, # dies with the main program
)
thread.start()
threads.append(thread)

for thread in threads:
thread.join()

assert all(rc == 0 for rc in return_codes), f"Some subprocesses failed: {return_codes}"
Loading