proxystore.store.factory¶

proxystore/store/factory.py

Factory implementations.

StoreFactory ¶

StoreFactory(
    key: ConnectorKeyT,
    store_config: StoreConfig,
    *,
    evict: bool = False,
)

Bases: Generic[ConnectorT, T]

Factory that resolves an object from a store.

Adds support for asynchronously retrieving objects from a Store instance.

The factory takes the store_config parameter that is used to reinitialize the store if the factory is sent to a remote process where the store has not already been initialized.

Calls to the factory from threads while another call (or resolve_async()) is resolving the object wait for that result, so the object is resolved once and every thread gets the same object.

Parameters:

key (ConnectorKeyT) –

Key corresponding to object in store.
store_config (StoreConfig) –

Store configuration used to reinitialize the store if needed.
evict (bool, default: False ) –

If True, evict the object from the store once resolve() is called.

Source code in proxystore/store/factory.py

def __init__(
    self,
    key: ConnectorKeyT,
    store_config: StoreConfig,
    *,
    evict: bool = False,
) -> None:
    # The key is typed Any because the type of the store's connector,
    # and so the type of its keys, is not known from a proxy.
    self.key: Any = key
    self.store_config = store_config
    self.evict = evict

    # Not included when a factory is serialized because it is specific
    # to this instance of the factory.
    self._resolve: _Resolve | None = None

get_store ¶

get_store() -> Store[ConnectorT]

Get store and reinitialize if necessary.

Raises:

StoreClosedError –

If the owner of the store closed it and removed its objects in this process.

Source code in proxystore/store/factory.py

def get_store(self) -> Store[ConnectorT]:
    """Get store and reinitialize if necessary.

    Raises:
        StoreClosedError: If the owner of the store closed it and
            removed its objects in this process.
    """
    return registry.get_or_create(self.store_config)

resolve ¶

resolve() -> T

Get object associated with key from store.

Raises:

ProxyResolveMissingKeyError –

If the key associated with this factory does not exist in the store.
StoreClosedError –

If the owner of the store closed it and removed its objects in this process.

Source code in proxystore/store/factory.py

def resolve(self) -> T:
    """Get object associated with key from store.

    Raises:
        ProxyResolveMissingKeyError: If the key associated with this
            factory does not exist in the store.
        StoreClosedError: If the owner of the store closed it and
            removed its objects in this process.
    """
    with Timer() as timer:
        store = self.get_store()
        obj = store.get(self.key, default=_MISSING_OBJECT)

        if obj is _MISSING_OBJECT:
            raise ProxyResolveMissingKeyError(
                self.key,
                type(store),
                store.name,
                store.id,
            )

        if self.evict:
            store.evict(self.key)

    if store.metrics is not None:
        total_time = timer.elapsed_ms
        store.metrics.add_time('factory.resolve', self.key, total_time)

    return cast(T, obj)

resolve_async ¶

resolve_async() -> None

Asynchronously get object associated with key from store.

Nothing happens if the object is already being resolved.

Source code in proxystore/store/factory.py

def resolve_async(self) -> None:
    """Asynchronously get object associated with key from store.

    Nothing happens if the object is already being resolved.
    """
    with _resolve_lock:
        current = self._resolve
        if current is not None and current.generation == _fork_generation:
            return
        logger.debug('Starting asynchronous resolve of %s', self.key)
        resolve = _Resolve(thread=None)
        resolve.future = _get_default_pool().submit(
            self._resolve_in_pool,
            resolve,
        )
        self._resolve = resolve

PollingStoreFactory ¶

PollingStoreFactory(
    key: ConnectorKeyT,
    store_config: StoreConfig,
    *,
    evict: bool = False,
    polling: PollingPolicy | None = None,
)

Bases: StoreFactory[ConnectorT, T]

Factory that polls a store until an object can be resolved.

This is an extension of the StoreFactory with the resolve() method overridden to poll the store until the target object is available. If the object is an exception, it is raised instead of returned.

Parameters:

key (ConnectorKeyT) –

Key corresponding to object in store.
store_config (StoreConfig) –

Store configuration used to reinitialize the store if needed.
evict (bool, default: False ) –

If True, evict the object from the store once resolve() is called.
polling (PollingPolicy | None, default: None ) –

Policy for polling the store for the object. If None, the default PollingPolicy is used.

Source code in proxystore/store/factory.py

def __init__(
    self,
    key: ConnectorKeyT,
    store_config: StoreConfig,
    *,
    evict: bool = False,
    polling: PollingPolicy | None = None,
) -> None:
    super().__init__(key, store_config, evict=evict)
    self.polling = polling if polling is not None else PollingPolicy()

resolve ¶

resolve() -> T

Get object associated with key from store.

Raises:

ProxyResolveMissingKeyError –

If the object associated with the key is not available after the timeout of the polling policy.
Exception –

The exception set on the future associated with this factory.

Source code in proxystore/store/factory.py

def resolve(self) -> T:
    """Get object associated with key from store.

    Raises:
        ProxyResolveMissingKeyError: If the object associated with the
            key is not available after the timeout of the polling policy.
        Exception: The exception set on the future associated with this
            factory.
    """
    obj = self._poll(self.polling.timeout)
    if obj is None:
        store = self.get_store()
        raise ProxyResolveMissingKeyError(
            self.key,
            type(store),
            store.name,
            store.id,
        )
    return obj[0]

get_store ¶

get_store() -> Store[ConnectorT]

Get store and reinitialize if necessary.

Raises:

StoreClosedError –

If the owner of the store closed it and removed its objects in this process.

Source code in proxystore/store/factory.py

def get_store(self) -> Store[ConnectorT]:
    """Get store and reinitialize if necessary.

    Raises:
        StoreClosedError: If the owner of the store closed it and
            removed its objects in this process.
    """
    return registry.get_or_create(self.store_config)

resolve_async ¶

resolve_async() -> None

Asynchronously get object associated with key from store.

Nothing happens if the object is already being resolved.

Source code in proxystore/store/factory.py

def resolve_async(self) -> None:
    """Asynchronously get object associated with key from store.

    Nothing happens if the object is already being resolved.
    """
    with _resolve_lock:
        current = self._resolve
        if current is not None and current.generation == _fork_generation:
            return
        logger.debug('Starting asynchronous resolve of %s', self.key)
        resolve = _Resolve(thread=None)
        resolve.future = _get_default_pool().submit(
            self._resolve_in_pool,
            resolve,
        )
        self._resolve = resolve