3
\$\begingroup\$

Without ping-pong websocket connections are usually dropped by intermediary proxies/routers after some time of inactivity, so you can't go make yourself a coffee because your chat connection will break by the time you come back ;-] (most browsers such as Chrome or Firefox currently do not do any automatic pinging and to my knowledge, there's even no JS API to do it manually). Java websocket API provides primitive methods for sending pings and reacting to pongs (reacting to pings is automatic), but doing this manually in every project is tedious, error prone, time wasting etc etc. This class provides an easy way to automate it.

This used to be a small and pretty simple class before I had to add RTT reporting to it. After that, some unobvious trade-offs and design choices had to be made. I'm mainly looking for opinions regarding these (they are documented in the comments starting with a phrase "design decision note") as I may be biased by the specific use-cases I was dealing with, so a broader perspective would be most welcome. Other than that, comments on API convenience and suggestions for its improvements would be very useful as well. Last, but not least, general comments on code & documentation quality as well as possible bug spotting are always great :)

The class is available on github and in maven-central both in javax and jakarta flavours. The below version comes from the cr-stackexchange-1 tag and tries to somehow deal with reporting of lost or timed-out pongs: these are not reported at all yet in the latest release: see "design decision note" below in the code for problems related to it.

Additional info: RTT reporting was originally requested by folks who develop some real-time online games. They use it mostly on their server side to calculate and adjust per-participant handicaps on the fly during matches or something like that, so they use intervals of about 200-500ms (to be honest I haven't yet put an effort to fully understand how they use it as I'm not really much of a gamer and at first it seemed like a straightforward thing on my side: it doesn't seem that straightforward anymore and I will definitely need to sit with them soon to analyze their use-case). Recently they've reported a bug/feature-request that they would love to receive reports about lost/timed-out pongs. Initially I've implemented sending such reports from scheduler's thread in PingPongPlayer.sendPing(...) method as soon as a loss was detected and then all hell broke out: both servers and clients that were using RTT reporting started to crash/misbehave as everyone (including myself and my own Endpoints ;-] ) silently expected that report delivery would adhere to container-thread concurrency contract... The below version tries to meet this expectation for the price of delaying loss reports. This fixed my own Endpoint crashes, but I guess some ppl may not be entirely happy because of the delayed loss reports. That's why before committing to any approach I decided to ask for opinions here, to have some more view-points to work with besides mine own and that of my good old online-gaming folks.

explanation for subsequent reviewers: the puzzling if (Thread.interrupted()) break; is to stop the pinging if service.stop() happens in the middle of the loop: true in pingingTask.cancel(true) sends an interrupt to the worker thread if it happens that the task being cancelled is just being executed. I will add a comment to this line fore sure.

Even more clarifications

So first, let me emphasize again, that the main purpose of this utility is to keep websocket connections alive if there's no app-level activity (intermediary proxies and NAT routers tend to drop connections after some time of inactivity). This is especially important for web app servers as neither Chrome nor Firefox currently provide any pinging capabilities (neither automatic nor manual). Expecting timely pongs and RTT discovery are "bonus" features. It is a non-goal to provide a new higher-level/alternative API/protocol over websockets, but merely to make websocket usage in Java a bit less tedious/problematic.

motivation for expect-timely-pongs mode

expect-timely-pongs is intended to detect and disconnect buggy, dead-locked peers. It was added after a websocket service of one of my associates providing some form of RPC-over-websocket was unintentionally DoS-ed by another service (acting as a websocket client to the first one), that was keep spawning new connections on and on as its instances/threads were getting dead-locked.
expect-timely-pongs was not designed to detect dead TCP connections and I'm not sure if it is able to provide any real value in this area in its current form (I'm completely almost sure that setting TCP KEEPALIVE flag will do a better job in this regard).

motivation for keep-alive-only mode

After the above situation, for a moment expect-timely-pongs became the only mode of operation, but I received a sad email from a developer who was using this utility as a part of her desktop Java app acting as a websocket client. Some of the users of this app often had to work over a very poor internet connection resulting in frequent timeouts and automatic disconnects. This was extremely frustrating for these users as upon disconnecting they were losing their state on the server and usually wanted to wait patiently until their connectivity improved (the server was not in control of the developer of this desktop client app). As a quick workaround the developer set a very high failure limit, but we agreed, that in case of user-agent client apps, disconnecting should generally be a user decision, not automatic and since then the utility has 2 modes of operation.

(NEWER VERSION AVAILABLE: I've applied many of the suggestions from J_H's answer and posted for a review in a new question here)

// Copyright 2021 Piotr Morgwai Kotarbinski, Licensed under the Apache License, Version 2.0
package pl.morgwai.base.servlet.utils;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.BiConsumer;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.websocket.*;
import javax.websocket.CloseReason.CloseCodes;
import javax.websocket.RemoteEndpoint.Async;

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;



