proxystore.endpoint.directory

proxystore/endpoint/directory.py

Endpoint directory layout and files.

CONNECTION_VERSION module-attribute

CONNECTION_VERSION = 1

Format version of the connection file.

ConnectionInfo

Bases: VersionedFile

Information that clients use to connect to a running endpoint.

The endpoint writes this to the connection file in its directory each time it starts and removes it when it stops (see EndpointDir.write_connection()).

Attributes:

• version (int) –

  Format version of the connection file.

• host (str) –

  Host address the endpoint is listening on.

• port (int) –

  Port the endpoint is listening on.

• token (EndpointToken) –

  Token that the client and endpoint prove they know.

• tls_fingerprint (str | None) –

  SHA-256 fingerprint of the endpoint's TLS certificate or None if the endpoint does not use TLS.

• hostname (str) –

  Name of the machine the endpoint is running on.

• pid (int) –

  Process ID of the endpoint on that machine.

EndpointOptions

Bases: TypedDict

Options of a new endpoint.

See EndpointConfig for the meaning and default of each option.

EndpointStatus

Bases: Enum

Status of an endpoint.

See EndpointDir.status().

RUNNING class-attribute instance-attribute

RUNNING = enum.auto()

Endpoint is running.

STOPPED class-attribute instance-attribute

STOPPED = enum.auto()

Endpoint is not running.

STALE class-attribute instance-attribute

STALE = enum.auto()

Endpoint on this host stopped without removing its connection file.

This happens if the endpoint process was killed or crashed. Starting or stopping the endpoint removes the stale connection file.

OTHER_HOST class-attribute instance-attribute

OTHER_HOST = enum.auto()

Endpoint was started on another host and may still be running there.

The connection file was written by an endpoint on another host which shares the endpoint directory. Whether the endpoint is still running on that host cannot be checked from this host.

EndpointLock

EndpointLock(path: str)

Advisory lock held by a running endpoint.

A running endpoint holds an exclusive lock (see fcntl.flock()) on the lock file in its directory for as long as it runs. The operating system releases the lock when the process exits, even if the process crashes, so the lock reliably shows if the endpoint is running on this host. Whether the lock is also visible to other hosts depends on the file system of the endpoint directory.

Some file systems (e.g., some network or parallel file systems) do not support locks. Then, acquiring the lock always succeeds, supported is False, and the status of the endpoint is determined from the PID in its connection file instead.

Attributes:

• path –

  Path of the lock file.

• supported –

  If the file system supports locks. This is True until locking fails because locks are not supported.

Parameters:

• path (str) –

  Path of the lock file.

Source code in proxystore/endpoint/directory.py

def __init__(self, path: str) -> None:
    self.path = path
    self.supported = True
    self._fd: int | None = None

held property

held: bool

This lock object holds the lock.

acquire

acquire() -> None

Acquire the lock without waiting.

Raises:

• RuntimeError –

  If this lock object already holds the lock.

• EndpointRunningError –

  If another process (or another lock object in this process) holds the lock.

Source code in proxystore/endpoint/directory.py

def acquire(self) -> None:
    """Acquire the lock without waiting.

    Raises:
        RuntimeError: If this lock object already holds the lock.
        EndpointRunningError: If another process (or another lock object
            in this process) holds the lock.
    """
    if self._fd is not None:
        raise RuntimeError('The lock is already held.')
    fd = os.open(self.path, os.O_RDWR | os.O_CREAT, 0o600)
    try:
        deadline = time.monotonic() + self._ACQUIRE_TIMEOUT
        while (acquired := self._flock(fd, fcntl.LOCK_EX)) is False:
            if time.monotonic() >= deadline:
                break
            time.sleep(0.005)
        if acquired is False:
            raise EndpointRunningError(
                'Endpoint '
                f'{os.path.basename(os.path.dirname(self.path))} is '
                f'already running (its lock {self.path} is held by '
                'another process).',
            )
    except BaseException:
        os.close(fd)
        raise
    self._fd = fd

release

release() -> None

Release the lock if it is held.

Source code in proxystore/endpoint/directory.py

