Java – Sharing buffer between multiple threads

Using a single Timer and an unbounded ExecutorService (thread pooling) for this type of general-purpose scheduling can be very powerful. However, it does not stop concurrent execution of the task, should the previous invocation not yet be completed.

We've combined Timer and ExecutorService just for this purpose, along with the ability to restrict the number of concurrent tasks based on the number of CPUs available to the system (appropriate for CPU-bound tasks):

http://www.aoindustries.com/docs/ao-concurrent/com/aoindustries/util/concurrent/ExecutorService.html

To stop concurrent invocation with only these two tools you'd have to roll your own protection.

I solved this problem recently by creating a "ConcurrencyLimiter" that will serialize access to an arbitrary piece of code/resource given an arbitrary key object. If a second thread tries perform the same task, it'll simply wait and use the result from the first thread. Thus, concurrency is avoided and both threads get a meaningful response.

http://www.aoindustries.com/docs/ao-concurrent/com/aoindustries/util/concurrent/ConcurrencyLimiter.html

Regarding limiting concurrency within the task scheduler itself, there is also a simple pure-Java implementation of cron that allows to avoid concurrent execution and handles all of the scheduling:

http://www.aoindustries.com/docs/ao-cron/com/aoindustries/cron/CronDaemon.html

Source at http://www.aoindustries.com/src/ao-concurrent.src.jar (depends on http://www.aoindustries.com/src/aocode-public.src.jar) and http://www.aoindustries.com/src/ao-cron.src.jar All LGPLv3 and free for the world, no external dependencies, and Java 1.6+.

The point here isn't to plug my own code. I just want to point-out that you can either handle it at the scheduling level or within the task itself.

Sorry - vague questions get vague answers.

The answer by scriptin is giving a reasonable solution, but doesn't answer your questions about the failure of the implementation you had.

• You get a 'queue empty' error when you call pop on a Ruby queue in non-blocking mode (true passed) and there are no items on the queue.
• Your "producer" thread is created and your "consumer" thread is created, but they'll run independently and start executing whenever Ruby feels like it.
• If your "consumer" runs before the "producer" has pushed anything onto the queue, then obviously the "consumer" will pop an empty queue and throw an exception.

The solution to this one is easy; the consumer shouldn't use non-blocking mode. Don't pass true, just call pop(). The whole point of the consumer thread is that it can block on a queue waiting for the producer, without the rest of the application halting.

As for calling lastfm.track.scrobble - yes, this is executing that method immediately and pushing the result on the queue, which is exactly not what you want :-D and is in fact the reason why "reliably" your consumer thread gets to pop its queue before the producer ever has a chance to push anything (even if the producer thread runs first, it hits a slow 'scrobble' call and sits waiting for HTTP requests/responses to complete; so the consumer thread executes and tries to read from the empty queue).

You want to push the parameters to scrobble onto the queue, then have the consumer thread do the scrobbling. It's "nicer"/more structured to make little classes encapsulating the work and push instances of those onto your queue, but a cheap-and-nasty solution just uses the Hash which scrobble takes as input. We also need to remember to loop over the queue entries inside the consumer, instead of only processing one! Thus (UNTESTED) we might have:

queue = SizedQueue.new(10)

Thread.new do 
  songs.each do |song|
      queue << { artist: song[:artist], track: song[:track] }
  end
  queue << "done"
end

consumer = Thread.new do
  loop do
    scrobble_args = queue.pop()
    Thread.exit if scrobble_args == "done"
    lastfm.track.scrobble(scrobble_args)
    puts "Scrobbled #{scrobble_args[:track] by scrobble_args[:artist]}."
  end
end

consumer.join

This should technically work but is totally pointless, because that one lonely consumer thread is just processing the queue in serial. So let's simplify this. Really, you just want to dump all of 'songs' into the queue and have the consumer thread do the work. Since we want to run in parallel, we'll make an array of threads.

# I'm not bothering with the producer thread here so the queue size is unlimited.
# You could just as well put the 'songs.each' inside a producer thread with a
# a sized queue, as per your original code, if you wanted.
#
queue = Queue.new
songs.each do | song |
  queue << song
end

consumers = []

1.upto( 10 ) do
  consumers << Thread.new do

    # Keep looping until asked to exit
    #
    loop do
      song = queue.pop()
      Thread.exit if song == 'done'
      lastfm.track.scrobble( artist: song[ :artist ], track: song[ :track ] )
      puts "Scrobbled #{ song[ :track ] } by #{ song[ :artist ] }."
    end
  end

  # Push one 'done' exit instruction for each consumer
  #
  queue << 'done'
end

# Wait for all consumer threads to complete
#
consumers.each do | consumer |
  consumer.join()
end

...and that should do the trick. The "Parallel" gem will probably be doing much the same sort of thing under the hood.