/**
 * Automatically pings and handles pongs from websocket {@link Session connections}.
 * Depending on constructor used, operates in either
 * {@link #WebsocketPingerService(int, int, boolean) expect-timely-pongs mode} or
 * {@link #WebsocketPingerService(int, boolean) keep-alive-only mode}.
 * The service can be used both on the client and the server side.
 * <p>
 * Instances are usually created at app startups and stored in locations easily reachable for
 * {@code Endpoint} instances or a code that manages them (for example as a
 * {@code ServletContext} attribute, a field in a class that creates client
 * {@link Session connections} or on some static var).<br/>
 * At app shutdowns, {@link #stop()} should be called to terminate the pinging
 * {@link ScheduledExecutorService scheduler}.</p>
 * <p>
 * Connections can be registered for pinging using {@link #addConnection(Session)}
 * and deregister using {@link #removeConnection(Session)}.</p>
 * <p>
 * If round-trip time discovery is needed, {@link #addConnection(Session, BiConsumer)} variant may
 * be used to receive RTT reports on each pong.</p>
 */
public class WebsocketPingerService {



    // design decision note: while it is possible to use unsolicited pongs for keep-alive-only,
    // some ping-pong implementations confuse them with malformed pongs and close connections.
    // Furthermore, using ping-pong allows to provide RTT reports in keep-alive-only mode also.



    /** 55s as majority of proxies and NAT routers have a timeout of at least 60s. */
    public static final int DEFAULT_INTERVAL_SECONDS = 55;

    final int failureLimit;  // negative value means keep-alive-only mode
    final boolean synchronizeSending;

    final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    /** Periodic on {@link #scheduler}, executes {@link #pingAllConnections()}. */
    final ScheduledFuture<?> pingingTask;
    final Random random = new Random();  // for ping content
    final ConcurrentMap<Session, PingPongPlayer> connectionPingPongPlayers =
            new ConcurrentHashMap<>();



    /**
     * Configures and starts the service in {@code expect-timely-pongs} mode: each timeout adds to a
     * given {@link Session connection}'s failure count, unmatched pongs are ignored.
     * @param interval interval between pings and also timeout for pongs. While this class does not
     *     enforce any hard limits, values below 100ms are probably not a good idea in most cases
     *     and anything below 20ms is pure Sparta.
     * @param unit unit for {@code interval}.
     * @param failureLimit limit of lost or timed-out pongs: if exceeded the given
     *     {@link Session connection} is closed with {@link CloseCodes#PROTOCOL_ERROR}. Each
     *     matching, timely pong resets the {@link Session connection}'s failure counter.
     * @param synchronizeSending whether to synchronize ping sending on a given
     *     {@link Session connection}. Whether it is necessary depends on the container
     *     implementation being used. For example it is not necessary on Jetty, but it is on Tomcat:
     *     see <a href="https://bz.apache.org/bugzilla/show_bug.cgi?id=56026">this bug report</a>.
     *     <br/>
     *     When using containers that do require such synchronization, all other message sending by
     *     {@code Endpoint}s must also be synchronized on the {@link Session connection} (please
     *     don't shoot the messenger...).
     * @throws IllegalArgumentException if {@code interval} is smaller than 1ms.
     */
    public WebsocketPingerService(
        long interval,
        TimeUnit unit,
        int failureLimit,
        boolean synchronizeSending
    ) {
        this.failureLimit = failureLimit;
        this.synchronizeSending = synchronizeSending;
        pingingTask = scheduler.scheduleAtFixedRate(this::pingAllConnections, 0L, interval, unit);
    }

    // design decision note: using interval as timeout simplifies things A LOT. Using a separate
    // SHORTER duration for a timeout is still pretty feasible and may be implemented if there's
    // enough need for it. Allowing a timeouts longer than intervals OTHO would required stacking
    // of pings and is almost certainly not worth the effort.



    /**
     * Configures and starts the service in {@code keep-alive-only} mode:
     * {@link Session connections} will <b>not</b> be actively closed unless an {@link IOException}
     * occurs. The params have the similar meaning as in
     * {@link #WebsocketPingerService(long, TimeUnit, int, boolean)}.
     */
    public WebsocketPingerService(long interval, TimeUnit unit, boolean synchronizeSending) {
        this(interval, unit, -1, synchronizeSending);
    }



