trinity.buffer package

Subpackages

• trinity.buffer.reader package
  • Submodules
    • trinity.buffer.reader.file_reader module
      • DummyProgressBar
      • BaseFileReader
      • SFTDataReader
      • DPODataReader
      • RolloutDataReader
      • RawDataReader
    • trinity.buffer.reader.queue_reader module
      • QueueReader
    • trinity.buffer.reader.sql_reader module
      • SQLReader
  • Module contents
• trinity.buffer.schema package
  • Submodules
    • trinity.buffer.schema.sql_schema module
      • TaskModel
      • ExperienceModel
      • SFTDataModel
      • DPODataModel
      • create_dynamic_table()
  • Module contents
    • create_dynamic_table()
    • Base
      • Base.__init__()
      • Base.metadata
      • Base.registry
• trinity.buffer.writer package
  • Submodules
    • trinity.buffer.writer.queue_writer module
      • QueueWriter
    • trinity.buffer.writer.sql_writer module
      • SQLWriter
  • Module contents

Submodules

• trinity.buffer.buffer module
  • get_buffer_reader()
  • get_buffer_writer()
• trinity.buffer.buffer_reader module
  • BufferReader
    • BufferReader.read()
    • BufferReader.read_async()
• trinity.buffer.buffer_writer module
  • BufferWriter
    • BufferWriter.write()
    • BufferWriter.write_async()
    • BufferWriter.acquire()
    • BufferWriter.release()
• trinity.buffer.queue module
  • linear_decay_priority()
  • QueueBuffer
    • QueueBuffer.put()
    • QueueBuffer.get()
    • QueueBuffer.qsize()
    • QueueBuffer.close()
    • QueueBuffer.stopped()
    • QueueBuffer.get_queue()
  • AsyncQueue
    • AsyncQueue.__init__()
    • AsyncQueue.close()
    • AsyncQueue.stopped()
  • AsyncPriorityQueue
    • AsyncPriorityQueue.capacity
    • AsyncPriorityQueue.priority_groups
    • AsyncPriorityQueue.priority_fn
    • AsyncPriorityQueue.reuse_cooldown_time
    • AsyncPriorityQueue.__init__()
    • AsyncPriorityQueue.put()
    • AsyncPriorityQueue.get()
    • AsyncPriorityQueue.qsize()
    • AsyncPriorityQueue.close()
    • AsyncPriorityQueue.stopped()
• trinity.buffer.utils module
  • retry_session()
  • default_storage_path()

Module contents

class trinity.buffer.BufferReader

Bases: ABC

Interface of the buffer reader.

abstract read(batch_size: int | None = None, strategy: ReadStrategy | None = None) → List

Read from buffer.

abstract async read_async(batch_size: int | None = None, strategy: ReadStrategy | None = None) → List

Read from buffer asynchronously.

class trinity.buffer.BufferWriter

Bases: ABC

Interface of the buffer writer.

abstract write(data: List) → None

Write to buffer.

abstract async write_async(data: List) → None

Write to buffer asynchronously.

abstract async acquire() → int

Acquire the buffer writer.

Returns:

The reference count of the buffer after acquiring.

Return type:

int

abstract async release() → int

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.get_buffer_reader(storage_config: StorageConfig, buffer_config: BufferConfig) → BufferReader

Get a buffer reader for the given dataset name.

trinity.buffer.get_buffer_writer(storage_config: StorageConfig, buffer_config: BufferConfig) → BufferWriter

Get a buffer writer for the given dataset name.