<commit_before>require 'nanite/reducer'
require 'nanite/dispatcher'
require 'nanite/persistence'
#AMQP.logging = true

module Nanite
  class Runner
    def start
      EM.run{
        MQ.new.rpc('mapper', Mapper.new(Reducer.new))
      }
    end
  end  
  
  class Mapper
    
    def log *args
      p args
    end
    
    def initialize(reducer)
      @reducer = reducer
      @db = Nanite::MapperStore.new
      @nanites = @db.load_agents
      @amq = MQ.new
      @amq.queue("mapper.pings").subscribe{ |msg|
        handle_pong(Marshal.load(msg))
      }
      @pings = {}
      send_pings
      EM.add_periodic_timer(60) { send_pings }
      log "loaded agents from store", @nanites.keys
    end
    
    def handle_pong(pong)
      if pongs = @pings[pong.token]
        pongs.reject! {|p| p == pong.from }
      end  
    end
    
    def send_pings
      log "send_ping:"
      tok = Nanite.gen_token
      @pings[tok] = []
      @nanites.keys.each do |agent|
        ping = Nanite::Ping.new(tok, agent, "mapper.pings")
        @pings[ping.token] << ping.to
        @amq.queue(agent).publish(Marshal.dump(ping))
      end
      EM.add_timer(2) do
        if @pings[tok].size == 0
          log "got all pings" 
        else
          log "missing pongs for:", @pings[tok]
          @pings[tok].each do |a|
            @nanites.delete(a)
            @db.delete_agent(a)
            log "removed #{a} from mapping/discovery"
          end  
        end
        @pings.delete(tok)  
      end  <commit_msg>no need for reducer to be passed into mapper anymore<commit_after>require 'nanite/reducer'
require 'nanite/dispatcher'
require 'nanite/persistence'
#AMQP.logging = true

module Nanite
  class Runner
    def start
      EM.run{
        MQ.new.rpc('mapper', Mapper.new)
      }
    end
  end  
  
  class Mapper
    
    def log *args
      p args
    end
    
    def initialize
      @db = Nanite::MapperStore.new
      @nanites = @db.load_agents
      log "loaded agents from store", @nanites.keys
      @amq = MQ.new
      @amq.queue("mapper.pings").subscribe{ |msg|
        handle_pong(Marshal.load(msg))
      }
      @pings = {}
      send_pings
      EM.add_periodic_timer(60) { send_pings }
    end
    
    def handle_pong(pong)
      if pongs = @pings[pong.token]
        pongs.reject! {|p| p == pong.from }
      end  
    end
    
    def send_pings
      log "send_pings"
      tok = Nanite.gen_token
      @pings[tok] = []
      @nanites.keys.each do |agent|
        ping = Nanite::Ping.new(tok, agent, "mapper.pings")
        @pings[ping.token] << ping.to
        @amq.queue(agent).publish(Marshal.dump(ping))
      end
      EM.add_timer(3) do
        if @pings[tok].size == 0
          log "got all pongs" 
        else
          log "missing pongs for:", @pings[tok]
          @pings[tok].each do |a|
            @nanites.delete(a)
            @db.delete_agent(a)
            log "removed #{a} from mapping/discovery"
          end  
        end
        @pings.delete(tok)  
      end  
{'fraction_non_alphanumeric': 0.11612475116124751, 'fraction_numerical': 0.0026542800265428003, 'mean_word_length': 1.8389830508474576, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2056460', 'n_tokens_mistral': 1083, 'n_tokens_neox': 1057, 'n_words': 225}