def release(self) -> None:
    """Release the lock if it is held."""
    if self._fd is None:
        return
    fd, self._fd = self._fd, None
    if self.supported:
        with contextlib.suppress(OSError):
            fcntl.flock(fd, fcntl.LOCK_UN)
    os.close(fd)

is_locked

is_locked() -> bool | None

Check if the lock is held.

Returns:

• bool | None –

  True if the lock is held by any lock object, False if the lock is not held, or None if locks are not supported.

Source code in proxystore/endpoint/directory.py

def is_locked(self) -> bool | None:
    """Check if the lock is held.

    Returns:
        `True` if the lock is held by any lock object, `False` if the \
        lock is not held, or `None` if locks are not supported.
    """
    try:
        fd = os.open(self.path, os.O_RDWR)
    except FileNotFoundError:
        return False if self.supported else None
    try:
        # A shared lock is used so concurrent checks do not conflict
        # with each other, only with the exclusive lock of the endpoint.
        acquired = self._flock(fd, fcntl.LOCK_SH)
        if acquired is None:
            return None
        if acquired:
            fcntl.flock(fd, fcntl.LOCK_UN)
        return not acquired
    finally:
        os.close(fd)

EndpointDir dataclass

EndpointDir(path: str)

Directory of an endpoint.

An endpoint directory contains the endpoint's configuration and the files created while it runs (e.g., its log and the connection file that clients use to connect).

Example

# Create a new endpoint with a new secret key
endpoint_dir = EndpointDir.create('my-ep', port=8765)
# Or open an existing endpoint
endpoint_dir = EndpointDir.from_name('my-ep')
config = endpoint_dir.read_config()

Attributes:

• path (str) –

  Path of the directory.

name property

name: str

Name of the endpoint (i.e., the name of the directory).

config_path property

config_path: str

Path to the endpoint configuration.

log_path property

log_path: str

Path to the log of the endpoint daemon.

lock_path property

lock_path: str

Path to the lock file held by the running endpoint.

secret_key_path property

secret_key_path: str

Path to the secret key of the endpoint.

peers_path property

peers_path: str

Path to the allowlist of peer endpoints.

peer_addrs_path property

peer_addrs_path: str

Path to the cache of peer addresses written by the endpoint.

connection_path property

connection_path: str

Path to the connection file clients use to connect.

from_name classmethod

from_name(
    name: str, proxystore_dir: str | None = None
) -> Self

Get the directory of an endpoint in a ProxyStore home directory.

The directory may not exist (e.g., because the endpoint has not been created yet).

Parameters:

• name (str) –

  Name of the endpoint.

• proxystore_dir (str | None, default: None ) –

  ProxyStore home directory. Defaults to home_dir().

Source code in proxystore/endpoint/directory.py

@classmethod
def from_name(
    cls,
    name: str,
    proxystore_dir: str | None = None,
) -> Self:
    """Get the directory of an endpoint in a ProxyStore home directory.

    The directory may not exist (e.g., because the endpoint has not been
    created yet).

    Args:
        name: Name of the endpoint.
        proxystore_dir: ProxyStore home directory. Defaults to
            [`home_dir()`][proxystore.utils.environment.home_dir].
    """
    if proxystore_dir is None:
        proxystore_dir = home_dir()
    return cls(os.path.join(proxystore_dir, name))

create classmethod

create(
    name: str,
    proxystore_dir: str | None = None,
    *,
    secret_key: SecretKey | None = None,
    port: int | None = None,
    **options: Unpack[EndpointOptions],
) -> Self

Create a new endpoint.

Creates the endpoint directory, only accessible by the owner, and writes a new secret key and the configuration of the endpoint.

Example

endpoint_dir = EndpointDir.create(
    'my-ep',
    port=8765,
    p2p=EndpointP2PConfig(relays='none'),
)

Parameters:

• name (str) –

  Name of the endpoint.

• proxystore_dir (str | None, default: None ) –

  ProxyStore home directory. Defaults to home_dir().

• secret_key (SecretKey | None, default: None ) –

  Secret key of the endpoint. A new key is generated if None.

