Initial work on PingManager

lib/api/server_list_server.rb

@@ -20,8 +20,12 @@ class W3DHub
 
         @status  = Status.new(@data[:status])
 
-        @ping_interval = 30_000
+        # if we're on unix and using the PingManager then check every second since
+        # we're not _actually_ pinging the server.
+        @ping_interval = W3DHub.unix? ? 1_000 : 60_000
         @last_pinged = Gosu.milliseconds + @ping_interval + 1_000
+
+        Store.ping_manager.add_address(@address)
       end
 
       def update(hash)
@@ -49,6 +53,16 @@ class W3DHub
         if force_ping || Gosu.milliseconds - @last_pinged >= @ping_interval
           @last_pinged = Gosu.milliseconds
 
+          if W3DHub.unix?
+            average_ping = Store.ping_manager.ping_for(@address)
+
+            @ping = average_ping.negative? ? NO_OR_BAD_PING : average_ping
+
+            States::Interface.instance&.update_server_ping(self)
+          end
+
+          return unless W3DHub.windows?
+
           W3DHub::BackgroundWorker.foreground_parallel_job(
             lambda do
               W3DHub.command("ping #{@address} #{W3DHub.windows? ? '-n 3' : '-c 3'}") do |line|

lib/ping.rb

@@ -1,70 +1,166 @@
+require "async"
 require "socket"
+require "securerandom"
 
-ICMPHeader = Data.define(:type, :code, :checksum, :_ping_id, :_sequence_id, :data)
+class W3DHub
+  class Ping
+    ICMPHeader = Data.define(:type, :code, :checksum, :_ping_id, :_sequence_id, :data)
+    EchoRequest = Struct.new(:ping_id, :sequence_id, :data, :time, :timed_out)
 
-ICMP_ECHOREPLY = 0 # Echo reply
+    ICMP_ECHOREPLY = 0
-ICMP_ECHO      = 8 # Echo request
+    ICMP_ECHO      = 8
-ICMP_SUBCODE   = 0
+    ICMP_SUBCODE   = 0
 
-# Perform a checksum on the message.  This is the sum of all the short
+    BIT_PACKER = "C2 n3 A*".freeze
-# words and it folds the high order bits into the low order bits.
+    MINIMUM_INTERVAL = 250 # ms # intervals below 200ms are considered rude and may be dropped due to flooding.
-#
+    ECHO_REQUEST_HISTORY = 30 # 100 # keep the last n requests
-def checksum(msg)
-  length    = msg.length
-  num_short = length / 2
-  check     = 0
 
-  msg.unpack("n#{num_short}").each do |short|
+    attr_reader :address
-    check += short
+
+    def initialize(address:, count: 10, ttl: 120, interval: 1_000, data: nil)
+      @address = address
+      @count = count
+      @ttl = ttl
+      @interval = interval.to_i < MINIMUM_INTERVAL ? MINIMUM_INTERVAL : interval # ms
+      @data = data
+
+      # circular buffer
+      @echo_requests = Array.new(ECHO_REQUEST_HISTORY) { EchoRequest.new(-1, -1, "", nil, false) }
+      @echo_requests_index = 0
+
+      # NOTE: The PING_ID _might_ be overruled by the kernel and should not be used
+      #       to check that any received echo replies are ours.
+      #
+      #       Sequence ID and Data appear to be unmodified.
+      @ping_id = SecureRandom.hex.to_i(16) & 0xffff
+      @sequence_id = SecureRandom.hex.to_i(16) & 0xffff
+
+      addresses = Addrinfo.getaddrinfo(@address, nil, Socket::AF_INET, :DGRAM)
+      raise "NO ADDRESSES!" if addresses.empty?
+
+      @socket_address = addresses.sample.to_sockaddr
+
+      @socket = Socket.new(Socket::AF_INET, Socket::SOCK_DGRAM, Socket::IPPROTO_ICMP)
+      @socket.setsockopt(Socket::SOL_SOCKET, Socket::IP_TTL, @ttl)
+    end
+
+    # Perform a checksum on the message.  This is the sum of all the short
+    # words and it folds the high order bits into the low order bits.
+    def message_checksum(message)
+      length    = message.length
+      num_short = length / 2
+      check     = 0
+
+      message.unpack("n#{num_short}").each do |short|
+        check += short
+      end
+
+      check += message[length - 1, 1].unpack1("C") << 8 if (length % 2).positive?
+
+      check = (check >> 16) + (check & 0xffff)
+      ~((check >> 16) + check) & 0xffff
+    end
+
+    def random_data
+      SecureRandom.hex
+    end
+
+    def monotonic_time
+      Process.clock_gettime(:CLOCK_MONOTONIC, :millisecond)
+    end
+
+    def verified?(message)
+      data = message.unpack(BIT_PACKER)
+      checksum = data[2]
+
+      # set checksum in message to 0
+      data[2] = 0
+
+      checksum == message_checksum(data.pack(BIT_PACKER))
+    end
+
+    def request_complete?(request)
+      request.timed_out || !request.time.nil?
+    end
+
+    def packet_loss
+      completed_requests = @echo_requests.select { |r| request_complete?(r) }
+      failed_requests = completed_requests.select(&:timed_out)
+
+      # 0% packet loss 😎
+      return 0.0 if failed_requests.empty?
+
+      # 100% packet loss
+      return 1.0 if failed_requests.size == completed_requests.size
+
+      failed_requests.size / completed_requests.size.to_f
+    end
+
+    def average_ping
+      times = @echo_requests.select { |r| request_complete?(r) && !r.timed_out }.map(&:time)
+
+      return -1 unless times.size.positive?
+
+      times.sum.to_f / times.size
+    end
+
+    # returns true if any echo requests have completed (reply received or timed out) and packet loss is less than 30%
+    def okay?
+      completed_requests = @echo_requests.select { |r| request_complete?(r) }.size
+
+      completed_requests.positive? && packet_loss < 0.3
+    end
+
+    def ping(count = @count)
+      return if count <= 0
+
+      Async do |task|
+        @count.times do
+          task.Async do |subtask|
+            @sequence_id = (@sequence_id + 1) % 0xffff
+            data = @data || random_data
+
+            checksum = 0
+            message = [ICMP_ECHO, ICMP_SUBCODE, checksum, @ping_id, @sequence_id, data].pack(BIT_PACKER)
+            checksum = message_checksum(message)
+            message = [ICMP_ECHO, ICMP_SUBCODE, checksum, @ping_id, @sequence_id, data].pack(BIT_PACKER)
+
+            @socket.send(message, 0, @socket_address)
+
+            s = monotonic_time
+            request = @echo_requests[@echo_requests_index]
+            request.ping_id = @ping_id
+            request.sequence_id = @sequence_id
+            request.data = data
+            request.time = nil
+            request.timed_out = false
+            @echo_requests_index = (@echo_requests_index + 1) % ECHO_REQUEST_HISTORY
+
+            subtask.with_timeout(2) do
+              loop do
+                data, _addrinfo = @socket.recvfrom(1500)
+
+                # ignore corruption
+                next unless verified?(data)
+
+                header = ICMPHeader.new(*data.unpack(BIT_PACKER))
+
+                if header.type == ICMP_ECHOREPLY && header._sequence_id == request.sequence_id && header.data == request.data
+                  duration = monotonic_time - s
+                  request.time = duration
+
+                  break
+                end
+              end
+            rescue Async::TimeoutError
+              request.timed_out = true
+            end
+          end
+
+          # Don't send out pings in a flood, it's considered rude.
+          sleep @interval / 1000.0
+        end
+      end
+    end
   end
-
-  if (length % 2).positive?
-    check += msg[length-1, 1].unpack1("C") << 8
-  end
-
-  check = (check >> 16) + (check & 0xffff)
-  ~((check >> 16) + check) & 0xffff
-end
-
-ip_address = "127.0.0.1" # "example.com" # "timecrafters.org" #
-@ping_id = 92_459_064_892 & 0xffff
-@sequence = 1 % 65_536
-data = ""
-data_size = 56
-data_size.times { |n| data << (n % 256).chr }
-
-check = 0
-packer = "C2 n3 A" << data_size.to_s
-message = [ICMP_ECHO, ICMP_SUBCODE, check, @ping_id, @sequence, data].pack(packer)
-check = checksum(message)
-message = [ICMP_ECHO, ICMP_SUBCODE, check, @ping_id, @sequence, data].pack(packer)
-message_header = ICMPHeader.new(*[ICMP_ECHO, ICMP_SUBCODE, check, @ping_id, @sequence, data])
-socket = Socket.new(Socket::AF_INET, Socket::SOCK_DGRAM, Socket::IPPROTO_ICMP)
-socket.send(message, 0, Socket.pack_sockaddr_in(0, ip_address))
-s = Time.now
-loop do
-  data, _addrinfo = socket.recvfrom(1500)
-  pp [message, data]
-  header = ICMPHeader.new(*data.unpack("C2 n3 A*"))
-  pp [message_header, header]
-  # reply_type = data[20, 2].unpack1("C2")
-  # pp reply_type
-
-  ping_id = header._ping_id
-  sequence = header._sequence_id
-
-  case header.type
-  when ICMP_ECHOREPLY
-    puts "ECHOREPLY"
-  end
-
-  pp [@ping_id, ping_id]
-  pp [@sequence, sequence]
-
-  if ping_id == @ping_id && sequence == @sequence && reply_type == ICMP_ECHOREPLY
-    puts "PING OKAY: #{Time.now - s}s"
-    break
-  end
-
-  break
 end

lib/ping_manager.rb

@@ -0,0 +1,54 @@
+class W3DHub
+  class PingManager
+    Container = Struct.new(:pinger, :last_ping_time_ms)
+    PING_INTERVAL = 60_000
+
+    def initialize
+      @containers = {}
+      @addresses = []
+    end
+
+    def monitor(task)
+      task.async do |subtask|
+        while BackgroundWorker.alive?
+          # activate new addresses
+          @addresses.each do |address|
+            @containers[address] ||= Container.new(Ping.new(address: address), -PING_INTERVAL * 2)
+          end
+
+          # cleanup old addresses
+          @containers.each_key do |key|
+            @containers.delete(key) unless @addresses.find { |a| a == key }
+          end
+
+          # ping the pingers
+          @containers.each_value do |container|
+            next unless Gosu.milliseconds - container.last_ping_time_ms >= PING_INTERVAL
+
+            container.last_ping_time_ms = Gosu.milliseconds
+
+            subtask.async do
+              container.pinger.ping
+              pp [container.pinger.address, container.pinger.average_ping]
+            end
+          end
+
+          sleep 0.001
+        end
+      end
+    end
+
+    def add_address(address)
+      @addresses << address
+      @addresses.uniq!
+    end
+
+    def ping_for(address)
+      @containers[address]&.pinger&.average_ping&.round || -1
+    end
+
+    def remove_address(address)
+      @addresses.delete(address)
+    end
+  end
+end

lib/window.rb

@@ -7,6 +7,9 @@ class W3DHub
       Store[:server_list] = []
       Store[:settings] = Settings.new
       Store[:application_manager] = ApplicationManager.new
+      Store[:ping_manager] = PingManager.new
+
+      BackgroundWorker.parallel_job(-> { Async { |task| Store.ping_manager.monitor(task) } }, nil)
 
       Store[:main_thread_queue] = []
 

w3d_hub_linux_launcher.rb

@@ -106,6 +106,8 @@ require_relative "lib/cache"
 require_relative "lib/settings"
 require_relative "lib/ww_mix"
 require_relative "lib/ico"
+require_relative "lib/ping"
+require_relative "lib/ping_manager"
 require_relative "lib/broadcast_server"
 require_relative "lib/hardware_survey"
 require_relative "lib/game_settings"