Create executor pool to take over network read/write tasks. #1227

zzmao · 2019-07-30T18:03:05Z

SSL transmission latency was high in recent performance test, especially in PUT.
The reason is selector.poll() needs to iterate all events and do SSL encryption + network IO for each of them. This increased round trip time dramatically in the queuing system.
The change introduced an executor pool to take over event handling and selector.poll() blocks until events done.
In the performance test, 7MB blobs were sent at 20QPS, PUT latency P95 was decreased to 380ms from 1.4kms with this change.

cgtz · 2019-07-30T20:14:00Z

ambry-network/src/main/java/com.github.ambry.network/Selector.java

@@ -556,16 +590,20 @@ private void read(SelectionKey key, Transmission transmission) throws IOExceptio
        transmission.onReceiveComplete();
        transmission.clearReceive();
      }
+    } catch (IOException e) {
+      handleReadWriteIOException(e, key);


Let's handle this from within the poll method since close modifies thread unsafe data structs

cgtz · 2019-07-30T20:14:53Z

ambry-network/src/main/java/com.github.ambry.network/Selector.java

@@ -556,16 +590,20 @@ private void read(SelectionKey key, Transmission transmission) throws IOExceptio
        transmission.onReceiveComplete();


let's have this method return the NetworkReceive if readComplete and then use the future result to modify completedReceives from the main thread.

jsjtzyy

Minor comments for now. Will keep reviewing when new code is added.

jsjtzyy · 2019-07-31T19:31:10Z

ambry-api/src/main/java/com.github.ambry/config/NetworkConfig.java

@@ -76,6 +76,13 @@
  @Default("false")
  public final boolean networkClientEnableConnectionReplenishment;

+  /**
+   * The size of the pool if selector executor pool. If the value is 0, executor pool won't be used.


minor, change to The size of the pool if selector executor pool is employed. When size is 0, executor pool won't be used

jsjtzyy · 2019-07-31T23:24:28Z

ambry-network/src/main/java/com.github.ambry.network/SSLTransmission.java

@@ -160,7 +160,9 @@ private int readFromSocketChannel() throws IOException {
  private boolean flush(ByteBuffer buf) throws IOException {
    int remaining = buf.remaining();
    if (remaining > 0) {
+      long start = SystemTime.getInstance().nanoseconds();


minor: consider using startTimeNs as you did in another method.

ambry-network/src/main/java/com.github.ambry.network/SSLTransmission.java

jsjtzyy · 2019-07-31T23:52:56Z

ambry-network/src/main/java/com.github.ambry.network/Selector.java

+          try {
+            future.get();
+          } catch (InterruptedException | ExecutionException e) {
+            logger.error("Read/Write key executor exception {}", e);


minor: no need for curly brackets here

cgtz · 2019-08-02T23:48:20Z

ambry-network/src/main/java/com.github.ambry.network/Selector.java

   *
   * @throws IOException If a send is given for which we have no existing connection or for which there is
   *         already an in-progress send
   */
  @Override
  public void poll(long timeoutMs, List<NetworkSend> sends) throws IOException {
+    if (executorPool != null) {
+      pollWithExecutorPool(timeoutMs, sends);
+      return;


minor: move the code below line 314 into another method like pollOnMainThread() and use an if-else to switch between them.

jsjtzyy · 2019-08-08T21:55:40Z

ambry-network/src/main/java/com.github.ambry.network/Selector.java

+      logger.error("Connection {} disconnected", socketDescription, e);
+    } else {
+      metrics.selectorIOErrorCount.inc();
+      logger.warn("Error in I/O with connection to {}", socketDescription, e);


why here we use warn instead of error level?

It was warn before. It may cause by network issue and is not critical to us.

jsjtzyy · 2019-08-08T22:01:10Z

ambry-network/src/main/java/com.github.ambry.network/Selector.java

+  /**
+   * Process read/write events on current thread.
+   */
+  public void pollOnMainThread(long timeoutMs, List<NetworkSend> sends) throws IOException {


can be private here

jsjtzyy · 2019-08-08T22:13:02Z

ambry-network/src/main/java/com.github.ambry.network/Selector.java

   */
-  private void read(SelectionKey key, Transmission transmission) throws IOException {
+  private NetworkReceive read(SelectionKey key, Transmission transmission) throws IOException {


both read and write methods can remove throws IOException since there is no exception thrown from these two methods.

jsjtzyy · 2019-08-08T22:22:01Z

ambry-network/src/main/java/com.github.ambry.network/Selector.java

+            }
+          }
+        } catch (InterruptedException | ExecutionException e) {
+          if (!(e.getCause() instanceof IOException)) {


Why not catch IOException and Exception separately?

copy logic from write future get()

jsjtzyy · 2019-08-08T22:24:42Z

ambry-network/src/test/java/com.github.ambry.network/SSLSelectorTest.java

@@ -307,6 +318,7 @@ private String blockingSSLConnect(int socketBufSize) throws IOException {
    String connectionId =
        selector.connect(new InetSocketAddress("localhost", server.port), socketBufSize, socketBufSize, PortType.SSL);
    while (!selector.connected().contains(connectionId)) {
+      System.out.println("here");


jsjtzyy · 2019-08-08T22:31:34Z

ambry-network/src/test/java/com.github.ambry.network/SSLSelectorTest.java

@@ -327,7 +339,7 @@ private void useCustomBufferSizeSelector(final Integer netReadBufSizeStart, fina
    selector.close();
    NetworkMetrics metrics = new NetworkMetrics(new MetricRegistry());
    Time time = SystemTime.getInstance();
-    selector = new Selector(metrics, time, clientSSLFactory) {
+    selector = new Selector(metrics, time, clientSSLFactory, 0) {


You can pass in different poolSize to this method.

jsjtzyy

LGTM once some minor comments are addressed.

codecov-io · 2019-08-13T21:23:46Z

Codecov Report

Merging #1227 into master will not change coverage.
The diff coverage is n/a.

@@            Coverage Diff            @@
##             master    #1227   +/-   ##
=========================================
  Coverage     83.62%   83.62%           
  Complexity       57       57           
=========================================
  Files             6        6           
  Lines           348      348           
  Branches         38       38           
=========================================
  Hits            291      291           
  Misses           45       45           
  Partials         12       12

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 89b82fb...6e60ab9. Read the comment docs.

zzmao requested review from cgtz and jsjtzyy July 30, 2019 18:03

cgtz reviewed Jul 30, 2019

View reviewed changes

jsjtzyy reviewed Aug 1, 2019

View reviewed changes

cgtz reviewed Aug 2, 2019

View reviewed changes

Introduce selector worker pool.

ad8dfc1

zzmao force-pushed the ssl branch from 538bd97 to ad8dfc1 Compare August 8, 2019 18:23

address commnet

e42cb83

jsjtzyy reviewed Aug 8, 2019

View reviewed changes

jsjtzyy approved these changes Aug 8, 2019

View reviewed changes

zzmao added 2 commits August 12, 2019 10:58

address yingyi comment

670eac4

address cgetz comment

6e60ab9

zzmao force-pushed the ssl branch from a4e25a4 to 6e60ab9 Compare August 13, 2019 21:09

cgtz merged commit db6d337 into linkedin:master Aug 13, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Create executor pool to take over network read/write tasks. #1227

Create executor pool to take over network read/write tasks. #1227

zzmao commented Jul 30, 2019

cgtz Jul 30, 2019

cgtz Jul 30, 2019

jsjtzyy left a comment

jsjtzyy Jul 31, 2019

jsjtzyy Jul 31, 2019

jsjtzyy Jul 31, 2019

cgtz Aug 2, 2019

jsjtzyy Aug 8, 2019

zzmao Aug 8, 2019

jsjtzyy Aug 8, 2019

zzmao Aug 12, 2019

jsjtzyy Aug 8, 2019

zzmao Aug 12, 2019

jsjtzyy Aug 8, 2019

cgtz Aug 13, 2019

zzmao Aug 13, 2019

jsjtzyy Aug 8, 2019

jsjtzyy Aug 8, 2019

jsjtzyy left a comment

codecov-io commented Aug 13, 2019

		@@ -556,16 +590,20 @@ private void read(SelectionKey key, Transmission transmission) throws IOExceptio
		transmission.onReceiveComplete();

Create executor pool to take over network read/write tasks. #1227

Create executor pool to take over network read/write tasks. #1227

Conversation

zzmao commented Jul 30, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsjtzyy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsjtzyy left a comment

Choose a reason for hiding this comment

codecov-io commented Aug 13, 2019

Codecov Report