What's it all about, Alfie?

Found Problem in JRuby 1.7.0 and Java Executors

JRuby

As part of my working in scaling up the code to a global system, I ran into some very odd problems in jruby-1.7.0. I'm not exactly sure if they are in jruby or in the JDK 1.7.0_05 implementation for CentOS 5, but I'm willing to bet they are in jruby, as I think the JDK has been hammered on a lot on these issues. So here's what I found.

It's all about using Java Executors in jruby-1.7.0. I'm creating a series of threads with the newFixedThreadPool() method, and then running through an array of things to send out to a REST API (Salesforce), and then waiting for it all to finish up.

  require 'java'
  java_include 'java.util.concurrent.Executors'
 
  executors = Executors.new_fixed_thread_pool(5)
  updates.each do |u|
    send_update(u)
  end

What I'm seeing is that when the threads are done processing, they don't all seem to "clear" the executor. For some reason, they aren't seen as "done" by the executor, but when looking at the REST API service, I know they completed. And it's always in the last batch of tasks.

This doesn't ever seem to happen on the Amazon EC2 hardware - only the nice, new, fast boxes in the datacenter.

So what I decided to do was to add a special timeout to the shutdown of the executor (start at line 36). This says that if we know how long any action should take, then if we get to the end of the processing queue in the executor, and we have waited long enough, then it's OK to forcibly shut down the executors and know that ready-of-not, it should have been done.

It's not ideal, and in most good cases, it shouldn't happen. But I'm getting a lot of problems with Salesforce and CouchDB as a part of this scaling, and I really have no idea what's going on inside either of those systems. Better to add this and be safe.

require 'java' java_import 'java.util.concurrent.Executors' java_import 'java.util.concurrent.TimeUnit' module Enumerable # From activesupport/lib/active_support/core_ext/enumerable.rb def sum(identity = 0, &block) if block_given? map(&block).sum(identity) else inject(:+) || identity end end def avg(&block) sum(&block).to_f / length end def parallel_each(options = {}) thread_count = options[:parallelism] || 4 if thread_count > 1 executor = Executors.new_fixed_thread_pool(thread_count) self.each do |obj| executor.execute do begin yield obj rescue Exception => e if error_handler = options[:on_error] error_handler.call(e) if error_handler.respond_to?(:call) end end end end executor.shutdown if (final_timeout = options[:shutdown_timeout].to_i) > 0 total_time = 0 begin return if executor.await_termination(5, TimeUnit::SECONDS) if executor.get_queue.size > 0 total_time = 0 else if (total_time += 5) > final_timeout if error_handler = options[:on_error] error_handler.call(Timeout::Error.new("[Enumerable::parallel_each] final #{executor.get_active_count} tasks could not complete in the provided #{final_timeout} sec timeout.")) if error_handler.respond_to?(:call) end # force a shutdown and return to the caller executor.shutdown_now return end end rescue Exception => e if error_handler = options[:on_error] error_handler.call(e) if error_handler.respond_to?(:call) end end until executor.is_terminated else sleep(0.01) until executor.is_terminated end else # don't use executors if we don't need to self.each do |obj| begin yield obj rescue Exception => e if error_handler = options[:on_error] error_handler.call(e) if error_handler.respond_to?(:call) end end end end end def frequencies self.reduce(Hash.new(0)) { |h, v| h[v] += 1; h } end end

This entry was posted on Wednesday, November 7th, 2012 at 3:10 pm and is filed under Coding, Cube Life. You can follow any responses to this entry through the RSS 2.0 feed. Both comments and pings are currently closed.

Found Problem in JRuby 1.7.0 and Java Executors

Pages

Archives

Categories