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). Estimated decode sizes determine group membership and per-file ceilings. Unknown sizes receive an equal share initially. A ceiling failure first retries within the remaining configured PCM budget while successful group entries stay retained. If that is insufficient, later retained audio is released and only those entries are prepared again after the full-ceiling retry. A known file that cannot fit the group cap starts with that exclusive path. 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, estimate_bytes: nil, exclusive_retry: nil, retained_bytes: nil, &prepare) ⇒ Pipeline

Returns a new instance of Pipeline.

Raises:

  • (ArgumentError) 


166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 166

def initialize(items, memory_byte_limit:, requested_workers:, enabled:, worker_limit: nil,
               estimate_bytes: nil, exclusive_retry: nil, retained_bytes: 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
  @estimate_bytes = estimate_bytes
  @exclusive_retry = exclusive_retry
  @retained_bytes = retained_bytes
  @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.



164
165
166
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 164

def effective_workers
  @effective_workers
end

#group_byte_limit ⇒ Object (readonly)

Returns the value of attribute group_byte_limit.



164
165
166
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 164

def group_byte_limit
  @group_byte_limit
end

#memory_byte_limit ⇒ Object (readonly)

Returns the value of attribute memory_byte_limit.



164
165
166
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 164

def memory_byte_limit
  @memory_byte_limit
end

#wait_seconds ⇒ Object (readonly)

Returns the value of attribute wait_seconds.



164
165
166
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 164

def wait_seconds
  @wait_seconds
end

Instance Method Details

#each(&block) ⇒ Object 



195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 195

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 = build_groups
  pool = WorkerPool.new(@effective_workers, &@prepare)
  group = groups.next
  group_index = 0
  pending = submit(group, group_index, pool)
  loop do
    prepared = resolve(pending)
    pending = nil
    next_group = next_group(groups)
    if retry_exclusively?(group, prepared)
      consume_with_exclusive_retry(group, prepared, pool, &block)
    else
      pending = submit(next_group, group_index + 1, pool) if next_group && !group.exclusive && !next_group.exclusive
      consume(prepared, &block)
    end
    break unless next_group

    pending ||= submit(next_group, group_index + 1, pool)
    group = next_group
    group_index += 1
  end
  completed = true
ensure
  cancel(pending) if defined?(pending) && pending
  pool&.close(cancel: !completed) if defined?(pool) && pool
end

#pipelined? ⇒ Boolean

Returns:

  • (Boolean) 


191
192
193
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 191

def pipelined?
  @enabled
end