    // block of constructor variants using some default values:
    /**
     * Calls {@link #WebsocketPingerService(long, TimeUnit, int, boolean)
     * WebsocketPingerService(intervalSeconds, SECONDS, failureLimit, synchronizeSending)}
     * ({@code expect-timely-pongs} mode).
     */
    public WebsocketPingerService(int intervalSeconds, int failureLimit, boolean synchronizeSending)
    {
        this(intervalSeconds, SECONDS, failureLimit, synchronizeSending);
    }
    /**
     * Calls {@link #WebsocketPingerService(long, TimeUnit, int, boolean)
     * WebsocketPingerService(intervalSeconds, SECONDS, failureLimit, false)}
     * ({@code expect-timely-pongs} mode).
     */
    public WebsocketPingerService(int intervalSeconds, int failureLimit) {
        this(intervalSeconds, SECONDS, failureLimit, false);
    }
    /**
     * Calls {@link #WebsocketPingerService(long, TimeUnit, int, boolean)
     * WebsocketPingerService(interval, unit, failureLimit, false)} ({@code expect-timely-pongs
     * } mode).
     */
    public WebsocketPingerService(long interval, TimeUnit unit, int failureLimit) {
        this(interval, unit, failureLimit, false);
    }
    /**
     * Calls {@link #WebsocketPingerService(long, TimeUnit, boolean)
     * WebsocketPingerService(intervalSeconds, SECONDS, synchronizeSending)}
     * ({@code keep-alive-only} mode).
     */
    public WebsocketPingerService(int intervalSeconds, boolean synchronizeSending) {
        this(intervalSeconds, SECONDS, synchronizeSending);
    }
    /**
     * Calls {@link #WebsocketPingerService(long, TimeUnit, boolean)
     * WebsocketPingerService(intervalSeconds, SECONDS, false)} ({@code keep-alive-only} mode).
     */
    public WebsocketPingerService(int intervalSeconds) {
        this(intervalSeconds, SECONDS, false);
    }
    /**
     * Calls {@link #WebsocketPingerService(long, TimeUnit, boolean)
     * WebsocketPingerService(interval, unit, false)} ({@code keep-alive-only} mode).
     */
    public WebsocketPingerService(long interval, TimeUnit unit) {
        this(interval, unit, false);
    }
    /**
     * Calls {@link #WebsocketPingerService(long, TimeUnit, boolean)
     * WebsocketPingerService}<code>({@link #DEFAULT_INTERVAL_SECONDS}, SECONDS, false)</code>
     * ({@code keep-alive-only} mode).
     */
    public WebsocketPingerService() {
        this(DEFAULT_INTERVAL_SECONDS, SECONDS, false);
    }



    /**
     * Registers {@code connection} for pinging.
     * Usually called in
     * {@link javax.websocket.Endpoint#onOpen(Session, javax.websocket.EndpointConfig) onOpen(...)}.
     */
    public void addConnection(Session connection) {
        addConnection(connection, null);
    }

    /**
     * Registers {@code connection} for pinging and receiving round-trip time reports via
     * {@code rttObserver}.
     * Usually called in
     * {@link javax.websocket.Endpoint#onOpen(Session, javax.websocket.EndpointConfig) onOpen(...)}.
     * <p>
     * Upon receiving a pong matching the most recent ping sent to a given
     * {@link Session connection}, {@code rttObserver} will be invoked with the round-trip time in
     * nanoseconds as the second argument and the given {@link Session connection} as the first.</p>
     * <p>
     * {@code rttObserver} will be called by a container {@code Thread} bound by the websocket
     * {@code Endpoint} concurrency contract, so as with normal websocket event handling, it should
     * not be performing any long-running operations to not delay processing of subsequent events.
     * Particularly, if {@code rttObserver} processing or processing of any other event blocks
     * arrival of a pong, the corresponding RTT report will be inaccurate.</p>
     * <p>
     * If the most recent ping has timed out or has been lost, {@code rttObserver} will be called
     * with a negative value as the second argument upon arriving of a <u>subsequent</u> pong. This
     * means, that if the other side does not send pongs at all, {@code rttObserver} will not be
     * called at all either: this is a consequence of the requirement for {@code rttObserver} to be
     * called by a container {@code Thread}. If RTT reports receiving is critical for a given app,
     * {@code expect-timely-pongs} mode should be used to disconnect misbehaving peers.<br/>
     * If more than 1 ping gets lost in a row and some pong finally arrives from the other side,
     * the number of reports indicating loss may be smaller than the actual number of pings lost.
     * </p>
     */
    public void addConnection(Session connection, BiConsumer<Session, Long> rttObserver) {
        connectionPingPongPlayers.put(
            connection,
            new PingPongPlayer(connection, failureLimit, synchronizeSending, rttObserver)
        );
    }

