WebSocket Implementation Based on Netty #91

hiranya911 · 2017-10-23T20:23:18Z

Java Admin SDK's RTDB implementation has been using a homegrown WebSocket implementation (the entire com.google.firebase.databas.tubesock package). I'm sure there were good reasons for doing it that way, but that code was pretty complicated and barely maintainable.

Starting from #83 (Firestore integration), we have a dependency on Netty, which is a popular library with excellent WebSocket support. This PR gets rid of our homegrown WebSocket code, and implements it using Netty's NioEventLoop. This results in a net removal of around 1600 lines of code. I believe the final outcome is much simpler and far more maintainable.

Unlike our old implementation which uses 2 threads (reader and writer) for WebSocket communication, Netty uses a single-threaded event loop. I have also tested this locally, and on GAE (java8) for several days straight without any problems.

…tore() methods.

…ase-admin-java into hkj-firestore-java

…Reader.java

hiranya911 · 2017-10-24T04:26:41Z

@mikelehen I simplified this PR as advised by removing some of the refactoring commits. It should be easier to review now. The PR states 23 files have been changed, but actually 15 of those files are total removals. Changes are localized to just 8 files.

The class that has changed most is WebsocketConnection. The others have fairly small changes.

schmidt-sebastian · 2017-10-26T16:49:43Z

src/main/java/com/google/firebase/database/core/Context.java

@@ -290,4 +284,4 @@ private String buildUserAgent(String platformAgent) {
            .append(platformAgent);
    return sb.toString();
  }
-}
+}


schmidt-sebastian · 2017-10-26T16:51:26Z

src/main/java/com/google/firebase/database/core/RepoManager.java

