@@ -5,7 +5,9 @@ require "optparse" require File.dirname(__FILE__) + '/../lib/nanite' options = {:root => Dir.pwd, - :services => [] } + :services => [], + :log_level => 'info', + :ping_time => 15 } # Build a parser for the command line arguments opts = OptionParser.new do |opts| @@ -46,14 +48,30 @@ opts = OptionParser.new do |opts| opts.on("-v", "--vhost VHOST", "This flag is for setting the rabbitmq vhost") do |vhost| options[:vhost] = vhost end - + opts.on("-j", "--json", "This flag is for using JSON as the wire format rather then marshaled ruby objects") do |json| options[:format] = :json end + opts.on("-d", "--daemonize", "This flag is for run the nanite as a daemon") do |d| + options[:daemonize] = true + end + + opts.on("--ping-time PINGTIME", "This flag is for setting the ping time, or how often the nanites contact the mappers") do |ping| + options[:ping_time] = ping + end + opts.on("-s", "--services SERVIVES", "This flag is for setting the nanite's global services: -s /slice/42,/foo/3") do |res| options[:services] = res.split(/,/).map { |r| r.strip } end + + opts.on("-l", "--log-level LEVEL", "This flag is for setting the log level") do |level| + options[:log_level] = level + end + + opts.on("-a", "--actor-threaded", "This flag is will spawn actors in multi-threaded mode") do |res| + options[:threaded_actors] = true + end end bin/nanite @@ -1,10 +1,9 @@ -class Mock < Nanite::Actor - expose :list +class Clock < Nanite::Actor + expose :time - def list(payload) - p "got request" - [1,2,3] + def time(payload) + Time.now end end -Nanite::Dispatcher.register(Mock.new) \ No newline at end of file +Nanite::Dispatcher.register(Clock.new) \ No newline at end of file examples/myagent/actors/mock.rb @@ -12,6 +12,7 @@ require 'nanite/marshal' require 'nanite/console' require 'extlib' require 'json' +require 'logger' module Nanite @@ -22,7 +23,9 @@ module Nanite attr_accessor :identity, :format, :status_proc, :results, :root, :vhost, :file_root, :files, :host attr_accessor :default_services, :last_ping, :ping_time - + + attr_writer :log_level + include FileStreaming def send_ping @@ -49,14 +52,30 @@ module Nanite rescue LoadError end Dir["#{Nanite.root}/actors/*.rb"].each do |actor| - puts "loading actor: #{actor}" + Nanite.log.info "loading actor: #{actor}" require actor end end + + def log_level + @log_level || Logger::INFO + end + def levels + @levels ||= { + 'fatal' => Logger::FATAL, + 'error' => Logger::ERROR, + 'warn' => Logger::WARN, + 'info' => Logger::INFO, + 'debug' => Logger::DEBUG + } + end + def start(opts={}) config = YAML::load(IO.read(File.expand_path(File.join(opts[:root], 'config.yml')))) rescue {} opts = config.merge(opts) + + Nanite.log_level = levels[opts[:Log_level]] Nanite.root = opts[:root] Nanite.format = opts[:format] || :marshal Nanite.identity = opts[:identity] || Nanite.gensym @@ -65,6 +84,8 @@ module Nanite Nanite.file_root = opts[:file_root] || "#{Nanite.root}/files" Nanite.default_services = opts[:services] || [] + daemonize(opts[:log_file] || "#{Nanite.identity}.log") if opts[:daemonize] + AMQP.start :user => opts[:user], :pass => opts[:pass], :vhost => Nanite.vhost, @@ -74,14 +95,20 @@ module Nanite load_actors advertise_services - EM.add_periodic_timer(15) do + EM.add_periodic_timer((opts[:ping_time]||15).to_i) do send_ping end Nanite.amq.queue(Nanite.identity, :exclusive => true).subscribe{ |msg| - Nanite::Dispatcher.handle(Nanite.load_packet(msg)) + if opts[:threaded_actors] + Thread.new(msg) do |msg_in_thread| + Nanite::Dispatcher.handle(Nanite.load_packet(msg_in_thread)) + end + else + Nanite::Dispatcher.handle(Nanite.load_packet(msg)) + end } - start_console if opts[:console] + start_console if opts[:console] && !opts[:daemonize] end def reducer @@ -115,6 +142,15 @@ module Nanite @results ||= {} end + def log + @log ||= begin + log = Logger.new((Nanite.root||Dir.pwd) / "nanite.#{Nanite.identity}.log") + log.level = Nanite.log_level + log + end + @log + end + def gensym values = [ rand(0x0010000), @@ -127,5 +163,15 @@ module Nanite ] "%04x%04x%04x%04x%04x%06x%06x" % values end + + protected + def daemonize(log_file) + exit if fork + Process.setsid + exit if fork + $stdin.reopen("/dev/null") + $stdout.reopen(log_file, "a") + $stderr.reopen($stdout) + end end end lib/nanite.rb @@ -24,14 +24,18 @@ module Nanite def handle(packet) case packet when Nanite::Pong + Nanite.log.debug "handling Pong: #{packet}" Nanite.last_ping = Time.now when Nanite::Advertise + Nanite.log.debug "handling Advertise: #{packet}" Nanite.last_ping = Time.now Nanite.advertise_services when Nanite::Request + Nanite.log.debug "handling Request: #{packet}" result = dispatch_request(packet) Nanite.amq.queue(packet.reply_to).publish(Nanite.dump_packet(result)) if packet.reply_to when Nanite::Result + Nanite.log.debug "handling Result: #{packet}" Nanite.reducer.handle_result(packet) end end lib/nanite/dispatcher.rb @@ -7,10 +7,43 @@ module Nanite attr_accessor :mapper + # Make a nanite request which expects a response. + # + # ==== Parameters + # type<String>:: The dispatch route for the request + # payload<Object>:: Payload to send. This will get marshalled en route + # + # ==== Options + # :selector<Symbol>:: Method for selecting an actor. Default is :least_loaded. + # :least_loaded:: Pick the nanite which has the lowest load. + # :all:: Send the request to all nanites which respond to the service. + # :random:: Randomly pick a nanite. + # :rr: Select a nanite according to round robin ordering. + # :timeout<Numeric>:: The timeout in seconds before giving up on a response. + # Defaults to 60. + # :target<String>:: Select a specific nanite via identity, rather than using + # a selector. + # + # ==== Block Parameters + # :results<Object>:: The returned value from the nanite actor. + # def request(type, payload="", opts = {}, &blk) Nanite.mapper.request(type, payload, opts, &blk) end + # Make a nanite request which does not expect a response. + # + # ==== Parameters + # type<String>:: The dispatch route for the request + # payload<Object>:: Payload to send. This will get marshalled en route + # + # ==== Options + # :selector<Symbol>:: Method for selecting an actor. Default is :least_loaded. + # :least_loaded:: Pick the nanite which has the lowest load. + # :all:: Send the request to all nanites which respond to the service. + # :random:: Randomly pick a nanite. + # :rr: Select a nanite according to round robin ordering. + # def push(type, payload="", opts = {}) Nanite.mapper.push(type, payload, opts) end @@ -28,9 +61,6 @@ module Nanite end attr_accessor :nanites, :timeouts - def log *args - p args - end def initialize(ping_time) @identity = Nanite.gensym @@ -39,7 +69,7 @@ module Nanite @amq = MQ.new @timeouts = {} setup_queues - log "starting mapper with nanites(#{@nanites.keys.size}):", @nanites.keys + Nanite.log.info "starting mapper with nanites(#{@nanites.keys.size}):\n#{@nanites.keys.join(',')}" EM.add_periodic_timer(@ping_time) do check_pings EM.next_tick { check_timeouts } @@ -87,7 +117,7 @@ module Nanite private def setup_queues - log "setting up queues" + Nanite.log.debug "setting up queues" @amq.queue("heartbeat#{@identity}",:exclusive => true).bind(@amq.fanout('heartbeat')).subscribe{ |ping| handle_ping(Nanite.load_packet(ping)) } @@ -114,7 +144,7 @@ module Nanite @nanites.each do |name, state| if (time - state[:timestamp]) > @ping_time @nanites.delete(name) - log "removed #{name} from mapping/registration" + Nanite.log.info "removed #{name} from mapping/registration" end end end @@ -123,7 +153,7 @@ module Nanite @nanites[reg.identity] = {:timestamp => Time.now, :services => reg.services, :status => reg.status} - log "registered:", reg.identity, @nanites[reg.identity] + Nanite.log.info "registered: #{reg.identity}, #{@nanites[reg.identity]}" end def least_loaded(res) @@ -157,7 +187,7 @@ module Nanite def check_timeouts - puts "checking timeouts" + Nanite.log.debug "checking timeouts" time = Time.now @timeouts.each do |tok, timeout| if time > timeout lib/nanite/mapper.rb @@ -41,7 +41,7 @@ module FileStreaming when Nanite::FileChunk @dest.write(packet.chunk) when Nanite::FileEnd - puts "file written: #{@dest}" + Nanite.log.debug "file written: #{@dest}" @dest.close if cback = Nanite.callbacks[@domain] cback.call(@filename, packet.meta) @@ -53,7 +53,7 @@ module FileStreaming end def subscribe_to_files(domain='global', &blk) - puts "subscribing to file broadcasts for #{domain}" + Nanite.log.info "subscribing to file broadcasts for #{domain}" @files ||= {} Nanite.callbacks[domain] = blk if blk Nanite.amq.queue("files#{Nanite.identity}").bind(Nanite.amq.topic('file broadcast'), :key => "nanite.filepeer.#{domain}").subscribe{ |packet| lib/nanite/streaming.rb dd5ffebb4b554f5c08de25343cb36ee1d7e6ead4 6ca5dfa0328217ef8373c2256de766fc9f2fcef8 raggi jftucker@gmail.com http://github.com/shift/nanite/commit/57e815749702e52cb6893fe73ea18f537de80caa 57e815749702e52cb6893fe73ea18f537de80caa 2008-12-17T23:33:39-08:00 2008-12-17T23:33:39-08:00 Merge branch 'master' of git://github.com/ezmobius/nanite * 'master' of git://github.com/ezmobius/nanite: Added in config.yml and command line setting to spawn actors in multi-threaded mode document request document push Update README with a (painful) solution for rabbitmqctl dying with {badrpc,nodedown} make ping time a little more robust make ping time configurable added actual log levels updating nanite to support a true log file adding logger code so we log in the right place implemented nanite-daemonization support Conflicts: lib/nanite.rb 66e515c18632f3e436e888a05712059d470130ff raggi jftucker@gmail.com