    // design decision note: it seems that in vast majority of cases it is most conveniently for
    // developers if a receiver of RTT reports is the Endpoint instance associated with the
    // connection which reports concern. Container Threads calling Endpoints are bound by a
    // concurrency contract requiring that each Endpoint instance is called by at most 1 Thread at a
    // time. Therefore it would create a lot of problems for developers if delivering  of RTT
    // reports didn't adhere to this contract either.



    /**
     * Removes {@code connection} from this service, so it will not be pinged anymore.
     * Usually called in
     * {@link javax.websocket.Endpoint#onClose(Session, CloseReason) onClose(...)}.
     * @return {@code true} if {@code connection} had been {@link #addConnection(Session) added} to
     *     this service before and has been successfully removed by this method, {@code false} if it
     *     had not been added and no action has taken place.
     */
    public boolean removeConnection(Session connection) {
        return connectionPingPongPlayers.remove(connection) != null;
    }



    /** Whether {@code connection} is {@link #addConnection(Session) registered} in this service. */
    public boolean containsConnection(Session connection) {
        return connectionPingPongPlayers.containsKey(connection);
    }



    /** The number of currently registered {@link Session connections}. */
    public int getNumberOfConnections() {
        return connectionPingPongPlayers.size();
    }



    /**
     * Stops the service.
     * After a call to this method the service becomes no longer usable and should be discarded.
     * @return {@link Session connections} that were registered at the time this method was called.
     */
    public Set<Session> stop(long timeout, TimeUnit unit) {
        pingingTask.cancel(true);
        scheduler.shutdown();
        for (var pingPongPlayer: connectionPingPongPlayers.values()) pingPongPlayer.deregister();
        try {
            scheduler.awaitTermination(timeout, unit);
        } catch (InterruptedException ignored) {}
        if ( !scheduler.isTerminated()) {  // this probably never happens
            log.warning("pinging scheduler failed to terminate");
            scheduler.shutdownNow();  // probably won't help as the task was cancelled already
        }
        final var remaining = Set.copyOf(connectionPingPongPlayers.keySet());
        connectionPingPongPlayers.clear();
        return remaining;
    }

    /** Calls {@link #stop(long, TimeUnit)} with a 500ms timeout. */
    public Set<Session> stop() {
        return stop(500L, MILLISECONDS);
    }



    /** Executed periodically in {@link #scheduler}'s {@link #pingingTask} */
    void pingAllConnections() {
        if (connectionPingPongPlayers.isEmpty()) return;
        final var pingData = new byte[8];  // enough to avoid collisions, but not overuse random
        random.nextBytes(pingData);
        for (var pingPongPlayer: connectionPingPongPlayers.values()) {
            if (Thread.interrupted()) break;
            pingPongPlayer.sendPing(pingData);
        }
    }



    /** Plays ping-pong with a single associated {@link Session connection}. */
    static class PingPongPlayer implements MessageHandler.Whole<PongMessage> {

        final Session connection;
        final Async connector;
        final int failureLimit;
        final boolean synchronizeSending;
        final BiConsumer<Session, Long> rttObserver;

        int failureCount = 0;
        /** Retained after the most recent ping for comparison with incoming pongs. */
        ByteBuffer pingDataBuffer;
        /**
         * Send timestamp of the most recent ping to which a matching pong has not been received
         * yet. {@code null} means that a matching pong to the most recent ping has been already
         * received.
         */
        Long pingTimestampNanos = null;
        /**
         * Raised in {@link #sendPing(byte[])} if the previous ping timed-out
         * ({@link #pingTimestampNanos} not {@code null}) to indicate that a loss report should be
         * sent to {@link #rttObserver} upon receiving a subsequent pong.
         */
        boolean previousPingTimedOut = false;



        /** For both modes: negative {@code failureLimit} means {@code keep-alive-only}. */
        PingPongPlayer(
            Session connection,
            int failureLimit,
            boolean synchronizeSending,
            BiConsumer<Session, Long> rttObserver
        ) {
            this.connection = connection;
            this.connector = connection.getAsyncRemote();
            this.synchronizeSending = synchronizeSending;
            this.rttObserver = rttObserver;
            this.failureLimit = failureLimit;
            connection.addMessageHandler(PongMessage.class, this);
        }



        /** Called by the {@link #scheduler}'s worker {@code  Thread}. */
        synchronized void sendPing(byte[] pingData) {
            if (pingTimestampNanos != null) {  // the previous ping has timed-out
                previousPingTimedOut = true;  // report loss on receiving a subsequent pong
                if (failureLimit >= 0) {  // expect-timely-pongs mode
                    failureCount++;
                    if (failureCount > failureLimit) {
                        closeFailedConnection("too many lost or timed-out pongs");
                        return;
                    }
                }
            }
            pingDataBuffer = ByteBuffer.wrap(pingData);  // retain for comparison with pongs
            try {
                if (synchronizeSending) {
                    synchronized (connection) {
                        connector.sendPing(pingDataBuffer);
                    }
                } else {
                    connector.sendPing(pingDataBuffer);
                }
                pingTimestampNanos = System.nanoTime();
                pingDataBuffer.rewind();  // required for comparing: see equals() javadoc
            } catch (IOException e) {
                // on most container implementations the connection is PROBABLY already closed, but
                // just in case:
                closeFailedConnection("failed to send ping");
            }
        }

