diff --git a/lib/stathat.rb b/lib/stathat.rb index b165d46..4eebee5 100644 --- a/lib/stathat.rb +++ b/lib/stathat.rb @@ -96,7 +96,7 @@ class Reporter def initialize @que = Queue.new - @runlock = Mutex.new + @running = false run_pool() end @@ -105,6 +105,10 @@ def finish() # XXX serialize queue? end + def running? + @running + end + def post_value(stat_key, user_key, value, timestamp, cb) args = { :key => stat_key, :ukey => user_key, @@ -139,12 +143,11 @@ def ez_post_count(stat_name, ezkey, count, timestamp, cb) private def run_pool - @runlock.synchronize { @running = true } + @running = true @pool = [] 5.times do |i| @pool[i] = Thread.new do - while true do - point = @que.pop + while (point = @que.pop) != :quit # XXX check for error? begin resp = Common::send_to_stathat(point[:url], point[:args]) @@ -154,18 +157,21 @@ def run_pool rescue pp $! end - @runlock.synchronize { - break unless @running - } end end end end def stop_pool() - @runlock.synchronize { - @running = false - } + @running = false + + # Each thread will stop after it receives + # `:quit` instead of a hash of stat + # information. Sending `@pool.length` quit + # messages ensures that all threads will be + # woken up, and thus avoid deadlocks. + @pool.length.times { @que << :quit } + @pool.each do |th| th.join if th && th.alive? end diff --git a/test/test_stathat.rb b/test/test_stathat.rb index 76140d9..cb7a9cd 100644 --- a/test/test_stathat.rb +++ b/test/test_stathat.rb @@ -66,4 +66,18 @@ def test_classic_value_bad_keys_sync assert_equal(r.msg, "invalid keys", "incorrect error message") assert_equal(r.status, 500, "incorrect status code") end + + def test_stat_on_finished_queue + reporter = StatHat::Reporter.instance + reporter.finish + + assert !StatHat::API.ez_post_value("test ez count queued stat", "test@stathat.com", 12) + end + + def test_reporter_finish + reporter = StatHat::Reporter.instance + reporter.finish + + assert !reporter.running? + end end