Signal/Signal-Server
1
0
Fork 0
mirror of https://github.com/signalapp/Signal-Server synced 2026-04-20 09:47:58 +01:00
Code Issues Packages Projects Releases Wiki Activity

Revert Lettuce connection pooling.

Browse Source
This commit is contained in:
Jon Chambers
2020-08-18 15:55:01 -04:00
committed by Jon Chambers
parent 809750b995
commit 42ea7a9814
5 changed files with 22 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
service/src/main/java/org/whispersystems/textsecuregcm/redis/FaultTolerantRedisCluster.java
View File

@@ -10,14 +10,10 @@ import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection;
import io.lettuce.core.codec.ByteArrayCodec;
import io.lettuce.core.support.ConnectionPoolSupport;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.whispersystems.textsecuregcm.configuration.CircuitBreakerConfiguration;
import org.whispersystems.textsecuregcm.configuration.RedisClusterConfiguration;
import org.whispersystems.textsecuregcm.configuration.RedisConnectionPoolConfiguration;
import org.whispersystems.textsecuregcm.util.CircuitBreakerUtil;
import org.whispersystems.textsecuregcm.util.Constants;
@@ -32,7 +28,7 @@ import static com.codahale.metrics.MetricRegistry.name;
/**
* A fault-tolerant access manager for a Redis cluster. A fault-tolerant Redis cluster provides managed,
* circuit-breaker-protected access to a pool of connections.
* circuit-breaker-protected access to connections.
*/
public class FaultTolerantRedisCluster {
@@ -40,15 +36,14 @@ public class FaultTolerantRedisCluster {
private final RedisClusterClient clusterClient;
private final GenericObjectPool<StatefulRedisClusterConnection<String, String>> stringConnectionPool;
private final GenericObjectPool<StatefulRedisClusterConnection<byte[], byte[]>> binaryConnectionPool;
private final StatefulRedisClusterConnection<String, String> stringConnection;
private final StatefulRedisClusterConnection<byte[], byte[]> binaryConnection;
private final List<StatefulRedisClusterPubSubConnection<?, ?>> pubSubConnections = new ArrayList<>();
private final CircuitBreakerConfiguration circuitBreakerConfiguration;
private final CircuitBreaker circuitBreaker;
private final Timer acquireAndExecuteTimer;
private final Timer executeTimer;
private static final Logger log = LoggerFactory.getLogger(FaultTolerantRedisCluster.class);
@@ -57,27 +52,18 @@ public class FaultTolerantRedisCluster {
this(name,
RedisClusterClient.create(clusterConfiguration.getUrls().stream().map(RedisURI::create).collect(Collectors.toList())),
clusterConfiguration.getTimeout(),
clusterConfiguration.getCircuitBreakerConfiguration(),
clusterConfiguration.getConnectionPoolConfiguration());
clusterConfiguration.getCircuitBreakerConfiguration());
}
@VisibleForTesting
FaultTolerantRedisCluster(final String name, final RedisClusterClient clusterClient, final Duration timeout, final CircuitBreakerConfiguration circuitBreakerConfiguration, final RedisConnectionPoolConfiguration connectionPoolConfiguration) {
FaultTolerantRedisCluster(final String name, final RedisClusterClient clusterClient, final Duration timeout, final CircuitBreakerConfiguration circuitBreakerConfiguration) {
this.name = name;
this.clusterClient = clusterClient;
this.clusterClient.setDefaultTimeout(timeout);
@SuppressWarnings("rawtypes") final GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
poolConfig.setMaxIdle(connectionPoolConfiguration.getPoolSize());
poolConfig.setMaxTotal(connectionPoolConfiguration.getPoolSize());
poolConfig.setMaxWaitMillis(connectionPoolConfiguration.getMaxWait().toMillis());
//noinspection unchecked
this.stringConnectionPool = ConnectionPoolSupport.createGenericObjectPool(clusterClient::connect, poolConfig);
//noinspection unchecked
this.binaryConnectionPool = ConnectionPoolSupport.createGenericObjectPool(() -> clusterClient.connect(ByteArrayCodec.INSTANCE), poolConfig);
this.stringConnection = clusterClient.connect();
this.binaryConnection = clusterClient.connect(ByteArrayCodec.INSTANCE);
this.circuitBreakerConfiguration = circuitBreakerConfiguration;
this.circuitBreaker = CircuitBreaker.of(name + "-read", circuitBreakerConfiguration.toCircuitBreakerConfig());
@@ -88,13 +74,12 @@ public class FaultTolerantRedisCluster {
final MetricRegistry metricRegistry = SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME);
this.acquireAndExecuteTimer = metricRegistry.timer(name(getClass(), name, "acquireConnectionAndExecute"));
this.executeTimer = metricRegistry.timer(name(getClass(), name, "execute"));
this.executeTimer = metricRegistry.timer(name(getClass(), name, "execute"));
}
void shutdown() {
stringConnectionPool.close();
binaryConnectionPool.close();
stringConnection.close();
binaryConnection.close();
for (final StatefulRedisClusterPubSubConnection<?, ?> pubSubConnection : pubSubConnections) {
pubSubConnection.close();
@@ -104,30 +89,26 @@ public class FaultTolerantRedisCluster {
}
public void useCluster(final Consumer<StatefulRedisClusterConnection<String, String>> consumer) {
acceptPooledConnection(stringConnectionPool, consumer);
useConnection(stringConnection, consumer);
}
public <T> T withCluster(final Function<StatefulRedisClusterConnection<String, String>, T> function) {
return applyToPooledConnection(stringConnectionPool, function);
return withConnection(stringConnection, function);
}
public void useBinaryCluster(final Consumer<StatefulRedisClusterConnection<byte[], byte[]>> consumer) {
acceptPooledConnection(binaryConnectionPool, consumer);
useConnection(binaryConnection, consumer);
}
public <T> T withBinaryCluster(final Function<StatefulRedisClusterConnection<byte[], byte[]>, T> function) {
return applyToPooledConnection(binaryConnectionPool, function);
return withConnection(binaryConnection, function);
}
private <K, V> void acceptPooledConnection(final GenericObjectPool<StatefulRedisClusterConnection<K, V>> pool, final Consumer<StatefulRedisClusterConnection<K, V>> consumer) {
private <K, V> void useConnection(final StatefulRedisClusterConnection<K, V> connection, final Consumer<StatefulRedisClusterConnection<K, V>> consumer) {
try {
circuitBreaker.executeCheckedRunnable(() -> {
try (final Timer.Context ignored = acquireAndExecuteTimer.time();
final StatefulRedisClusterConnection<K, V> connection = pool.borrowObject()) {
try (final Timer.Context ignored2 = executeTimer.time()) {
consumer.accept(connection);
}
try (final Timer.Context ignored = executeTimer.time()) {
consumer.accept(connection);
}
});
} catch (final Throwable t) {
@@ -141,15 +122,11 @@ public class FaultTolerantRedisCluster {
}
}
private <T, K, V> T applyToPooledConnection(final GenericObjectPool<StatefulRedisClusterConnection<K, V>> pool, final Function<StatefulRedisClusterConnection<K, V>, T> function) {
private <T, K, V> T withConnection(final StatefulRedisClusterConnection<K, V> connection, final Function<StatefulRedisClusterConnection<K, V>, T> function) {
try {
return circuitBreaker.executeCheckedSupplier(() -> {
try (final Timer.Context ignored = acquireAndExecuteTimer.time();
final StatefulRedisClusterConnection<K, V> connection = pool.borrowObject()) {
try (final Timer.Context ignored2 = executeTimer.time()) {
return function.apply(connection);
}
try (final Timer.Context ignored = executeTimer.time()) {
return function.apply(connection);
}
});
} catch (final Throwable t) {