        private void closeFailedConnection(String reason) {
            if (log.isLoggable(Level.FINE)) {
                log.fine("failure on connection " + connection.getId() + ": " + reason);
            }
            try {
                connection.close(new CloseReason(CloseCodes.PROTOCOL_ERROR, reason));
            } catch (IOException ignored) {}  // this MUST mean the connection is already closed...
        }



        /** Called by a container {@code Thread}. */
        @Override
        public void onMessage(PongMessage pong) {
            final var pongTimestampNanos = System.nanoTime();
            boolean reportPreviousPingTimedOut;
            Long rttToReport = null;
            synchronized (this) {
                reportPreviousPingTimedOut = rttObserver != null && previousPingTimedOut;
                if (pong.getApplicationData().equals(pingDataBuffer)) {
                    rttToReport = rttObserver != null && !(/*collision*/ pingTimestampNanos == null)
                            ? pongTimestampNanos - pingTimestampNanos
                            : null;
                    pingTimestampNanos = null;  // indicate the expected pong was received on time
                    failureCount = 0;
                }
            }
            if (reportPreviousPingTimedOut) rttObserver.accept(connection, -1L);
            if (rttToReport != null) rttObserver.accept(connection, rttToReport);
        }



        /**
         * Removes pong handler.
         * Called by the service on {@link Session connections} remaining after {@link #stop()}.
         */
        void deregister() {
            try {
                connection.removeMessageHandler(this);
            } catch (RuntimeException ignored) {
                // connection was closed in the mean time and some container implementations
                // throw a RuntimeException in case of any operation on a closed connection
            }
        }
    }



    static final Logger log = Logger.getLogger(WebsocketPingerService.class.getName());
}

\$\endgroup\$
2

2 Answers 2

2
\$\begingroup\$

as I may be biased by the specific use-cases I was dealing with, so a broader perspective ...

This is a wonderful remark, thank you for sharing. There's always more for each of us to see.

magic number

    /** 55s as majority of proxies and NAT routers have a timeout of at least 60s. */

Excellent, thank you for helping future maintenance engineers to understand

  • the design rationale, and
  • what classes of device to test against if they make a change

I feel this is a moderately long time, imposing only small overhead, and so there should be little motivation to support keepalive-only mode. If client pings, IMHO it is reasonable to expect a timely pong.

I will observe that fixed delays are undesirable, as common-mode events (e.g. network burst from cron job at the top of each hour) will cause network peers to all become synchronized together, similar to the "pull in" effect of a PLL. Standard advice is "jitter your timers!" by rolling a random number. Self synchronization badly affected North American core routers for more than a year, to the point where BGP4 now specifies details of how certain timers shall be jittered.

Oh, dear! I just saw the call to scheduleAtFixedRate(). My intent was along the lines of

    timeOfNextScheduledEvent += 55 + uniformRandomPlusOrMinus(3);  // seconds

but I suppose a scheduled event that randomly sleeps a few seconds before sending is the next best thing.

I completely agree that "allowing timeouts longer than intervals" and stacking them is out-of-scope. The engineering assumption at the core of this is that RTT from speed-of-light, network queueing, and CPU queueing, is much less than the configured ping interval. Seems like a solid assumption to me.

(I appreciate and agree with your other Design Decision notes.)

overloads

There's a fair number of WebsocketPingerService signatures in there. Maybe look at the dozen or two existing callers in production code, and cull the unused overloads?

interrupted

    void pingAllConnections() {
            ...
            if (Thread.interrupted()) break;

I confess I don't understand what's happening here.

Shouldn't we throw fatal InterruptedException if someone asked us to shutdown? And maybe also stop()? I'm just unclear on the contract, on the various responsibilities.

keep-alive

    /** For both modes: negative {@code failureLimit} means {@code keep-alive-only}. */

Maybe you need to support this? If so, it's worth writing down the reasons.

Consider ditching support for it.

If you retain it, consider enforcing that failureLimit shall always be positive, and introduce yet another boolean parameter to select keep-alive mode. I suggest this, just because I have seen a bunch of (nicely written) comments that have to keep explaining the peculiar meaning of a negative limit.

Alternatively, consider creating a Config object for the various options. (And, uhh, yeah, you have my sympathies about the whole TomCat synchronizeSending thing.)

           } catch (IOException ignored) {}

