Use Server-sent events to add live updates for files, players, and markers#819
Use Server-sent events to add live updates for files, players, and markers#819pR0Ps wants to merge 5 commits into
Conversation
…rkers SSE Implementation: - A `SseConnectionManager` is created per-map in the `MapRequestHandler`. - Whenever a client connects to the `live/sse` endpoint, a new `SseConnection` is created for it and is added to the `SseConnectionManager`. - The `SseConnectionManager` receives requests to push updates and sends them to all known `SseConnections`. If sending fails, the connection is cleaned up and removed. - Sending happens on a background thread so normal server operations are not blocked on clients picking up events. Backend events: - Tile updates are event-based: when a new tile is rendered an event is emitted which causes the tile's X+Y and lod to be broadcasted to all connected clients. - Players and other markers are still polling-based, but the polling is done server-side (and only once per endpoint) and connected clients are notified only when the data changes. - Polling was implemented by replacing the `CachedRateLimitDataSupplier` with a `LiveDataSupplierBroadcaster`. This `Supplier` wrapper can be used in the same way where requests to `get()` it's data are rate-limited and cached. However, it also runs a background thread that (when listeners are added) will automatically poll the supplier and send events when the data changes. Frontend changes: - On page load, an `EventSource` connects to the `<map>/live/sse` endpoint and watches for events. - "tile" updates cause the specified tiles to be force-reloaded - "player" updates are fed directly into the existing `playerMarkerManager` - "marker" updates are fed directly into the existing `markerFileManager` - Since this takes the place of the marker managers polling, the marker managers were updated to be able to be paused, and are started in a paused state. - In order to ensure that markers are still updated if the SSE endpoint can't be contacted, the existing polling marker managers will be unpaused if the SSE connection fails to connect or disconnects. If the SSE connection returns, the marker managers will be paused again. - Tile force-reloading is handled by removing the specified tile from the `revalidatedUrls` cache, the calling `load` on it. This forces the normal tile loading mechanism to update the tile, reloading it.
| public LiveDataSupplierBroadcaster(Supplier<T> dataSupplier, long pollIntervalMillis) { | ||
| this.dataSupplier = dataSupplier; | ||
| this.pollIntervalMillis = pollIntervalMillis; | ||
| this.pollThread = new Thread(this::pollLoop, String.format("%sPoller", dataSupplier.getClass().getName())); |
There was a problem hiding this comment.
This will create 2 Platform-Threads for each map that is loaded. Some modded servers have a lot of dimensions and thus a lot of maps, which results in a ton of threads.
Since Platform-Threads are not free, I would love to avoid that somehow.
Maybe we can make use of the BlueMap.SCHEDULER here instead?
There was a problem hiding this comment.
Good call, I updated this file to just use BlueMap.SCHEDULER.scheduleWithFixedDelay(...) to fire the updates, and wrapped the actual call it fires with BlueMap.THREAD_POOL.execute(...) so that a slow data supplier doesn't block anything else the scheduler is working on.
| /** | ||
| * Queues an SSE event to be sent to all live connections on the background broadcast thread. | ||
| * Returns immediately without blocking. | ||
| * | ||
| * @param eventType the SSE event type | ||
| * @param data the event data payload | ||
| */ | ||
| public void broadcast(String eventType, String data) { | ||
| if (closed) return; | ||
| broadcastQueue.offer(new String[]{eventType, data}); | ||
| } | ||
|
|
||
| private void broadcastLoop() { | ||
| try { | ||
| while (!closed) { | ||
| String[] event = broadcastQueue.take(); | ||
| broadcastSync(event[0], event[1]); | ||
| } | ||
| } catch (InterruptedException ignored) {} | ||
| } | ||
|
|
||
| /** | ||
| * Sends an SSE event to all live connections synchronously. | ||
| * Dead or broken connections are removed automatically. | ||
| */ | ||
| private void broadcastSync(String eventType, String data) { | ||
| if (connections.isEmpty()) return; | ||
| List<SseConnection> toRemove = new ArrayList<>(); | ||
| for (SseConnection conn : connections) { | ||
| try { | ||
| conn.send(eventType, data); | ||
| } catch (IOException ignored) {} | ||
| if (conn.isClosed()) { | ||
| toRemove.add(conn); | ||
| } | ||
| } | ||
| if (!toRemove.isEmpty()) { | ||
| boolean fire; | ||
| synchronized (this) { | ||
| fire = connections.removeAll(toRemove) && connections.isEmpty(); | ||
| } | ||
| if (fire) notifyHasConnections(false); | ||
| } | ||
| } |
There was a problem hiding this comment.
One doubt i have here is that if one connection is not responsive or very slow, this would probably slow down the update-process for all connections?
Also i would like to have some safe-guard against the broadcastQueue growing indefinitely if somehow more events come in than can can be processed/sent :)
Does anything speak against creating a virtual-thread per SseConnection to send the events?
And then maybe each connection could have their own broadcastQueue that can be size-limited.
(Just thinking out loud right now)
There was a problem hiding this comment.
Hmm, yes, a slow client would currently slow down delivery to other clients: broadcastSync sends the events to each connection viaSseConnection.send() which blocks once its internal pipe's buffer is filled up. Unblocking that requires the HttpResponseOutputStream to pull the data out of it and write it to the connected client. If the client reads extremely slowly, both the output buffer, as well as the SseConnection's pipe's buffer would fill up, blocking SseConnection.send(), blocking broadcastSync from moving on to other connections.
I think you're right about the fix:
Since each SseConnection is already running in its own virtual thread (they're spawned from HttpServer.handleConnection()), I should be able to move the queue + send logic into the SseConnection itself. That way the SseConnectionManager can broadcast without blocking by just firing the events into each SseConnection's queue and letting it handle draining them to the client. That would also make it easy to implement a per-connection max queue size.
There was a problem hiding this comment.
I've addressed these issues in 6253d80, mostly in the way I stated above.
This is an Nginx-specific header, but may be supported by other reverse proxies as well. At the very lest, it can't hurt?
The polling of data at a fixed interval is now scheduled by the BlueMap.SCHEDULER and the work of doing the updates is run in the BlueMap.THREAD_POOL. This makes the LiveDataSupplierBroadcaster only responsible for starting/stopping the scheduled task when listeners connect/disconnect. This avoids the overhead of having to spin up a polling thread for every supplier * every map, as well is a much simpler implementation.
Previously a single slow consumer of events would block all other consumers. This was because the `SseConnectionManager` would handle sending data to each connection in a single worker thread. This commit refactors the code so that the `SseConnectionManager` is no longer responsible for actually sending the data to each connection. Instead, it simply enqueues events to every managed `SseConnection` which handles sending them. Details: - Each `SseConnection` now has a queue and a virtual thread that pulls events off it and sends them to the consumer. This means that a slow consumer now only blocks its own connection. - The per-connection queues now limit how many events can be queued to send. After the limit is reached, any new events are simply dropped to avoid slow clients forcing the server to buffer an unlimited number of old events. - An `onClose` callback can now be registered on the `SseConnection` class. This allows it to own its entire lifecycle, while still allowing the `SseConnectionManager` to be notified when it's been closed and needs removing from the set of managed connections.
2 participants
SSE Implementation:
SseConnectionManageris created per-map in theMapRequestHandler.live/sseendpoint, a newSseConnectionis created for it and is added to theSseConnectionManager.SseConnectionManagerreceives requests to push updates and sends them to all knownSseConnections. If sending fails, the connection is cleaned up and removed.Backend events:
CachedRateLimitDataSupplierwith aLiveDataSupplierBroadcaster. ThisSupplierwrapper can be used in the same way where requests toget()it's data are rate-limited and cached. However, it also runs a background thread that (when listeners are added) will automatically poll the supplier and send events when the data changes.Frontend changes:
EventSourceconnects to the<map>/live/sseendpoint and watches for events.playerMarkerManagermarkerFileManagerrevalidatedUrlscache, the callingloadon it. This forces the normal tile loading mechanism to update the tile, reloading it.