• port (int | None, default: None ) –

  Port of the endpoint. A random port in the range [10240, 20480] is chosen if None.

• options (Unpack[EndpointOptions], default: {} ) –

  Other options of the endpoint (see EndpointOptions).

Returns:

• Self –

  The new endpoint directory.

Raises:

• EndpointExistsError –

  If an endpoint with the name already exists.

• ValueError –

  If the configuration is invalid.

Source code in proxystore/endpoint/directory.py

@classmethod
def create(
    cls,
    name: str,
    proxystore_dir: str | None = None,
    *,
    secret_key: SecretKey | None = None,
    port: int | None = None,
    **options: Unpack[EndpointOptions],
) -> Self:
    """Create a new endpoint.

    Creates the endpoint directory, only accessible by the owner, and
    writes a new secret key and the configuration of the endpoint.

    Example:
        ```python
        endpoint_dir = EndpointDir.create(
            'my-ep',
            port=8765,
            p2p=EndpointP2PConfig(relays='none'),
        )
        ```

    Args:
        name: Name of the endpoint.
        proxystore_dir: ProxyStore home directory. Defaults to
            [`home_dir()`][proxystore.utils.environment.home_dir].
        secret_key: Secret key of the endpoint. A new key is generated
            if `None`.
        port: Port of the endpoint. A random port in the range
            [10240, 20480] is chosen if `None`.
        options: Other options of the endpoint (see
            [`EndpointOptions`][proxystore.endpoint.directory.EndpointOptions]).

    Returns:
        The new endpoint directory.

    Raises:
        EndpointExistsError: If an endpoint with the name already exists.
        ValueError: If the configuration is invalid.
    """
    secret_key = SecretKey.generate() if secret_key is None else secret_key
    port = random.randint(10 * 1024, 20 * 1024) if port is None else port
    config = EndpointConfig(
        name=name,
        id=secret_key.endpoint_id,
        port=port,
        **options,
    )
    endpoint_dir = cls.from_name(name, proxystore_dir)
    os.makedirs(os.path.dirname(endpoint_dir.path), exist_ok=True)
    # Clients trust the files in the endpoint directory, so only the
    # owner can create or replace files in it.
    try:
        os.mkdir(endpoint_dir.path, mode=0o700)
    except FileExistsError:
        raise EndpointExistsError(
            f'An endpoint named {name} already exists in '
            f'{os.path.dirname(endpoint_dir.path)}.',
        ) from None
    endpoint_dir.write_secret_key(secret_key)
    # The configuration is written last because an endpoint is only
    # found (see find_all()) once it has a configuration.
    endpoint_dir.write_config(config)
    return endpoint_dir

find_all classmethod

find_all(
    proxystore_dir: str | None = None,
) -> list[tuple[Self, EndpointConfig]]

Find all endpoints with a valid configuration.

Parameters:

• proxystore_dir (str | None, default: None ) –

  ProxyStore home directory to search in. Defaults to home_dir().

Returns:

• list[tuple[Self, EndpointConfig]] –

  List of each endpoint directory and its configuration.

Source code in proxystore/endpoint/directory.py

@classmethod
def find_all(
    cls,
    proxystore_dir: str | None = None,
) -> list[tuple[Self, EndpointConfig]]:
    """Find all endpoints with a valid configuration.

    Args:
        proxystore_dir: ProxyStore home directory to search in. Defaults
            to [`home_dir()`][proxystore.utils.environment.home_dir].

    Returns:
        List of each endpoint directory and its configuration.
    """
    if proxystore_dir is None:
        proxystore_dir = home_dir()
    endpoints: list[tuple[Self, EndpointConfig]] = []
    if not os.path.isdir(proxystore_dir):
        return endpoints

    # Endpoint directories are always direct children of the home
    # directory (see from_name()).
    with os.scandir(proxystore_dir) as entries:
        paths = sorted(entry.path for entry in entries if entry.is_dir())

    for path in paths:
        endpoint_dir = cls(path)
        try:
            config = endpoint_dir.read_config()
        except (FileNotFoundError, ValueError):
            continue
        endpoints.append((endpoint_dir, config))

    return endpoints

read_config