Thank you for the informative identifier in closeFailedConnection, very helpful.

Similarly in deregister, also accompanied by a helpful comment.

observer

OK, so I've read through a bunch of code, and I still don't exactly see how an observer would make use of this to offer an improved UX to a human user. Here's the crux of it:

            if (reportPreviousPingTimedOut) rttObserver.accept(connection, -1L);
            if (rttToReport != null) rttObserver.accept(connection, rttToReport);

Given that we only notify observer when something arrives from the network, it's unclear whether a limit of more than 1 would be useful.

I'm going to make up a scenario. For five minutes you're using an app that created a single WebSocket, and everything is peachy. You go to fix a cup of coffee. Some evil family member, perhaps your puppy, powers off the WiFi router so there's no internet connectivity.

At subsequent times

  1. you return to interact with the app, and
  2. the WAP is powered back on so connectivity is restored (maybe not in that order).

Every 55 seconds the app queues up some TCP payload to deliver. Now depending on the WiFi outage duration, TCP may have backed off to very long (more than a minute) retrans attempts, or may have given up, sent RST, and torn down the connection. Let's assume TCP has backed off but is still hopeful of receiving an ACK.

So we sent PING 1 and heard a PONG, then the WAP died at noon, we sent PING 2, then PING 3, the WAP powered up roughly a minute later, and at maybe 12:03-ish we get PONG 2 immediately followed by PONG 3. And we tell the observer -1, immediately followed by an RTT of many seconds. (A subsequent PING 4 would quickly yield a PONG 4 measurement of less than 500 msec.)

Here's where I'm puzzled. Should observer hear the -1 and immediately give up, displaying fatal error to user? Should observer hang out for a little while, hoping for the PONG 3 measurement?

We don't tell observer about outgoing PINGs. Would it maybe be more helpful to the UX if we did, and then observer expects response within 500 msec? Observer could timeout after 500 msec and display a Toast message or use some screen real estate to display a "no net!" icon.

Maybe around the fifty second mark we should send observer a -1 to indicate the PING was lost? That is, send the notification based on time, rather than due to onMessage() being called.

Summary: It's not clear how observer would use the current Public API to offer a better UX to the end user. It's worth writing up in ReadMe / javadoc.

alternate protocol

There's nothing especially wrong with what you send on the wire currently. But instead of random bytes, consider sending this:

Roll a secret GUID just once (it doesn't need to be very secret).

Fill the send buffer with (timestamp, SHA3(secret + timestamp)), or at least with a hash prefix.

Validate the hash in onMessage, rather than using .equals() on the most-recently-sent random bytes. Send observer strictly positive latency measurements. Let observer make the judgement call about what to do with "large" latencies.

\$\endgroup\$
2
  • \$\begingroup\$ Again I'm very grateful for your insightful comments! There's a lot to process here and I need to focus on something else for a couple of days, but I will definitely give you some more thoughts and explanations later. For now I've updated the OP with some background info on known RTT reporting usages: it may explain some things. Other than that, I will definitely try to add some jittering to ping intervals in the future versions (perhaps I will be able to reuse this). \$\endgroup\$
    – morgwai
    Jan 22 at 10:58
  • 1
    \$\begingroup\$ Regarding the puzzling if (Thread.interrupted()) break;: yeah, this definitely deserves a comment ;-) It's to stop the pinging if service.stop() happens in the middle of the loop (true in pingingTask.cancel(true) sends an interrupt to the worker thread if it happens that the task being cancelled is just being executed). \$\endgroup\$
    – morgwai
    Jan 22 at 11:02
0
\$\begingroup\$

Thank you for the additional review context.

existing design

Based on the code, context, and default parameter settings, I had interpreted this code as solving a detect dead connection use case more quickly than would be reported by TCP timeouts, especially when an idle connection suffers a network disconnect. Presumably the higher level app would resort to some combination of telling a human "no net!" and/or attempting to re-establish a new WebSocket.

