Class: Cohere::Transcribe::Runtime::Preparation::Pipeline

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/cohere/transcribe/runtime/preparation.rb

Overview

Bounded ordered preparation with at most one group ahead of ASR.

A pipelined group receives at most half of the configured decoded-PCM budget (and never more than 512 MiB). Jobs inside it divide that cap, so the current and one in-flight next group cannot together exceed the configured PCM budget. Native decoder implementation transients are outside this retained-PCM accounting, as they are in the Python path.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(items, memory_byte_limit:, requested_workers:, enabled:, worker_limit: nil, &prepare) ⇒ Pipeline

Returns a new instance of Pipeline.

Raises:

  • (ArgumentError) 


91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 91

def initialize(items, memory_byte_limit:, requested_workers:, enabled:, worker_limit: nil, &prepare)
  raise ArgumentError, "prepare block is required" unless prepare

  @items = items.to_a.freeze
  @memory_byte_limit = Integer(memory_byte_limit)
  raise ArgumentError, "memory_byte_limit must be positive" unless @memory_byte_limit.positive?

  @prepare = prepare
  @wait_seconds = 0.0
  @enabled = enabled && @items.length > 1
  @worker_limit = worker_limit.nil? ? nil : Integer(worker_limit)
  raise ArgumentError, "worker_limit must be positive" if @worker_limit && !@worker_limit.positive?

  @group_byte_limit = if @enabled
                        [[@memory_byte_limit / 2, 1].max, MAX_PIPELINE_GROUP_BYTES].min
                      else
                        @memory_byte_limit
                      end
  @effective_workers = resolve_workers(requested_workers)
end

Instance Attribute Details

#effective_workers ⇒ Object (readonly)

Returns the value of attribute effective_workers.



89
90
91
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 89

def effective_workers
  @effective_workers
end

#group_byte_limit ⇒ Object (readonly)

Returns the value of attribute group_byte_limit.



89
90
91
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 89

def group_byte_limit
  @group_byte_limit
end

#memory_byte_limit ⇒ Object (readonly)

Returns the value of attribute memory_byte_limit.



89
90
91
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 89

def memory_byte_limit
  @memory_byte_limit
end

#wait_seconds ⇒ Object (readonly)

Returns the value of attribute wait_seconds.



89
90
91
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 89

def wait_seconds
  @wait_seconds
end

Instance Method Details

#each(&block) ⇒ Object 



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 116

def each(&block)
  return enum_for(__method__) unless block
  return if @items.empty?

  unless pipelined?
    @items.each { |item| block.call(@prepare.call(item, @memory_byte_limit, 0)) }
    return
  end

  groups = @items.each_slice(@effective_workers).to_a
  pool = WorkerPool.new(@effective_workers, &@prepare)
  pending = submit(groups.first, 0, pool)
  groups.each_with_index do |_group, group_index|
    prepared = resolve(pending)
    pending = nil
    pending = submit(groups.fetch(group_index + 1), group_index + 1, pool) if group_index + 1 < groups.length
    consume(prepared, &block)
  end
  completed = true
ensure
  cancel(pending) if defined?(pending) && pending
  pool&.close(cancel: !completed) if defined?(pool) && pool
end

#pipelined? ⇒ Boolean

Returns:

  • (Boolean) 


112
113
114
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 112

def pipelined?
  @enabled
end