Does not work with enumerators

[cjbottaro]

Hello,

This code doesn't work:

enumerator = File.open("/usr/share/dict/words").each_line

producer = lambda do
  begin
    enumerator.next
  rescue StopIteration
    Parallel::Stop
  end
end

Parallel.each producer, in_processes: 4 do |i|
  print "#{Process.pid} #{i}\n"
end

It results in fiber called across threads (FiberError).

I poked around in the code; why are there 4 threads "feeding" the processes? Why not just have the main thread pluck items from the lambda to give to an available worker?

Thanks!

[grosser]

keeping the lambda on the main thread sounds like a good idea ...
internally it does a @mutex.synchronize around the lambda call ... but not
sure how to make it stay on the main thread ...

On Fri, May 6, 2016 at 2:02 PM, Christopher J. Bottaro <
notifications@github.com> wrote:

Hello,

This code doesn't work:

enumerator = File.open("/usr/share/dict/words").each_line

producer = lambda do
begin
enumerator.next
rescue StopIteration
Parallel::Stop
end
end

Parallel.each producer, in_processes: 4 do |i|
print "#{Process.pid} #{i}\n"
end

It results in fiber called across threads (FiberError).

I poked around in the code; why are there 4 threads "feeding" the
processes? Why not just have the main thread pluck items from the lambda to
give to an available worker?

Thanks!

—
You are receiving this because you are subscribed to this thread.
Reply to this email directly or view it on GitHub
#174

[denisdefreyne]

Not sure if it helps, but you don’t need the lambda, and this gem does work with enumerators. For example:

enum = File.open("/usr/share/dict/words").each_line
Parallel.map(enum, &:upcase)

[edsu]

But won't that load the entire file into memory because of this?

[grosser]

File.open("/usr/share/dict/words").each_line.respond_to?(:call)
=> false

so it will call .to_a
the original example look about right though

[grosser]

first we need a solution to this:

enumerator = File.open("/usr/share/dict/words").each_line
3.times { Thread.new { enumerator.next } }
(irb):2:in `next': fiber called across threads (FiberError)

it works with blocks, so it's not completely broken:

foo = -> { 1 }
3.times { Thread.new { 100000.times { foo.call } } }

so we need to find something that works in multiple threads ...

we could also rebuild the parallel process execution to use not use threads, by using IO.select or similar, but I expect that to be a lot of work too