rccursach rccursach - 2 months ago 10
Ruby Question

non-blocking ruby data processing or method call

I have a ruby script that reads data from a serial port.
That data might be some raw binary string that represents a specific protocol datagram (i'm trying XBee API as now).

this data has to be processed in a long run of method calls, something like:


  1. Read serial

  2. Parse binary datagram

  3. Parse payload

  4. Transform values (ie: dates from timestamps, linear regressions, etc)

  5. Transform to JSON

  6. Insert into database



Data income frequency is faster than my ability to process. i need to do something like this:

loop do
begin
res = @xbee.getresponse
return_super_fast_and_work_that_in_the_background res
rescue => e
puts e #append to some log here or something
end
end


So, what i can imagine is that i might need to collect a decent quantity of these datagrams and then process all of them in batch.

But i can't imagine how to implement such method:

#return_super_fast_and_work_in_the_background()


All the examples i can find are related to non-blocking IO or networking tasks and Eventmachine.

I have redis around, might be handy here, and can launch another scripts aside this one. (in fact i have a sinatra api connected to the database, and a pubsub/websocket waiting to be used in between to notify when new data is coming)

Any suggestions will be much appreciated!

Max Max
Answer

You can just spin off a thread each time you get more data.

def do_work res
  # parse, transform, insert, etc.
end

def read_loop
  loop do
    begin
      res = @xbee.getresponse
      Thread.new(res, &method(:do_work))
    rescue => e
      # ...
    end
  end
end

If your do_work method touches some common resource (a log, a database, stdout, etc.) you will need to protect that resource with a Mutex to prevent the different threads from stepping on each other. Also note that Ruby isn't truly multithreaded, so while this will achieve your goal of quickly returning to get more data, it will not actually offer a processing speedup.