read_config() -> EndpointConfig

Read the endpoint configuration.

Raises:

• EndpointNotFoundError –

  If the configuration file does not exist.

• EndpointConfigError –

  If the configuration contains an invalid value, cannot be parsed, or the name of the endpoint does not match the name of the directory.

Source code in proxystore/endpoint/directory.py

def read_config(self) -> EndpointConfig:
    """Read the endpoint configuration.

    Raises:
        EndpointNotFoundError: If the configuration file does not exist.
        EndpointConfigError: If the configuration contains an invalid
            value, cannot be parsed, or the name of the endpoint does
            not match the name of the directory.
    """
    try:
        config = read_model(EndpointConfig, self.config_path)
    except FileNotFoundError:
        self.check_exists()
        raise EndpointNotFoundError(
            f'Endpoint directory {self.path} does not contain a valid '
            'configuration.',
        ) from None

    # The directory name is used to find an endpoint by name, so the
    # name in the configuration must match it.
    if config.name != self.name:
        raise EndpointConfigError(
            f'The endpoint name in {self.config_path} ({config.name}) '
            f'does not match the name of the directory ({self.name}). '
            'Rename the directory or change the name in the '
            'configuration so they match.',
        )
    return config

write_config

write_config(config: EndpointConfig) -> None

Atomically write the endpoint configuration.

The directory is created, only accessible by the owner, if needed.

Parameters:

• config (EndpointConfig) –

  Configuration to write.

Source code in proxystore/endpoint/directory.py

def write_config(self, config: EndpointConfig) -> None:
    """Atomically write the endpoint configuration.

    The directory is created, only accessible by the owner, if needed.

    Args:
        config: Configuration to write.
    """
    os.makedirs(self.path, mode=0o700, exist_ok=True)
    write_model(self.config_path, config)

resolve_path

resolve_path(path: str) -> str

Resolve a path in the configuration of the endpoint.

~ is expanded to the user's home directory, and a relative path is relative to the endpoint directory.

Source code in proxystore/endpoint/directory.py

def resolve_path(self, path: str) -> str:
    """Resolve a path in the configuration of the endpoint.

    `~` is expanded to the user's home directory, and a relative path
    is relative to the endpoint directory.
    """
    path = os.path.expanduser(path)
    return path if os.path.isabs(path) else self._join(path)

lock

lock() -> EndpointLock

Get the lock held by the running endpoint.

See EndpointLock.

Source code in proxystore/endpoint/directory.py

def lock(self) -> EndpointLock:
    """Get the lock held by the running endpoint.

    See [`EndpointLock`][proxystore.endpoint.directory.EndpointLock].
    """
    return EndpointLock(self.lock_path)

write_secret_key

write_secret_key(secret_key: SecretKey) -> None

Atomically write the secret key of the endpoint.

The file is only readable by the owner.

Source code in proxystore/endpoint/directory.py

def write_secret_key(self, secret_key: SecretKey) -> None:
    """Atomically write the secret key of the endpoint.

    The file is only readable by the owner.
    """
    write_private_file(self.secret_key_path, secret_key.to_bytes())

read_secret_key

read_secret_key(endpoint_id: EndpointId) -> SecretKey

Read the secret key of the endpoint.

Parameters:

• endpoint_id (EndpointId) –

  ID the secret key must match (i.e., the ID in the configuration of the endpoint).

Raises:

• EndpointConfigError –

  If the secret key file does not exist, is malformed, or does not match the ID.

Source code in proxystore/endpoint/directory.py

def read_secret_key(self, endpoint_id: EndpointId) -> SecretKey:
    """Read the secret key of the endpoint.

    Args:
        endpoint_id: ID the secret key must match (i.e., the ID in the
            configuration of the endpoint).

    Raises:
        EndpointConfigError: If the secret key file does not exist, is
            malformed, or does not match the ID.
    """
    try:
        with open(self.secret_key_path, 'rb') as f:
            data = f.read()
    except FileNotFoundError:
        raise EndpointConfigError(
            f'Endpoint directory {self.path} does not contain a secret '
            'key. Remove the endpoint and configure it again with '
            '"proxystore-endpoint configure".',
        ) from None
    try:
        secret_key = SecretKey(data)
    except ValueError:
        raise EndpointConfigError(
            f'Secret key file at {self.secret_key_path} is malformed.',
        ) from None

    if secret_key.endpoint_id != endpoint_id:
        raise EndpointConfigError(
            f'The endpoint ID in the configuration ({endpoint_id}) '
            'does not match the secret key '
            f'({secret_key.endpoint_id}) in {self.path}.',
        )
    return secret_key

