bin/localhost spec/benchmarks/bm-20081111.txt spec/benchmarks/bm-20081113.txt @@ -1,29 +1,26 @@ -#!/usr/bin/env ruby +#!/usr/bin/env ruby -s lib_path = File.expand_path(File.join(File.dirname(__FILE__), "..", "lib")) $LOAD_PATH.unshift lib_path unless $LOAD_PATH.include?(lib_path) require 'gisting' -SERVERS = [ - ["127.0.0.1", 9081, Gisting::MapServer], - ["127.0.0.1", 9082, Gisting::MapServer], - ["127.0.0.1", 9083, Gisting::MapServer], - ["127.0.0.1", 9084, Gisting::MapServer], - ["127.0.0.1", 9085, Gisting::MapServer], +$m ||= false +$r ||= false - ["127.0.0.1", 9091, Gisting::ReduceServer], - ["127.0.0.1", 9092, Gisting::ReduceServer] -] - -def start_servers(servers) - EM::run { - servers.each do |server_conf| - host, port, gistable = server_conf - puts "Starting server #{gistable.to_s} on #{host}:#{port}" - EM::start_server host, port, gistable - end - } +def fail_unless(valid) + if !valid + puts "\nusage: #{__FILE__} -m[ap] or -r[educe] localhost:9081" + exit + end end + +fail_unless(($m ^ $r) && ARGV.size == 1) +host, port = ARGV.first.split(":") +gistable = Gisting::MapServer if $m +gistable = Gisting::ReduceServer if $r -start_servers(SERVERS) \ No newline at end of file +EM::run { + puts "Starting server #{gistable.to_s} on #{host}:#{port}" + EM::start_server host, port.to_i, gistable +} bin/server @@ -14,8 +14,15 @@ def args "/Volumes/gisting/datasets/aol-data/user-ct-test-collection-04.txt", "/Volumes/gisting/datasets/aol-data/user-ct-test-collection-05.txt", ] - # args += ["/Users/mchung/Public/datasets/sample1.data", "/Users/mchung/Public/datasets/sample2.data"] - # args += ["/Users/mchung/Public/datasets/aoldb_dev.txt"] + args = [ + "/Users/mchung/Public/datasets/sample1.data", + "/Users/mchung/Public/datasets/sample2.data" + ] + + args = [ + "/Volumes/gisting/datasets/aoldb_dev-aa", + "/Volumes/gisting/datasets/aoldb_dev-ab", + ] end if __FILE__ == $0 examples/term_count.rb @@ -65,7 +65,7 @@ module Gisting # This smells funny def stop! - # Stop all client connections and ends the reactor in Cluster + # Instructs the Cluster to stop all client connections and end the Map/Reduce reactor EM::stop_event_loop end lib/gisting/job.rb @@ -11,7 +11,7 @@ module Gisting @jobs_map_completed = 0 @jobs_reduce_started = 0 @jobs_reduce_completed = 0 - @next_map_job, @next_intermediate_job, @next_reduce_job = 0, 0, 0 # Counter + @next_map_job, @next_intermediate_job, @next_reduce_job = 0, 0, 0 # Counters end def init_map_job(input) @@ -23,12 +23,12 @@ module Gisting end def start_map_job(file_pattern) - @map_jobs[file_pattern][:started_at] = Time.now + @map_jobs[file_pattern][:map_started_at] = Time.now @jobs_map_started +=1 end def stop_map_job(file_pattern, intermediate_job) - @map_jobs[file_pattern][:ended_at] = Time.now + @map_jobs[file_pattern][:map_ended_at] = Time.now @map_jobs[file_pattern][:intermediate_job] = intermediate_job @jobs_map_completed += 1 end lib/gisting/job_progress.rb @@ -13,6 +13,10 @@ module Gisting Result.new(job) rescue BusyClusterFail => e puts e.message + rescue Exception => e + puts "Caught exception..." + puts e.message + # job.stop! end alias :MapReduce :map_reduce lib/gisting/map_reduce.rb @@ -9,6 +9,7 @@ module Gisting @data_source = file_pattern @map_proc = proc setup_emit + pp ["map_runner", @data_source, @map_proc] end def Emit(key, value) lib/gisting/map_runner.rb @@ -7,6 +7,7 @@ module Gisting puts "MapServer Starting..." end + # Executes a MapRunner with a file pattern def receive_data(job) begin puts "Running MapRunner" @@ -14,12 +15,17 @@ module Gisting runner = MapRunner.new(file_pattern, proc) thread = Proc.new { runner.map! } - callback = Proc.new { send_data(runner.output) } + callback = Proc.new { send_reply(runner.output) } EM.defer(thread, callback) rescue => e pp [:map_server, e] end end + + def send_reply(output) + send_data(output) + puts "Completed MapRunner" + end end end lib/gisting/map_server.rb @@ -9,7 +9,7 @@ module Gisting @output_file = output @map_data_input = intermediate_file setup_emit - pp ["red_proc", @red_proc] + pp ["red_runner", @map_data_input, @red_proc, @output_file] end def Emit(count) lib/gisting/reduce_runner.rb @@ -7,8 +7,7 @@ module Gisting puts "ReduceServer Starting..." end - # Delegates an +map output+ task to a ReduceRunner and returns - # (via +send_data+) the name of the intermediate file + # Executes a ReduceRunner with an intermediate file def receive_data(job) begin puts "Running ReduceRunner" @@ -16,12 +15,17 @@ module Gisting runner = ReduceRunner.new(intermediate_file, proc, output) thread = Proc.new { runner.reduce! } - callback = Proc.new { send_data(runner.output) } + callback = Proc.new { send_reply(runner.output) } EM.defer(thread, callback) rescue => e pp [:red_server, e] end end + + def send_reply(output) + send_data(output) + puts "Completed ReduceRunner" + end end end lib/gisting/reduce_server.rb a06f1ae8d77b04658ff37d2c9c6c2fdaeae67aa8 Marc Chung mchung@gmail.com http://github.com/mchung/gisting/commit/ca26cfbf9f63a43c53ad0917d885e1cf9b052715 ca26cfbf9f63a43c53ad0917d885e1cf9b052715 2009-01-21T23:39:13-08:00 2009-01-21T23:39:13-08:00 Clean up the bin and use proper EM callbacks 184bcf2b4ade56cbdd2549a72ca86fabf38d99a7 Marc Chung mchung@gmail.com