docs/links.txt @@ -7,8 +7,10 @@ Nanite.identity = 'mapper' db, password = ARGV[0], ARGV[1] || 'testing' -Nanite::Runner.new.start :user => Nanite.identity, - :pass => password, - :vhost => '/nanite', - :host => '0.0.0.0', - :db => db \ No newline at end of file +Nanite::Runner.new.start :user => Nanite.identity, + :pass => password, + :vhost => '/nanite', + :host => '0.0.0.0', + :db => db, + :ping_time => 15, + :heartbeat_time => 10 \ No newline at end of file bin/mapper @@ -2,7 +2,7 @@ puts `rabbitmqctl add_vhost /nanite` -%w[mapper client merb tom jay edward yehuda james corey josh ezra womble jeff loren lance sam kevin kirk].each do |agent| +%w[mapper shoes client merb tom jay edward yehuda james corey josh ezra womble jeff loren lance sam kevin kirk].each do |agent| puts `rabbitmqctl add_user #{agent} testing` puts `rabbitmqctl map_user_vhost #{agent} /nanite` puts `rabbitmqctl set_permissions #{agent} /nanite /data all` bin/rabbitconf @@ -57,7 +57,7 @@ module Nanite class << self attr_accessor :identity - attr_accessor :default_resources, :last_ping + attr_accessor :default_resources, :last_ping, :ping_time def root @root ||= File.expand_path(File.dirname(__FILE__)) @@ -104,7 +104,8 @@ module Nanite end end - Nanite.amq.queue(Nanite.identity, :exclusive => true).subscribe{ |msg| + Nanite.amq.queue(Nanite.identity, :exclusive => true).subscribe{ |headers,msg| + p headers Nanite::Dispatcher.handle(Marshal.load(msg)) } end lib/nanite.rb @@ -19,32 +19,5 @@ module Nanite "<Actor[#{self.class.name}]: provides=#{provides.map{|n| n.to_s }.join(', ')}>" end - def spawn(arg, *argv, &block) - # setup output hash - output = {:stdout => '', :stderr => ''} - default_options = {'stdout' => output[:stdout], 'stderr' => output[:stderr]} - - # add default options to fill output hash - if argv.size > 1 and Hash === argv.last - argv.last.merge!(default_options) - else - argv.push default_options - end - - - # catch any spawn errors and return - begin - status = Open4::spawn(arg, *argv) - rescue Open4::SpawnError => e - raise ActorInternalError, e.message - end - - if block_given? - formatted_output = yield(output[:stdout]) - output[:stdout] = formatted_output - end - - output.merge({:status => status}) - end end end \ No newline at end of file lib/nanite/actor.rb @@ -7,14 +7,17 @@ end require File.dirname(__FILE__) + "/../nanite" require File.dirname(__FILE__) + "/../nanite/agent" -name = 'shoes' +name, pass = 'shoes', 'testing' Nanite.identity = name Nanite.default_resources = [] ARGV.clear Thread.new do EM.run do + AMQP.start :user => Nanite.identity, + :pass => pass, + :vhost => '/nanite' Shoes.p "running event loop" - Nanite.run_event_loop + Nanite.run_event_loop false end end lib/nanite/client.rb @@ -8,8 +8,10 @@ module Nanite def start(opts={}) EM.run{ db = opts.delete(:db) + ping_time = opts.delete(:ping_time) || 30 + heartbeat_time = opts.delete(:heartbeat_time) || 20 AMQP.start opts - MQ.new.rpc('mapper', Mapper.new(db)) + MQ.new.rpc('mapper', Mapper.new(db, ping_time, heartbeat_time)) } end end @@ -20,7 +22,9 @@ module Nanite p args end - def initialize(db) + def initialize(db, ping_time, heartbeat_time) + @ping_time = ping_time + @heartbeat_time = heartbeat_time @db = Nanite::MapperStore.new(db) @nanites = @db.load_agents log "loaded agents from store", @nanites.keys @@ -28,7 +32,7 @@ module Nanite @pings = {} setup_as_slave EM.add_periodic_timer(30) { log "current nanites(#{@nanites.keys.size}):", @nanites.keys } - EM.add_periodic_timer(30) { send_pings if @role == :master } + EM.add_periodic_timer(@ping_time) { send_pings if @role == :master } end def promote_to_master @@ -47,7 +51,7 @@ module Nanite @amq.queue("mapper.slave",:exclusive => true).subscribe{ |msg| handle_slave_packet(Marshal.load(msg)) } - EM.add_periodic_timer(20) { check_master if @role == :slave } + EM.add_periodic_timer(@heartbeat_time) { check_master if @role == :slave } check_master request_mapper_state log "running as slave" @@ -69,7 +73,7 @@ module Nanite ping = Nanite::Ping.new(tok, 'mapper.master.heartbeat', "mapper.slave") @master_pings << ping @amq.queue('mapper.master.heartbeat').publish(Marshal.dump(ping)) - EM.add_timer(15) do + EM.add_timer(@heartbeat_time * 0.66) do unless @master_pings.size == 0 log "no pong from master, promoting myself to master" promote_to_master @@ -155,7 +159,7 @@ module Nanite @pings[ping.token] << ping.to @amq.queue(agent).publish(Marshal.dump(ping)) end - EM.add_timer(20) do + EM.add_timer(@ping_time * 0.66) do if @pings[tok].size == 0 log "got all pongs" else @@ -208,7 +212,7 @@ module Nanite def send_op(op, target) log "send_op:", op, target - @amq.queue(target).publish(Marshal.dump(op)) + @amq.queue(target).publish(Marshal.dump(op), :from => 'mapper') end def allowed?(from, to) lib/nanite/mapper.rb 7ef29bea1b7b6277438ea039a1ec525c76eef764 Ezra Zygmuntowicz ez@engineyard.com http://github.com/bmizerany/nanite/commit/ba6735ab1b594ec8a0f5cdad330db21d4cb56111 ba6735ab1b594ec8a0f5cdad330db21d4cb56111 2008-08-19T17:08:48-07:00 2008-08-19T17:08:48-07:00 latest touch ups 662abd5fbd49636cb6d9572483da6e64ce0f88eb Ezra Zygmuntowicz ez@engineyard.com