peers

peers() -> Peers

Get the peers of the endpoint.

This reads the configuration to get the ID of the endpoint.

Raises:

• EndpointNotFoundError –

  If the configuration does not exist.

• EndpointConfigError –

  If the configuration is invalid.

Source code in proxystore/endpoint/directory.py

def peers(self) -> Peers:
    """Get the peers of the endpoint.

    This reads the configuration to get the ID of the endpoint.

    Raises:
        EndpointNotFoundError: If the configuration does not exist.
        EndpointConfigError: If the configuration is invalid.
    """
    return Peers(self.peers_path, owner_id=self.read_config().id)

write_connection

write_connection(info: ConnectionInfo) -> None

Atomically write the connection file of the running endpoint.

The file is only readable by the owner because it contains the endpoint's token.

Source code in proxystore/endpoint/directory.py

def write_connection(self, info: ConnectionInfo) -> None:
    """Atomically write the connection file of the running endpoint.

    The file is only readable by the owner because it contains the
    endpoint's token.
    """
    write_model(self.connection_path, info)

read_connection

read_connection() -> ConnectionInfo

Read the connection file of the running endpoint.

Raises:

• FileNotFoundError –

  If the connection file does not exist (e.g., because the endpoint is not running).

• EndpointConfigError –

  If the connection file is malformed or has an unsupported format version.

Source code in proxystore/endpoint/directory.py

def read_connection(self) -> ConnectionInfo:
    """Read the connection file of the running endpoint.

    Raises:
        FileNotFoundError: If the connection file does not exist (e.g.,
            because the endpoint is not running).
        EndpointConfigError: If the connection file is malformed or has
            an unsupported format version.
    """
    return read_model(ConnectionInfo, self.connection_path)

remove_connection

remove_connection(
    info: ConnectionInfo | None = None,
) -> None

Remove the connection file if it exists.

Parameters:

• info (ConnectionInfo | None, default: None ) –

  Only remove the connection file if it contains this information (i.e., it was not replaced by another instance of the endpoint).

Source code in proxystore/endpoint/directory.py

def remove_connection(self, info: ConnectionInfo | None = None) -> None:
    """Remove the connection file if it exists.

    Args:
        info: Only remove the connection file if it contains this
            information (i.e., it was not replaced by another instance
            of the endpoint).
    """
    if info is not None:
        try:
            if self.read_connection() != info:
                return
        except (FileNotFoundError, ValueError):
            return
    with contextlib.suppress(FileNotFoundError):
        os.remove(self.connection_path)

status

status() -> EndpointStatus

Get the status of the endpoint.

The endpoint is running if it holds its lock (see lock()). If the file system does not support locks, the endpoint is running if the process in its connection file is running on this host.

Returns:

• EndpointStatus –

  The status of the endpoint.

Raises:

• EndpointNotFoundError –

  If the endpoint directory does not exist.

Source code in proxystore/endpoint/directory.py

def status(self) -> EndpointStatus:
    """Get the status of the endpoint.

    The endpoint is running if it holds its lock (see
    [`lock()`][proxystore.endpoint.directory.EndpointDir.lock]). If the
    file system does not support locks, the endpoint is running if the
    process in its connection file is running on this host.

    Returns:
        The status of the endpoint.

    Raises:
        EndpointNotFoundError: If the endpoint directory does not exist.
    """
    return self._status()[0]

check_stopped

check_stopped() -> None

Check that the endpoint is not running.

A stale connection file left by an endpoint which stopped unexpectedly (see EndpointStatus.STALE) is removed.

Raises:

• EndpointNotFoundError –

  If the endpoint directory does not exist.

