Github user aarondav commented on a diff in the pull request:
https://github.com/apache/spark/pull/3625#discussion_r21413717
--- Diff:
network/common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java
---
@@ -143,43 +190,41 @@ public void initChannel(SocketChannel ch) {
assert client != null : "Channel future completed successfully with
null client";
// Execute any client bootstraps synchronously before marking the
Client as successful.
- long preBootstrap = System.currentTimeMillis();
+ long preBootstrap = System.nanoTime();
logger.debug("Connection to {} successful, running bootstraps...",
address);
try {
for (TransportClientBootstrap clientBootstrap : clientBootstraps) {
clientBootstrap.doBootstrap(client);
}
} catch (Exception e) { // catch non-RuntimeExceptions too as
bootstrap may be written in Scala
- long bootstrapTime = System.currentTimeMillis() - preBootstrap;
- logger.error("Exception while bootstrapping client after " +
bootstrapTime + " ms", e);
+ long bootstrapTimeMs = (System.nanoTime() - preBootstrap) / 1000000;
+ logger.error("Exception while bootstrapping client after " +
bootstrapTimeMs + " ms", e);
client.close();
throw Throwables.propagate(e);
}
- long postBootstrap = System.currentTimeMillis();
-
- // Successful connection & bootstrap -- in the event that two threads
raced to create a client,
- // use the first one that was put into the connectionPool and close
the one we made here.
- TransportClient oldClient = connectionPool.putIfAbsent(address,
client);
- if (oldClient == null) {
- logger.debug("Successfully created connection to {} after {} ms ({}
ms spent in bootstraps)",
- address, postBootstrap - preConnect, postBootstrap - preBootstrap);
- return client;
- } else {
- logger.debug("Two clients were created concurrently after {} ms,
second will be disposed.",
- postBootstrap - preConnect);
- client.close();
- return oldClient;
- }
+ long postBootstrap = System.nanoTime();
+
+ logger.debug("Successfully created connection to {} after {} ms ({} ms
spent in bootstraps)",
+ address, (postBootstrap - preConnect) / 1000000, (postBootstrap -
preBootstrap) / 1000000);
+
+ return client;
}
/** Close all connections in the connection pool, and shutdown the
worker thread pool. */
@Override
public void close() {
- for (TransportClient client : connectionPool.values()) {
- try {
- client.close();
- } catch (RuntimeException e) {
- logger.warn("Ignoring exception during close", e);
+ // Go through all clients and close them if they are active.
+ for (ClientPool clientPool : connectionPool.values()) {
+ for (int i = 0; i < clientPool.clients.length; i++) {
+ TransportClient client = clientPool.clients[i];
+ if (client != null) {
+ clientPool.clients[i] = null;
+ try {
+ client.close();
--- End diff --
Q: Can we use JavaUtils.closeQuietly(client) here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
