honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

simplify Connection class 

The connection class can be greatly simplified now that we don't ping anymore. Pings required a special initial state (UNKNOWN) for connections, to indicate that they require pinging although they are not dead. At this point we don't need the State enum anymore, as connections can only be dead or alive based on the number of failed attempts. markResurrected is also not needed, as it was again a way to make pings required. RestClient can simply pick a dead connection now and use it, no need to change its state when picking the connection.
This commit is contained in:
javanna 2016-05-13 12:19:45 +02:00 committed by Luca Cavanna
parent cdffc3d15b
commit 85a7721185
2 changed files with 15 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
client/src/main/java/org/elasticsearch/client/Connection.java
View File

@ -24,16 +24,13 @@ import org.apache.http.HttpHost;
import java.util.concurrent.TimeUnit;
/**
* Represents a connection to a host. It holds the host that the connection points to.
* Allows the transport to deal with very simple connection objects that are immutable.
* Any change to the state of a connection should be made through the connection pool.
* Represents a connection to a host. It holds the host that the connection points to and the state of the connection to it.
*/
public class Connection {
private static final long DEFAULT_CONNECTION_TIMEOUT_MILLIS = TimeUnit.MINUTES.toMillis(1);
private static final long MAX_CONNECTION_TIMEOUT_MILLIS = TimeUnit.MINUTES.toMillis(30);
private final HttpHost host;
private volatile State state = State.UNKNOWN;
private volatile int failedAttempts = -1;
private volatile int failedAttempts = 0;
private volatile long deadUntil = -1;
/**
@ -53,7 +50,7 @@ public class Connection {
/**
* Marks connection as dead. Should be called in case the corresponding node is not responding or caused failures.
* Once marked dead, the number of failed attempts will be incremented on each call to this method. A dead connection
* should be retried once {@link #shouldBeRetried()} returns true, which depends on the number of previous failed attempts
* should be retried once {@link #isBlacklisted()} returns true, which depends on the number of previous failed attempts
* and when the last failure was registered.
*/
void markDead() {
@ -63,64 +60,35 @@ public class Connection {
MAX_CONNECTION_TIMEOUT_MILLIS);
this.deadUntil = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMillis);
this.failedAttempts = ++failedAttempts;
this.state = State.DEAD;
}
}
/**
* Marks this connection alive. Should be called when the corresponding node is working properly.
* Will reset the number of failed attempts that were counted in case the connection was previously dead,
* as well as its dead timeout.
* Will reset the number of failed attempts that were counted in case the connection was previously dead, as well as its timeout.
*/
void markAlive() {
if (this.state != State.ALIVE) {
if (this.failedAttempts > 0) {
synchronized (this) {
this.deadUntil = -1;
this.failedAttempts = 0;
this.state = State.ALIVE;
}
}
}
/**
* Resets the connection to its initial state, so it will be retried. To be called when all the connections in the pool
* are dead, so that one connection can be retried. Note that calling this method only changes the state of the connection,
* it doesn't reset its failed attempts and dead until timestamp. That way if the connection goes back to dead straightaway
* all of its previous failed attempts are taken into account.
*/
void markResurrected() {
if (this.state == State.DEAD) {
synchronized (this) {
this.state = State.UNKNOWN;
}
}
}
/**
* Returns the timestamp till the connection is supposed to stay dead till it can be retried
* Returns the timestamp till the connection is supposed to stay dead. After that moment the connection should be retried
*/
public long getDeadUntil() {
return deadUntil;
}
/**
* Returns true if the connection is alive, false otherwise.
* Returns true when the connection should be skipped due to previous failures, false in case the connection is alive
* or dead but ready to be retried. When the connection is dead, returns false when it is time to retry it, depending
* on how many failed attempts were registered and when the last failure happened (minimum 1 minute, maximum 30 minutes).
*/
public boolean isAlive() {
return state == State.ALIVE;
}
/**
* Returns true in case the connection is not alive but should be used/retried, false otherwise.
* Returns true in case the connection is in unknown state (never used before) or resurrected. When the connection is dead,
* returns true when it is time to retry it, depending on how many failed attempts were registered and when the last failure
* happened (minimum 1 minute, maximum 30 minutes).
*/
public boolean shouldBeRetried() {
return state == State.UNKNOWN || (state == State.DEAD && System.nanoTime() - deadUntil >= 0);
}
private enum State {
UNKNOWN, DEAD, ALIVE
public boolean isBlacklisted() {
return failedAttempts > 0 && System.nanoTime() - deadUntil < 0;
}
}

10
client/src/main/java/org/elasticsearch/client/RestClient.java
View File

@ -148,7 +148,7 @@ public final class RestClient implements Closeable {
*/
private Iterator<Connection> nextConnection() {
if (this.connections.isEmpty()) {
throw new IllegalStateException("no connections available in the connection pool");
throw new IllegalStateException("no connections available");
}
List<Connection> rotatedConnections = new ArrayList<>(connections);
@ -157,7 +157,7 @@ public final class RestClient implements Closeable {
Iterator<Connection> connectionIterator = rotatedConnections.iterator();
while (connectionIterator.hasNext()) {
Connection connection = connectionIterator.next();
if (connection.isAlive() == false && connection.shouldBeRetried() == false) {
if (connection.isBlacklisted()) {
connectionIterator.remove();
}
}
@ -170,8 +170,7 @@ public final class RestClient implements Closeable {
}
});
Connection connection = sortedConnections.get(0);
connection.markResurrected();
logger.trace("marked connection resurrected for " + connection.getHost());
logger.trace("trying to resurrect connection for " + connection.getHost());
return Collections.singleton(connection).iterator();
}
return rotatedConnections.iterator();
@ -307,8 +306,7 @@ public final class RestClient implements Closeable {
}
/**
* Sets the hosts that the client will send requests to. Mandatory if no connection pool is specified,
* as the provided hosts will be used to create the default static connection pool.
* Sets the hosts that the client will send requests to.
*/
public Builder setHosts(HttpHost... hosts) {
if (hosts == null || hosts.length == 0) {