@@ -54,6 +55,8 @@ public static void destroy(Context ctx) {
      instance.destroyInternal(ctx);
    } finally {
      ctx.stop();
+      // TODO(hkj): https://github.com/netty/netty/issues/7310
+      FastThreadLocal.removeAll();


Can you verify that this still works with multiple database clients, one shutting down and the other one in continued use?

Tested with a multi-app set up. Seems to work fine. After the first app has been deleted, the 2nd app continues to receive updates. I'll continue to test this scenario a bit deeply.

schmidt-sebastian · 2017-10-26T16:51:59Z

src/main/java/com/google/firebase/database/core/RepoManager.java

@@ -54,6 +55,8 @@ public static void destroy(Context ctx) {
      instance.destroyInternal(ctx);
    } finally {
      ctx.stop();
+      // TODO(hkj): https://github.com/netty/netty/issues/7310


Since this issue was closed as intended, it might make sense to just summarize the problem here in a comment and remove the reference.

schmidt-sebastian · 2017-10-26T16:54:06Z

src/main/java/com/google/firebase/internal/RevivingScheduledExecutor.java

@@ -88,6 +90,7 @@ public RevivingScheduledExecutor(
      final long initialDelayMs,
      final long timeoutMs) {
    super(0);
+    checkNotNull(threadFactory, "thread factory must not be null");


100% Nit: "threadFactory must not be null" or "Thread factory must not be null" (applies to other places as well)

schmidt-sebastian · 2017-10-26T16:58:29Z

src/main/java/com/google/firebase/database/connection/NettyWebSocketClient.java

+ */
+class NettyWebSocketClient implements WebsocketConnection.WSClient {
+
+  private static final int DEFAULT_WSS_PORT = 443;


Can we use the port from the URI?

It turns out if the URI string is specified as https://test.firebaseio.com (which is what most users do), URI.getPort() returns -1.

schmidt-sebastian · 2017-10-26T17:23:26Z

src/main/java/com/google/firebase/database/connection/NettyWebSocketClient.java

+
+    @Override
+    public void channelRead0(ChannelHandlerContext context, Object message) throws Exception {
+      Channel channel = context.channel();


How about just doing:

try { if (message instanceof FullHttpResponse) { checkState(!handshaker.isHandshakeComplete()); handshaker.finishHandshake(channel, (FullHttpResponse) message); delegate.onOpen(); } else if (message instanceof TextWebSocketFrame) { delegate.onMessage(((TextWebSocketFrame) frame).text()); } else { checkState(message instanceof CloseWebSocketFrame); delegate.onClose(); } } catch (Exception e) { delegate.onError(e); }

Done. Thanks, looks much better now.

schmidt-sebastian · 2017-10-26T17:39:48Z

src/main/java/com/google/firebase/database/connection/WebsocketConnection.java

-        logger.debug("Tubesock: " + msg);
-      }
-    }
+  private static class StringList {


I thought for the longest time (up until this very line) that you were using an existing class. How much do we gain from having our own custom implementation, versus just doing it inline in the calling class?

And if you do decide to keep it, a small comment would be nice.

We were using a custom StringListReader class for this purpose, which was far more complex than it needed to be. I just created a simpler version of it. Added a class comment to explain its purpose. I do believe the util class makes the core a bit more readable.

schmidt-sebastian · 2017-10-26T17:40:13Z

src/main/java/com/google/firebase/database/connection/WebsocketConnection.java

    }

-    @Override
-    public void connect() {
+    String combine() {


This deservers a better name or at least a comment :) Thanks!

Added comment.

mikelehen

Few issues and misc nits, but overall I think it looks reasonable. I'm still nervous about this change, so I'd recommend any extra testing you can do (in particular trying to hit error cases / reconnects, etc. that won't be exercised by our automated tests)

mikelehen · 2017-10-26T16:41:32Z

src/main/java/com/google/firebase/database/connection/NettyWebSocketClient.java

+    checkState(channel == null, "channel already initialized");
+    try {
+      final SslContext sslContext = SslContextBuilder.forClient()
+          .trustManager(InsecureTrustManagerFactory.INSTANCE).build();


Is this intentional? I assume this means we're not validating certs which seems undesirable.

Used a proper trust manager. Thanks for pointing it out.

mikelehen · 2017-10-26T16:46:45Z

src/main/java/com/google/firebase/database/connection/NettyWebSocketClient.java

+            @Override
+            protected void initChannel(SocketChannel ch) {
+              ChannelPipeline p = ch.pipeline();
+              p.addLast(sslContext.newHandler(ch.alloc(), uri.getHost(), DEFAULT_WSS_PORT));


Why aren't we using the port from the uri instance? (here and below)

URI.getPort() returns -1 when the source URI string doesn't explicitly state the port.

I am fine with using DEFAULT_WSS_PORT in that case, but if you explicitly specify a port, this code should respect it. It may not matter much in practice, since prod firebase uses 443, but if somebody on the RTDB team wanted to test against a localhost instance or something, they might need to use a different port.

Fair enough. Changed to:

int port = uri.getPort() != -1 ? uri.getPort() : DEFAULT_WSS_PORT;

mikelehen · 2017-10-26T16:49:32Z

src/main/java/com/google/firebase/database/connection/NettyWebSocketClient.java

+              p.addLast(sslContext.newHandler(ch.alloc(), uri.getHost(), DEFAULT_WSS_PORT));
+              p.addLast(
+                  new HttpClientCodec(),
+                  new HttpObjectAggregator(8192),


Where did this 8192 come from? I can't remember what the maximum frame size we use is, but I thought it was at least 16k, and we should leave some buffer. (and please add a comment regarding the reasoning for the value chosen)

I believe this only applies to the original handshake request. Once the HTTP connection has been upgraded to a 2-way WebSocket connection, this no longer has any effect. Having said that I've bumped this to 32K and also added a comment.

mikelehen · 2017-10-26T16:51:46Z

src/main/java/com/google/firebase/database/connection/NettyWebSocketClient.java

+    } finally {
+      group.shutdownGracefully();
+      executorService.shutdown();
+      // TODO(hkj): https://github.com/netty/netty/issues/7310


This issue is closed. Can you remove or improve this TODO?

mikelehen · 2017-10-26T17:00:07Z

src/main/java/com/google/firebase/database/connection/NettyWebSocketClient.java

+    public void channelRead0(ChannelHandlerContext context, Object message) throws Exception {
+      Channel channel = context.channel();
+      if (!handshaker.isHandshakeComplete()) {
+        checkState(message instanceof FullHttpResponse);


This is running on the firebase-websocket-worker thread, right? What happens when we throw an exception? Does it get handled, logged, etc?

Tested by injecting some exceptions. It fires the NettyWebSocketClient.exceptionCaught() event, which we hand off to delegate.onError(). The net result is, the existing connection gets closed gracefully, and the client moves on to the next connection attempt.

If this check fails, that means there's a bug in this code and we should at minimum log it (and crashing the process wouldn't be out-of-the-question). So if this just silently retries without logging anything, then I don't think that's sufficient.

delegate.onError() logs the exception (delegate is WebsocketConnection.WSClientHandlerImpl in this case):

if (e.getCause() != null && e.getCause() instanceof EOFException) { logger.error("WebSocket reached EOF", e); } else { logger.error("WebSocket error", e); }

mikelehen · 2017-10-26T17:30:54Z

src/main/java/com/google/firebase/database/connection/WebsocketConnection.java

-                logger.debug("WebSocket reached EOF.");
-              } else {
-                logger.debug("WebSocket error.", e);
-              }
              onClosed();


I don't think NettyWebsocketConnection always closes the connection before calling onError, so I wonder if this should be closeAndNotify() instead?

That is what this method (onClosed()) does:

private void onClosed() { if (!isClosed) { if (logger.logsDebug()) { logger.debug("closing itself"); } closeAndNotify(); } }

mikelehen · 2017-10-26T17:32:09Z

src/main/java/com/google/firebase/database/connection/WebsocketConnection.java

-        logger.debug("Tubesock: " + msg);
-      }
-    }
+  private static class StringList {


Can you comment this a bit? It's a bit wonky since it's expecting a specific number of elements, unlike a normal list... You may want to make it a FrameBuffer or something to make it more obvious that it's special. But I don't feel strongly.

Added comments.

mikelehen · 2017-10-26T17:35:06Z

src/main/java/com/google/firebase/database/connection/WebsocketConnection.java

+    private final AtomicInteger remaining = new AtomicInteger(0);
+    private List<String> buffer;
+
+    void initialize(int capacity) {


Why not make this the constructor and just create a new instance for each websocket message? I don't see a strong reason for this to be a reusable class.

mikelehen · 2017-10-26T17:35:37Z

src/main/java/com/google/firebase/database/connection/WebsocketConnection.java

-    @Override
-    public void send(String msg) {
-      ws.send(msg);
+    private final AtomicInteger remaining = new AtomicInteger(0);


Why AtomicInteger? This class should only be accessed by a single thread, right? Using AtomicInteger suggests otherwise which makes the code less clear.

mikelehen · 2017-10-26T17:46:25Z

src/main/java/com/google/firebase/database/core/RepoManager.java

@@ -54,6 +55,8 @@ public static void destroy(Context ctx) {
      instance.destroyInternal(ctx);
    } finally {
      ctx.stop();
+      // TODO(hkj): https://github.com/netty/netty/issues/7310


Please make this actionable or remove it. (but if you remove the TODO but keep the removeAll() call, please add a comment explaining what it's for)

hiranya911 · 2017-10-26T23:29:24Z

Made the suggested changes. Over to you for the next round.

mikelehen

Left a couple more comments, but overall looks good. Thanks!

schmidt-sebastian · 2017-10-30T22:38:20Z

src/main/java/com/google/firebase/database/connection/WebsocketConnection.java

 class WebsocketConnection {

  private static final long KEEP_ALIVE_TIMEOUT_MS = 45 * 1000; // 45 seconds
  private static final long CONNECT_TIMEOUT_MS = 30 * 1000; // 30 seconds
  private static final int MAX_FRAME_SIZE = 16384;
-  private static long connectionId = 0;
+  private static final AtomicLong CONN_ID = new AtomicLong(0);


@mikelehen's comment regarding the suggested thread safety applies here as well.

I almost called this out too, but actually since this is static it can technically be accessed by multiple threads (each Repo has its own run loop with its own thread and so these global connection ids could be assigned simultaneously on different threads). So I think AtomicLong is applicable here (although a race is unlikely in practice and wouldn't be particularly harmful, since the ID is just used for logging).

rockwotj

Just poking around here, feel free to ignore.

rockwotj · 2017-11-03T06:47:38Z

src/main/java/com/google/firebase/database/connection/NettyWebSocketClient.java

+                  // Set the max size for the HTTP responses. This only applies to the WebSocket
+                  // handshake response from the server.
+                  new HttpObjectAggregator(32 * 1024),
+                  WebSocketClientCompressionHandler.INSTANCE,


We don't compress websocket frames and don't plan on it, so this probably isn't needed

Done. Thanks @rockwotj

hiranya911 added 27 commits August 31, 2017 16:35

Prototyping Firestore integration for Java

705abcd

Some code cleanup

084453e

Updating comments

39dec4c

Code cleanup

cd8030c

Merge branch 'master' into hkj-firestore-java

9be05c9

Removed getInstance() methods from FirestoreClient and added getFires…

dfc6bec

…tore() methods.

Merge branch 'hkj-firestore-java' of github.com:FirebasePrivate/fireb…

ef3881f

…ase-admin-java into hkj-firestore-java

Merged with master

0cf020b

Minor cleanup

ab3469a

Adding Firestore OAuth scopes

d64b4f6

Merge branch 'master' into hkj-firestore-java

d87f364

Some documentation updates

8ed4e5f

Reverted a doc change

48104da

Updating GCS dependency version (#89)

1771a01

Netty-based websocket impl

4201ff1

Refactored the Netty stuff to a separate util class

7e7f694

More code clean up and removing tubesock package

7a74047

Using the SDK ThreadManager for DB websocket threads

8e90610

More code cleanup and error checks

b910380

Handling connection drop events

a227cae

Handling connection loss events

6d8d90f

Adding documentation

e276f9a

Removed string list reader

c10753d

Merged with latest master

da73c79

Merge branch 'master' into hkj-firestore-java

193fcde

Merge branch 'hkj-firestore-java' into hkj-netty-rtdb

a6613ce

Removing redundant whitespace

59d1f5a

hiranya911 requested a review from mikelehen October 23, 2017 20:26

hiranya911 assigned mikelehen Oct 23, 2017

hiranya911 requested a review from schmidt-sebastian October 23, 2017 20:41

hiranya911 added 5 commits October 23, 2017 20:44

src/main/java/com/google/firebase/database/connection/util/StringList…

b87f536

…Reader.java

Added StringListReader back

4d7c74a

Re-updated websocket client

536eb61

reverting more syntactic changes

83c4e8b

Adding new line

06f5515

schmidt-sebastian reviewed Oct 26, 2017

View reviewed changes

mikelehen suggested changes Oct 26, 2017

View reviewed changes

hiranya911 assigned hiranya911 and unassigned schmidt-sebastian Oct 26, 2017

Added documentation; Using a more secure trust manager; Other cleanup

bc74249

hiranya911 assigned mikelehen and unassigned hiranya911 Oct 26, 2017

Removed unused import

8829759

mikelehen approved these changes Oct 26, 2017

View reviewed changes

Using the port on URI when available

a3411d6

hiranya911 assigned schmidt-sebastian and unassigned mikelehen Oct 27, 2017

schmidt-sebastian reviewed Oct 30, 2017

View reviewed changes

schmidt-sebastian approved these changes Oct 30, 2017

View reviewed changes

rockwotj reviewed Nov 3, 2017

View reviewed changes

hiranya911 assigned hiranya911 and unassigned schmidt-sebastian Nov 9, 2017

hiranya911 added 3 commits November 9, 2017 17:06

Merged with master

3adf4a7

Removing websocket commpression handler

884bd18

Declaring Netty dependencies explicitly (using the same version as gRPC)

d4a1438

hiranya911 changed the base branch from hkj-firestore-java to master November 22, 2017 00:45

hiranya911 merged commit 7187f05 into master Nov 22, 2017

hiranya911 deleted the hkj-netty-rtdb branch November 22, 2017 00:47

WebSocket Implementation Based on Netty #91

WebSocket Implementation Based on Netty #91

Uh oh!

Conversation

hiranya911 commented Oct 23, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hiranya911 commented Oct 24, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hiranya911 Oct 26, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mikelehen left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hiranya911 commented Oct 23, 2017 •

edited

Loading

hiranya911 Oct 26, 2017 •

edited

Loading