Class: CZTop::Z85::Pipe::Strategy::Parallel

Inherits:
CZTop::Z85::Pipe::Strategy show all
Defined in:
lib/cztop/z85/pipe.rb

Overview

Uses three threads:

  1. reads from source

  2. encodes/decodes

  3. writes to sink

This might give a performance increase on truly parallel platforms such as Rubinius and JRuby (and multiple CPU cores).

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Parallel

Initializes the 2 sized queues used.



112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/cztop/z85/pipe.rb', line 112

def initialize(*)
  super
  # @source
  # |
  # V
  @source_queue = SizedQueue.new(20) # limit memory usage
  # |
  # V
  # xcode
  # |
  # V
  @sink_queue   = SizedQueue.new(20) # limit memory usage
  # |
  # V
  # @sink
end

Instance Method Details

#execute ⇒ Object

Runs the algorithm.

Raises:

  • (void) 


131
132
133
134
135
 
# File 'lib/cztop/z85/pipe.rb', line 131

def execute
  Thread.new { read }
  Thread.new { xcode }
  write
end

#read ⇒ void (private)

This method returns an undefined value.

Reads all chunks and pushes them into the source queue. Then pushes a nil into the queue.



142
143
144
145
146
147
 
# File 'lib/cztop/z85/pipe.rb', line 142

def read
  while chunk = @source.read(@read_sz)
    @source_queue << chunk
  end
  @source_queue << nil
end

#write ⇒ void (private)

This method returns an undefined value.

Pops all chunks from the sink queue and writes them to the sink.



170
171
172
173
174
 
# File 'lib/cztop/z85/pipe.rb', line 170

def write
  while chunk = @sink_queue.pop
    @sink << chunk
  end
end

#xcode ⇒ void (private)

This method returns an undefined value.

Pops all chunks from the source queue, encodes or decodes them, and pushes the result into the sink queue. Then pushes a nil into the queue.



153
154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/cztop/z85/pipe.rb', line 153

def xcode
  # Encode all but the last chunk with pure Z85.
  previous_chunk = nil
  while true
    chunk = @source_queue.pop

    # call @xcode for the trailing nil-chunk as well
    @sink_queue << @xcode.(chunk, previous_chunk)

    break if chunk.nil?
    previous_chunk = chunk
  end
  @sink_queue << nil
end