Use a cached threadpool for execution of connect. To avoid failing connect blocking other connects.
// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.jrt;
+import com.yahoo.concurrent.ThreadFactoryFactory;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
class Connector {
- private class Run implements Runnable {
- public void run() {
+ private final Transport parent;
+ private final ExecutorService executor = Executors.newCachedThreadPool(ThreadFactoryFactory.getDaemonThreadFactory("jrt.connector"));
+ private boolean done = false;
+
+ private void connect(Connection conn) {
try {
- Connector.this.run();
+ conn.transportThread().addConnection(conn.connect());
} catch (Throwable problem) {
parent.handleFailure(problem, Connector.this);
}
}
- }
-
- private Thread thread = new Thread(new Run(), "<jrt-connector>");
- private Transport parent;
- private ThreadQueue connectQueue = new ThreadQueue();
- private boolean done = false;
- private boolean exit = false;
public Connector(Transport parent) {
this.parent = parent;
- thread.setDaemon(true);
- thread.start();
}
- public void connectLater(Connection c) {
- if ( ! connectQueue.enqueue(c)) {
- c.transportThread().addConnection(c);
- }
+ public void connectLater(Connection conn) {
+ try {
+ executor.execute(() -> connect(conn));
+ } catch (RejectedExecutionException e) {
+ conn.transportThread().addConnection(conn);
}
- private void run() {
- try {
- while (true) {
- Connection conn = (Connection) connectQueue.dequeue();
- conn.transportThread().addConnection(conn.connect());
}
- } catch (EndOfQueueException e) {}
+
+ public Connector shutdown() {
+ executor.shutdown();
+ join();
synchronized (this) {
done = true;
notifyAll();
- while (!exit) {
- try { wait(); } catch (InterruptedException x) {}
}
- }
- }
-
- public Connector shutdown() {
- connectQueue.close();
return this;
}
@@ -60,7 +52,6 @@ class Connector {
}
public synchronized Connector exit() {
- exit = true;
notifyAll();
return this;
}
@@ -68,7 +59,7 @@ class Connector {
public void join() {
while (true) {
try {
- thread.join();
+ executor.awaitTermination(60, TimeUnit.SECONDS);
return;
} catch (InterruptedException e) {}
}