Partly this is due to human psychology / reaction times, and the kinds of events that happen on a ~ 55 second time scale, that is, one probe per minute. In contrast a TCP retransmission of WebSocket data will typically be triggered within 100 ms (by SACK when the congestion window is open and we're streaming data) or within a couple of seconds (by an RTO retransmission timeout alarm expiring). Infrequent probes are unlikely to directly witness such events.

There is a rich and storied literature around TCP retransmission strategies, where we're largely concerned with measuring the unknown loads of core routers possessing unknown quantities of bandwidth and free buffer memory. Our goal at the TCP layer is to estimate the most effective strategy for retransmitting lost segments or quickly sending new ones. A WebSocket app, in contrast, sees (assuming no RST) a reliable byte stream which will deliver queued data after some variable latency. After an app writes a byte of data it is committed; there is no option to take it back, or Nagle-style coalesce it with subsequent sends. Lower layers will do what they're going to do, with little visibility or control available at the app layer.

It turns out the actual design goal was to continuously estimate latency of the underlying TCP connection.

proposed design

I would ditch the timer-based approach entirely, preferring synchronous Ack / Pong responses from each command sent.

Introduce a layer, maybe call it a "channel". The game application requests a channel, and it in turn obtains a WebSocket with its associated TCP connection. I will assume Nagling is disabled, that is, when the congestion window is open TCP may send immediately without awaiting an ACK for a segment sent a moment ago. When sending short, e.g. one-byte, messages, this lets transmit events be spaced together more closely than RTT roundtrip times.

Suppose the business domain is a multi-player vehicle Simulator where we observe and avoid hazards that keep popping up, or a multi-player FPS where we maneuver through a shoot-or-be-shot scenario. In all cases the cloud server sends a Notification that something new has happened in the player's environment, and after some positive delay (network queueing time + human reaction / think time) the player sends Response commands, such as moving or firing.

simple

The simplest thing that could possibly work would be for the channel layer to associate a Serial number with each Notification going out. Upon receiving one, the channel would immediately send an Ack / Pong (call it what you will) that includes Serial plus player's local timestamp. This allows the cloud server to continuously estimate RTT.

Server maintains a pqueue or a circular list of recently sent Serials, together with time of sending and optionally the message length in bytes. The wire encoding of Serial could be small, just the low order 8 or even 5 bits, given the reliable byte stream and the expectation that only a few unacknowledged messages will be outstanding at any given moment. (When the congestion window fills up, any app-level attempts to send will be blocked until TCP ACKs eventually open the window. Optionally storing message lengths would let us both estimate bandwidth and estimate whether next message is likely to fill the window. Server can deterministically increase observed RTT by sending a "large" message which will get chopped into ~ 1460 byte segments and which possibly will have to wait on multiple RTTs for ACKs to open the window.)

So server has a bunch of recent network latencies available for examination, to estimate how much network queueing delay happened before the Notification could display something new on player's screen. And then player might click the fire button to send a Fire command Response message, which will include local timestamp, and that server can additionally timestamp upon receipt. Which reveals the player's reaction time.

The game logic is then free to make adjustments, trying to put all players on an even playing field. For example, someone with measured 60 ms lag might be given more sensitive steering controls or more powerful rocket acceleration compared to another player with 20 ms lag.

NTP

There is a robust literature on time synchronization, much of it devoted to the Network Time Protocol. Each of its UDP packets has timestamp slots labeled {t1, t2, t3, t4}. The arrangement is client fills in t1 upon sending, server fills in t2 upon receipt then soon after fills in t3 upon sending reply, and finally client fills in t4 upon receipt. The timestamps are not directly comparable. Simplistically, think of some client vs server skew of 3600 seconds, or even 10 seconds. We might compute server CPU queueing delay of 5 ms with t3 - t2, and client's observed network RTT with t4 - t1. But comparing a client stamp with a server stamp tells us more about the skew than about packet latencies.

An NTP daemon retains the eight most recent measurements, applies a min() filter to them, and draws an inference about the probable offset between the two clocks. (There's also an inference about the different clock frequencies which is out-of-scope here.) The min-filter is crucial, and it tells us something about the physical world, assuming the measurement happened when router queues were mostly empty-ish. It tells us about speed-of-light geographic limits on latency, plus serialization latencies of installed equipment along the path. The other seven measurements tell us about dynamic network conditions, that is, they tell us about congestion-induced queueing delays. Notice that our channel knows about some of those queueing delays, since it knows about its recent send events and the total number of bytes it recently asked TCP to queue up for delivery.

Layering our channel atop TCP rather than UDP removes some scheduling flexibility, so I will just declare by fiat that the t2 and t3 timestamps are always equal when we "immediately" reply to a Notification. That is, there is "zero" latency to send such a reply. Not true, but close enough. A game running at 60 FPS should be able to manage sub-millisecond replies.

slightly fancier

Sending an app-level {serial, timestamp} Ack message is strictly overhead, which can slow down "real" messages that are trying to move or fire. If an Ack just barely fit into the congestion window, thereby closing it, that could delay a subsequent message, arriving on its heels, by an RTT latency. Modern TCPs are supposed to be able to immediately send four segments after an RTT duration of connection idle, but that's still finite. So we might prefer to piggyback each Ack, appending it to the end of "real" payloads. This would be a record boundary service offered by the channel layer, scheduling the delayed sending of certain data. Each outbound message might have zero or more Acks appended to it. The idea is to issue a single write(2) to the network socket, so all that data is efficiently placed in a single segment on the wire, totaling maybe a hundred or just a few hundred bytes.

Upon receiving a Notification when there's already IDK, three? pending Acks, we could just synchronously originate a Ping message or a Null message, which flushes those Acks back out to the server.

observer

The game server should retain some fixed number of recent Ack latency observations. How best to offer a summary report of them to the game application? Hard to say. You will want to chat with the folks consuming such data. Here are some possibilities.

immediate reports

Channel simply up-delivers each Ack to the application, and lets the app worry about how to interpret the firehose.

min filter

Channel maintains circular buffer of several recent Acks, so when a (synchronous) API call interrogates it, it can immediately report on an attribute of the physical network path: the minimum observed latency. This is the best latency we could reasonably hope for from any Responses that might be sent in the near future.

It is possible that such a summary figure might encompass observations from the past two seconds, or even the past twenty seconds. I don't see any motivation for increasing size of circular buffer when Acks are arriving "quickly" and shrinking it during a relative lull.

most recent

Maybe retaining just a single Ack suffices, as it might be a good estimator of the latency the very next Notification will experience. It takes into account cross-traffic congestion on core routers induced by other Internet users, and also locally induced congestion due to lots of large Notification messages going out plus Responses sent by the player. It will be a noisy metric.

smoothed filter

Game traffic tends to be bursty, so locally induced congestion is bursty. We could average the last eight measurements, but usually a smoothed average that weights recent measurements more heavily is more suitable.

Retaining two Acks lets us compute e.g. a .67 + .33 weighting.

With three we might opt for .50 + .30 + .20, or perhaps .50 + .25 + .25.

Given a bunch of measurements it's convenient to use binary weights of 1/2, 1/4, 1/8, 1/16 ..., finishing up with a pair of equal weights so they sum to unity. Or feed all historic measurements into an exponential decay in the way unix uptime does:

estimate /= 2.0
estimate += new_measurement / 2.0

In fact, any positive decay factor alpha, less than unity, would work. We’re not limited to just 0.5. Multiply estimate by alpha, and add in (1 - alpha) * new_measurement.

congestion

Both the cloud server end and the player's end are able to locally induce congestion, which will delay delivery of subsequent messages sends. Using tcpprobe from the server's end would allow discovery of two important dynamic TCP variables.

  • cwnd -- not very interesting, as congestion window changes with every packet
  • snd_ssthresh -- sending Slow Start threshhold depends on RTT and bottleneck bandwidth, and is quite stable over ten-second intervals

Learning that threshold value a few times a minute could be helpful. The game knows the RTT, and knows how many TCP bytes it has recently sent. So it could definitely know whether it's in danger of filling the congestion window so the connection will be "stuck" with writes being blocked for at least an RTT.

Absent such instrumentation, server might simply saturate the connection for one to three seconds by sending random incompressible data, to measure observed bandwidth.

There are low- and high- priority messages to be sent, for example a friendly vehicle lumbering along in the distant background versus nearby enemy zooming straight at you. And we often can choose to send low- or high- fidelity data, by adjusting poly count, graphics resolution, sending fewer updates per minute and so on.

A game server that knows whether there's idle bandwidth available to it or not, and that knows how to sensibly conserve or use that bandwidth, would offer a better UX to diverse players over diverse congestion conditions.

Another strategy is to maintain a secondary "mostly idle" WebSocket which is only used for short high-priority messages. This can accommodate 4 segments (maybe 5 KiB) per RTT. Causal relationship between the two connections then gets a bit tricky.


There's significant degrees of freedom available to the design.

In the end, consumers of the API will have to describe what they hope to get from it, more precisely than has happened up until now. Understanding the space of design options can inform a more productive Requirements Gathering phase.

\$\endgroup\$
2
  • 1
    \$\begingroup\$ it seems I've managed to mislead you with my previous updates: apologies. The main purpose of this utility is to prevent connection dropping by intermediary proxies and routers in case of temporary inactivity. RTT reporting is a "bonus" feature, but it is not my goal to provide versatile, bullet-proof solution for this. Furthermore my online gaming folks are a bit secretive and I couldn't get much info from them. I've pointed them to your answer as it contains a lot of valuable info. Regardless, I implemented several of the advices from your 1st answer in the newly added version: many thanks!! \$\endgroup\$
    – morgwai
    Feb 8 at 0:20
  • \$\begingroup\$ FYI: as per site rules, the new version was moved to a separate question: codereview.stackexchange.com/q/289386/249240 \$\endgroup\$
    – morgwai
    Feb 8 at 1:00

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.