• EndpointRunningError –

  If the endpoint is running on this host or may be running on another host.

Source code in proxystore/endpoint/directory.py

def check_stopped(self) -> None:
    """Check that the endpoint is not running.

    A stale connection file left by an endpoint which stopped
    unexpectedly (see
    [`EndpointStatus.STALE`][proxystore.endpoint.directory.EndpointStatus.STALE])
    is removed.

    Raises:
        EndpointNotFoundError: If the endpoint directory does not exist.
        EndpointRunningError: If the endpoint is running on this host or
            may be running on another host.
    """
    status, info = self._status()
    if status == EndpointStatus.RUNNING:
        raise EndpointRunningError(
            f'Endpoint {self.name} is already running.',
        )
    if status == EndpointStatus.OTHER_HOST:
        assert info is not None
        raise EndpointRunningError(
            f'Endpoint {self.name} appears to be running on '
            f'{info.hostname} (PID {info.pid}). Stop the endpoint on '
            f'{info.hostname}. If it is not running, delete the '
            f'connection file at {self.connection_path} and try again.',
        )
    if status == EndpointStatus.STALE:
        logger.debug(
            'Removing stale connection file (%s)',
            self.connection_path,
        )
        self.remove_connection(info)

remove

remove() -> None

Remove the endpoint directory and all of its files.

Raises:

• EndpointNotFoundError –

  If the endpoint directory does not exist.

• EndpointRunningError –

  If the endpoint is running or may be running on another host.

Source code in proxystore/endpoint/directory.py

def remove(self) -> None:
    """Remove the endpoint directory and all of its files.

    Raises:
        EndpointNotFoundError: If the endpoint directory does not exist.
        EndpointRunningError: If the endpoint is running or may be
            running on another host.
    """
    self.check_stopped()
    shutil.rmtree(self.path)

check_exists

check_exists() -> None

Check that the endpoint directory exists.

Raises:

• EndpointNotFoundError –

  If the endpoint directory does not exist.

Source code in proxystore/endpoint/directory.py

def check_exists(self) -> None:
    """Check that the endpoint directory exists.

    Raises:
        EndpointNotFoundError: If the endpoint directory does not exist.
    """
    if not os.path.isdir(self.path):
        raise EndpointNotFoundError(
            f'An endpoint named {self.name} does not exist in '
            f'{os.path.dirname(os.path.normpath(self.path))}.',
        )

restrict_permissions

restrict_permissions() -> bool

Remove all group and other permissions from the directory.

Clients trust the connection file in the endpoint directory, so no one other than the owner may be able to create, replace, or rename files in it. The directory also contains the endpoint's secret key, database, and log which may contain user data. Group and other permissions are also removed from the secret key file.

Returns:

• bool –

  True if the permissions of the directory or secret key file were changed.

Source code in proxystore/endpoint/directory.py

def restrict_permissions(self) -> bool:
    """Remove all group and other permissions from the directory.

    Clients trust the connection file in the endpoint directory, so no
    one other than the owner may be able to create, replace, or rename
    files in it. The directory also contains the endpoint's secret key,
    database, and log which may contain user data. Group and other
    permissions are also removed from the secret key file.

    Returns:
        `True` if the permissions of the directory or secret key file \
        were changed.
    """
    changed = False
    for path in (self.path, self.secret_key_path):
        try:
            mode = stat.S_IMODE(os.stat(path).st_mode)
        except FileNotFoundError:
            continue
        if mode & 0o077 != 0:
            os.chmod(path, mode & ~0o077)
            changed = True
    return changed

is_own_process

is_own_process(pid: int) -> bool

Check if a process with the PID exists and is owned by this user.

Source code in proxystore/endpoint/directory.py

def is_own_process(pid: int) -> bool:
    """Check if a process with the PID exists and is owned by this user."""
    if pid <= 0:
        return False
    try:
        os.kill(pid, 0)
    except (ProcessLookupError, PermissionError):
        # PermissionError means the PID belongs to another user. The endpoint
        # always runs as the current user, so the endpoint exited and its PID
        # was reused by the OS.
        return False
    return True