emcien / iso_latte Goto Github PK

Hello,

i propose to add "detach" feature, which should be only an option to existing interface.

The rationale for this feature is:

• No blocking for main thread.
• Multiple forks.

Usage example with rabbitmq:

require 'iso_latte'
require 'bunny'

Message = Struct.new(:meta, :payload)

Job = Struct.new(:msg) do
  def perform
    $stderr.puts("Performing #{msg.payload.inspect}")
    sleep(rand(5)+1)
    $stderr.puts("Done!")
  end
end

conn = Bunny.new.start
channel = conn.create_channel

# consumer allowed to be sent up to 5 messages without acknowledgements 
channel.prefetch(5)

channel.queue('jobs').subscribe(manual_ack: true) do |_, meta, payload|
  msg = Message.new(meta, payload)

  IsoLatte.fork(detach: true,
                finish: proc { |success,rc| success ? msg.meta.ack : msg.meta.reject }) do
    conn.close
    Job.new(msg).perform
  end
end

In result i'll have worker supervisor, which able to handle upto 5 jobs simultaneously. Each job starts from scratch! no membloats! shitty jobs code protection!

Initially we just want normal auto-tests, but a build-matrix to test on other platforms would be nice too.

I am getting TypeError: no _dump_data is defined for class Binding from the Marshal.dump in the exception pipe code. This seems to be because e.cause produces another exception, which also has __better_errors_bindings_stack defined on it.

I guess I'll need to recurse? It's in the fork, so it shouldn't hurt anything.

This is very cool. I timeout option would be very useful too. That kinda covers the last condition of "my subprocess has been zombified and doesn't like it's ever going to return".

Rather than the current threading approach, which will require careful thought when any change is made to the timeout code, we can move to a more explicit approach:

Process.waitpid2 and Timeout.timeout can each run in a thread - rather than the current blocking wait, we'll use Process::WNOHANG and sleep-loop, checking a queue and then trying a wait every quarter-second. When the timeout goes off, it'll go off in a thread, and push a :timeout onto the queue. This approach prevents either thread from interrupting the other, and makes adding complexity to the timeout code (as in #3) much more reasonable.

Suggestion from moomaka here.

Right now the fork function accepts a 'timeout' option, and after that long has elapsed it will send a SIGKILL to the child.

Two approaches would be better.

1. We could use a SIGTERM -> SIGKILL chain to allow well-behaved programs to terminate cleanly (ruby processes would typically raise a signal error, which would propagate back out to the parent, but even ffi code can clean up from a SIGTERM). The main drawback is the amount of race-condition and timing code that would need to be written and tested.
2. We could allow the user to supply the signal type, so that the parent will send whatever signal is specified. This is pretty easy, and gets us most of the power, but without guaranteeing termination.

We could reasonably combine the two by allowing a set of timeouts and signals to be supplied, like { kill: 100, hup: 58 }, and manage all of the timeouts. Then we could interpret timeout: 10 to mean timeout: { kill: 11, term: 10 }.

Right now I'm not sure that it will work on other rubies - installing and running specs on various platforms would be useful. #4's matrix should probably be done afterward.

@nevinera How to load environment into fork to use ActiveRecord at here.
Ex:

IsoLatte.fork(
  stderr:   "tmp/suberr.txt",
  finish:   ->(success, rc) { puts "Finished. Success? #{success}" },
  success:  ->() { puts "Was successful" },
  kill:     ->() { puts "Received a SIGKILL" },
  fault:    ->() { puts "Received a SIGABRT, probably a segmentation fault" },
  exit:     ->(rc) { puts "subprocess exited explicitly with #{rc}" }
) do
  results = Product.search_by(categories, keywords, page).results
end

Right now we have a pipe open between the parent and child so that any ruby exceptions encountered can be marshaled back across it and re-raised. It would be equally useful to be able to conveniently return a result from the block.

Add an option to 'retrieve_result' (it'll default to false), which will cause a result-pipe to be opened, and the child process to marshal and send the result of the block back to the parent. The result will be available instead of the child's status object, and will be nil if the child does not complete successfully and without exiting.

https://github.com/emcien/iso_latte/blob/master/lib/iso_latte.rb#L87