trinity.buffer package

Subpackages

• trinity.buffer.operators package
  • Subpackages
    • trinity.buffer.operators.filters package
      • Submodules
      • Module contents
    • trinity.buffer.operators.mappers package
      • Submodules
      • Module contents
  • Submodules
    • trinity.buffer.operators.data_juicer_operator module
      • DataJuicerOperator
    • trinity.buffer.operators.experience_operator module
      • ExperienceOperator
  • Module contents
    • ExperienceOperator
      • ExperienceOperator.process()
      • ExperienceOperator.create_operators()
      • ExperienceOperator.close()
    • RewardFilter
      • RewardFilter.__init__()
      • RewardFilter.process()
    • RewardSTDFilter
      • RewardSTDFilter.__init__()
      • RewardSTDFilter.process()
    • RewardShapingMapper
      • RewardShapingMapper.__init__()
      • RewardShapingMapper.process()
    • DataJuicerOperator
      • DataJuicerOperator.__init__()
      • DataJuicerOperator.process()
      • DataJuicerOperator.close()
• trinity.buffer.pipelines package
  • Submodules
    • trinity.buffer.pipelines.experience_pipeline module
      • get_input_buffers()
      • ExperiencePipeline
    • trinity.buffer.pipelines.task_pipeline module
      • check_and_run_task_pipeline()
      • TaskPipeline
  • Module contents
    • ExperiencePipeline
      • ExperiencePipeline.__init__()
      • ExperiencePipeline.prepare()
      • ExperiencePipeline.process()
      • ExperiencePipeline.close()
    • TaskPipeline
      • TaskPipeline.__init__()
      • TaskPipeline.convert_pipeline_config()
      • TaskPipeline.process()
      • TaskPipeline.close()
    • check_and_run_task_pipeline()
• 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.file_writer module
      • JSONWriter
    • 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.ray_wrapper module
  • DBWrapper
    • DBWrapper.__init__()
    • DBWrapper.get_wrapper()
    • DBWrapper.write()
    • DBWrapper.read()
    • DBWrapper.acquire()
    • DBWrapper.release()
  • FileWrapper
    • FileWrapper.__init__()
    • FileWrapper.get_wrapper()
    • FileWrapper.write()
    • FileWrapper.read()
    • FileWrapper.acquire()
    • FileWrapper.release()
  • is_database_url()
  • is_json_file()
  • QueueWrapper
    • QueueWrapper.__init__()
    • QueueWrapper.acquire()
    • QueueWrapper.release()
    • QueueWrapper.length()
    • QueueWrapper.put_batch()
    • QueueWrapper.get_batch()
    • QueueWrapper.get_wrapper()
• 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.