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. A file that cannot fit that cap is prepared alone with the full configured ceiling and without overlap. 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, &prepare) ⇒ Pipeline

Returns a new instance of Pipeline.

Raises:

  • (ArgumentError) 


96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 96

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



94
95
96
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 94

def effective_workers
  @effective_workers
end

#group_byte_limit ⇒ Object (readonly)

Returns the value of attribute group_byte_limit.



94
95
96
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 94

def group_byte_limit
  @group_byte_limit
end

#memory_byte_limit ⇒ Object (readonly)

Returns the value of attribute memory_byte_limit.



94
95
96
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 94

def memory_byte_limit
  @memory_byte_limit
end

#wait_seconds ⇒ Object (readonly)

Returns the value of attribute wait_seconds.



94
95
96
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 94

def wait_seconds
  @wait_seconds
end

Instance Method Details

#each(&block) ⇒ Object 



123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 123

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)
  pending = submit(groups.first, 0, pool)
  groups.each_with_index do |group, group_index|
    prepared = resolve(pending)
    pending = nil
    next_group = groups[group_index + 1]
    pending = submit(next_group, group_index + 1, pool) if next_group && !group.exclusive && !next_group.exclusive
    consume(prepared, &block)
    pending ||= submit(next_group, group_index + 1, pool) if next_group
  end
  completed = true
ensure
  cancel(pending) if defined?(pending) && pending
  pool&.close(cancel: !completed) if defined?(pool) && pool
end

#pipelined? ⇒ Boolean

Returns:

  • (Boolean) 


119
120
121
 
# File 'lib/cohere/transcribe/runtime/preparation.rb', line 119

def pipelined?
  @enabled
end