trinity.buffer

Subpackages

Submodules

trinity.buffer.buffer module

The buffer module

trinity.buffer.buffer.get_buffer_reader(storage_config: StorageConfig, buffer_config: BufferConfig) → BufferReader[source]: Get a buffer reader for the given dataset name.

trinity.buffer.buffer.get_buffer_writer(storage_config: StorageConfig, buffer_config: BufferConfig) → BufferWriter[source]: Get a buffer writer for the given dataset name.

trinity.buffer.buffer_reader module

Reader of the buffer.

class trinity.buffer.buffer_reader.BufferReader[source]

Bases: ABC

Interface of the buffer reader.

abstract read(batch_size: int | None = None, strategy: ReadStrategy | None = None) → List[source]: Read from buffer.

trinity.buffer.buffer_writer module

Writer of the buffer.

class trinity.buffer.buffer_writer.BufferWriter[source]

Bases: ABC

Interface of the buffer writer.

abstract write(data: List) → None[source]: Write to buffer.

abstract acquire() → int[source]

Acquire the buffer writer.

Returns:: The reference count of the buffer after acquiring.
Return type:: int

abstract release() → int[source]

Release the buffer writer. After release, the buffer writer can not be used again.

Returns:: The reference count of the buffer after releasing.
Return type:: int

trinity.buffer.queue module

A queue implemented by Ray Actor.

trinity.buffer.queue.is_database_url(path: str) → bool[source]

trinity.buffer.queue.is_json_file(path: str) → bool[source]

class trinity.buffer.queue.QueueActor(storage_config: StorageConfig, config: BufferConfig)[source]

Bases: object

An asyncio.Queue based queue actor.

FINISH_MESSAGE = '$FINISH$'

__init__(storage_config: StorageConfig, config: BufferConfig) → None[source]

async acquire() → int[source]

async release() → int[source]: Release the queue.

length() → int[source]: The length of the queue.

async put_batch(exp_list: List) → None[source]: Put batch of experience.

async get_batch(batch_size: int) → List[source]: Get batch of experience.

classmethod get_actor(storage_config: StorageConfig, config: BufferConfig)[source]: Get the queue actor.

trinity.buffer.ray_wrapper module

class trinity.buffer.ray_wrapper.DBWrapper(storage_config: StorageConfig, config: BufferConfig)[source]

Bases: object

A wrapper of a SQL database.

If wrap_in_ray in StorageConfig is True, this class will be run as a Ray Actor, and provide a remote interface to the local database.

For databases that do not support multi-processing read/write (e.g. sqlite, duckdb), we recommend setting wrap_in_ray to True

__init__(storage_config: StorageConfig, config: BufferConfig) → None[source]

classmethod get_wrapper(storage_config: StorageConfig, config: BufferConfig)[source]

write(data: list) → None[source]

read(batch_size: int | None = None, strategy: ReadStrategy | None = None) → List[source]

acquire() → int[source]

release() → int[source]

class trinity.buffer.ray_wrapper.FileWrapper(storage_config: StorageConfig, config: BufferConfig)[source]

Bases: object

A wrapper of a local jsonl file.

If wrap_in_ray in StorageConfig is True, this class will be run as a Ray Actor, and provide a remote interface to the local file.

This wrapper is only for writing, if you want to read from the file, use StorageType.QUEUE instead.

__init__(storage_config: StorageConfig, config: BufferConfig) → None[source]

classmethod get_wrapper(storage_config: StorageConfig, config: BufferConfig)[source]

write(data: List) → None[source]

read() → List[source]

acquire() → int[source]

release() → int[source]

trinity.buffer.utils module

trinity.buffer.utils.retry_session(session_maker, max_retry_times: int, max_retry_interval: float)[source]: A Context manager for retrying session.

trinity.buffer.utils.default_storage_path(storage_config: StorageConfig, buffer_config: BufferConfig) → str[source]

Module contents

trinity.buffer.get_buffer_reader(storage_config: StorageConfig, buffer_config: BufferConfig) → BufferReader[source]: Get a buffer reader for the given dataset name.

trinity.buffer.get_buffer_writer(storage_config: StorageConfig, buffer_config: BufferConfig) → BufferWriter[source]: Get a buffer writer for the given dataset name.