proxystore.endpoint.directory

proxystore/endpoint/directory.py

Endpoint directory layout and files.

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

endpoint_dir = EndpointDir.from_home('/path/to/proxystore', 'my-ep')
assert endpoint_dir.path == '/path/to/proxystore/my-ep'
config = endpoint_dir.read_config()

Attributes:

• path (str) –

  Path of the directory.

config_path property

config_path: str

Path to the endpoint configuration.

database_path property

database_path: str

Path to the default SQLite database for persisting objects.

log_path property

log_path: str

Path to the log of the endpoint daemon.

pid_path property

pid_path: str

Path to the PID file of the endpoint daemon.

connection_path property

connection_path: str

Path to the connection file clients use to connect.

from_home classmethod

from_home(proxystore_dir: str, name: str) -> Self

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

Parameters:

• proxystore_dir (str) –

  ProxyStore home directory (see home_dir()).

• name (str) –

  Name of the endpoint.

Source code in proxystore/endpoint/directory.py

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

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

find_all classmethod

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

Find all endpoints with a valid configuration.

Parameters:

• proxystore_dir (str) –

  ProxyStore home directory to search in (see 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
) -> list[tuple[Self, EndpointConfig]]:
    """Find all endpoints with a valid configuration.

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

    Returns:
        List of each endpoint directory and its configuration.
    """
    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_home()).
    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:

• FileNotFoundError –

  If the configuration file does not exist.

• ValueError –

  If the configuration contains an invalid value or cannot be parsed.

Source code in proxystore/endpoint/directory.py

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

    Raises:
        FileNotFoundError: If the configuration file does not exist.
        ValueError: If the configuration contains an invalid value or
            cannot be parsed.
    """
    try:
        with open(self.config_path, 'rb') as f:
            return load(EndpointConfig, f)
    except FileNotFoundError:
        raise FileNotFoundError(
            f'Endpoint directory {self.path} does not contain a valid '
            'configuration.',
        ) from None
    except ValueError as e:
        # Includes TOML decoding and pydantic validation errors.
        raise ValueError(
            f'Unable to parse ({self.config_path}): {e!s}.',
        ) from None

write_config

write_config(config: EndpointConfig) -> None

Write the endpoint configuration, creating the directory if needed.

Parameters:

• config (EndpointConfig) –

  Configuration to write.

Source code in proxystore/endpoint/directory.py

def write_config(self, config: EndpointConfig) -> None:
    """Write the endpoint configuration, creating the directory if needed.

    Args:
        config: Configuration to write.
    """
    # Clients trust the connection file in the endpoint directory, so
    # only the owner can create or replace files in it.
    os.makedirs(self.path, mode=0o700, exist_ok=True)
    with open(self.config_path, 'wb') as f:
        dump(config, f)

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.
    """
    data = {
        'host': info.host,
        'port': info.port,
        'token': info.token.hex(),
        'tls_fingerprint': info.tls_fingerprint,
    }
    write_private_file(self.connection_path, json.dumps(data).encode())

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).

• ValueError –

  If the connection file is malformed.

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).
        ValueError: If the connection file is malformed.
    """
    with open(self.connection_path, 'rb') as f:
        contents = f.read()
    try:
        data = json.loads(contents)
        info = ConnectionInfo(
            host=data['host'],
            port=data['port'],
            token=bytes.fromhex(data['token']),
            tls_fingerprint=data['tls_fingerprint'],
        )
    except (TypeError, KeyError, ValueError):
        info = None
    if (
        info is None
        or not isinstance(info.host, str)
        or not isinstance(info.port, int)
        or len(info.token) != TOKEN_SIZE
        or not isinstance(info.tls_fingerprint, (str, type(None)))
    ):
        raise ValueError(
            f'Connection file at {self.connection_path} is malformed.',
        )
    return info

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)

running_pid

running_pid() -> int | None

Get the PID of the endpoint daemon if it is running.

Returns:

• int | None –

  The PID in the PID file if that process is running as the current user on this host, otherwise None (e.g., the PID file is missing or malformed, the endpoint stopped unexpectedly, or the endpoint is running on a different host).

Source code in proxystore/endpoint/directory.py

def running_pid(self) -> int | None:
    """Get the PID of the endpoint daemon if it is running.

    Returns:
        The PID in the PID file if that process is running as the \
        current user on this host, otherwise `None` (e.g., the PID file \
        is missing or malformed, the endpoint stopped unexpectedly, or \
        the endpoint is running on a different host).
    """
    try:
        with open(self.pid_path) as f:
            pid = int(f.read().strip())
    except (OSError, ValueError):
        return None
    return pid if is_own_process(pid) else None

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 database and log which may contain user data.

Returns:

• bool –

  True if the permissions of the directory 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 database and
    log which may contain user data.

    Returns:
        `True` if the permissions of the directory were changed.
    """
    mode = stat.S_IMODE(os.stat(self.path).st_mode)
    if mode & 0o077 == 0:
        return False
    os.chmod(self.path, mode & ~0o077)
    return True

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