Skip to content

refactor to allow for a connection pool #113

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

refactor to allow for a connection pool #113

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/Shared/ProviderConfiguration.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ internal class ProviderConfiguration
public int OperationTimeoutInMilliSec { get; set; }
public string ConnectionString { get; set; }
public string RedisSerializerType { get; set; }
public int ConnectionPoolSize { get; set; }

/* Empty constructor required for testing */
internal ProviderConfiguration()
Expand All @@ -41,7 +42,6 @@ internal static ProviderConfiguration ProviderConfigurationForSessionState(NameV
configuration.ThrowOnError = GetBoolSettings(config, "throwOnError", true);
int retryTimeoutInMilliSec = GetIntSettings(config, "retryTimeoutInMilliseconds", 5000);
configuration.RetryTimeout = new TimeSpan(0, 0, 0, 0, retryTimeoutInMilliSec);

// Get request timeout from config
HttpRuntimeSection httpRuntimeSection = ConfigurationManager.GetSection("system.web/httpRuntime") as HttpRuntimeSection;
configuration.RequestTimeout = httpRuntimeSection.ExecutionTimeout;
Expand All @@ -61,7 +61,6 @@ internal static ProviderConfiguration ProviderConfigurationForOutputCache(NameVa

// No retry login for output cache provider
configuration.RetryTimeout = TimeSpan.Zero;

// Session state specific attribute which are not applicable to output cache
configuration.ThrowOnError = true;
configuration.RequestTimeout = TimeSpan.Zero;
Expand All @@ -77,6 +76,7 @@ private ProviderConfiguration(NameValueCollection config)
EnableLoggingIfParametersAvailable(config);
// Get connection host, port and password.
// host, port, accessKey and ssl are firest fetched from appSettings if not found there than taken from web.config
ConnectionPoolSize = GetIntSettings(config, "ConnectionPoolSize", 1);
ConnectionString = GetConnectionString(config);
Host = GetStringSettings(config, "host", "127.0.0.1");
Port = GetIntSettings(config, "port", 0);
Expand Down
50 changes: 35 additions & 15 deletions src/Shared/RedisSharedConnection.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
//

using System;
using System.Collections.Concurrent;
using StackExchange.Redis;

namespace Microsoft.Web.Redis
Expand All @@ -12,14 +13,15 @@ internal class RedisSharedConnection
{
private ProviderConfiguration _configuration;
private ConfigurationOptions _configOption;
private Lazy<ConnectionMultiplexer> _redisMultiplexer;
private ConcurrentQueue<Lazy<ConnectionMultiplexer>> _redisMultiplexer;

internal static DateTimeOffset lastReconnectTime = DateTimeOffset.MinValue;
internal static DateTimeOffset firstErrorTime = DateTimeOffset.MinValue;
internal static DateTimeOffset previousErrorTime = DateTimeOffset.MinValue;
static object reconnectLock = new object();
internal static TimeSpan ReconnectFrequency = TimeSpan.FromSeconds(60);
internal static TimeSpan ReconnectErrorThreshold = TimeSpan.FromSeconds(30);
private int _poolSize;

// Used for mocking in testing
internal RedisSharedConnection()
Expand All @@ -28,7 +30,8 @@ internal RedisSharedConnection()
public RedisSharedConnection(ProviderConfiguration configuration)
{
_configuration = configuration;

_redisMultiplexer = new ConcurrentQueue<Lazy<ConnectionMultiplexer>>();
_poolSize = configuration.ConnectionPoolSize;
// If connection string is given then use it otherwise use individual options
if (!string.IsNullOrEmpty(configuration.ConnectionString))
{
Expand Down Expand Up @@ -68,7 +71,14 @@ public RedisSharedConnection(ProviderConfiguration configuration)

public IDatabase Connection
{
get { return _redisMultiplexer.Value.GetDatabase(_configOption.DefaultDatabase ?? _configuration.DatabaseId); }
get
{
//return _redisMultiplexer.Value.GetDatabase(_configOption.DefaultDatabase ?? _configuration.DatabaseId);
Lazy<ConnectionMultiplexer> multiPlexorLazy = null;
while (!_redisMultiplexer.TryDequeue(out multiPlexorLazy)) continue;
_redisMultiplexer.Enqueue(multiPlexorLazy);
return multiPlexorLazy.Value.GetDatabase(_configOption.DefaultDatabase ?? _configuration.DatabaseId);
}
}

public void ForceReconnect()
Expand Down Expand Up @@ -110,8 +120,7 @@ public void ForceReconnect()
firstErrorTime = DateTimeOffset.MinValue;
previousErrorTime = DateTimeOffset.MinValue;

var oldMultiplexer = _redisMultiplexer;
CloseMultiplexer(oldMultiplexer);
CloseMultiplexer();
CreateMultiplexer();
}
}
Expand All @@ -122,26 +131,37 @@ private void CreateMultiplexer()
{
if (LogUtility.logger == null)
{
_redisMultiplexer = new Lazy<ConnectionMultiplexer>(() => ConnectionMultiplexer.Connect(_configOption));
for (int i = 0; i < _poolSize; i++)
{
_redisMultiplexer.Enqueue(new Lazy<ConnectionMultiplexer>(() => ConnectionMultiplexer.Connect(_configOption)));
}
}
else
{
_redisMultiplexer = new Lazy<ConnectionMultiplexer>(() => ConnectionMultiplexer.Connect(_configOption, LogUtility.logger));
for (int i = 0; i < _poolSize; i++)
{
_redisMultiplexer.Enqueue(new Lazy<ConnectionMultiplexer>(() => ConnectionMultiplexer.Connect(_configOption,LogUtility.logger)));
}
}
lastReconnectTime = DateTimeOffset.UtcNow;
}

private void CloseMultiplexer(Lazy<ConnectionMultiplexer> oldMultiplexer)

private void CloseMultiplexer()
{
if (oldMultiplexer.Value != null)
Lazy<ConnectionMultiplexer> oldMultiplexer = null;
while (!_redisMultiplexer.TryDequeue(out oldMultiplexer))
{
try
if (oldMultiplexer.Value != null)
{
oldMultiplexer.Value.Close();
}
catch (Exception)
{
// Example error condition: if accessing old.Value causes a connection attempt and that fails.
try
{
oldMultiplexer.Value.Close();
}
catch (Exception)
{
// Example error condition: if accessing old.Value causes a connection attempt and that fails.
}
}
}
}
Expand Down
1 change: 1 addition & 0 deletions test/Shared/Utility.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ internal static ProviderConfiguration GetDefaultConfigUtility()
configuration.RequestTimeout = new TimeSpan(0, 1, 30); //1.5 min
configuration.Host = "127.0.0.1";
configuration.Port = 0;
configuration.ConnectionPoolSize = 1;
configuration.AccessKey = null;
configuration.UseSsl = false;
configuration.DatabaseId = 0;
Expand Down