Skip to content

S
StackExchange.Redis
Commit f343388c authored by Marc Gravell's avatar Marc Gravell
Browse files
Options

moved all socket-server code to Pipelines.Sockets.Unofficial

parent aaa6afa0
Hide whitespace changes
Inline Side-by-side
Showing with 66 additions and 84 deletions
KestrelRedisServer/Startup.cs
View file @ f343388c
......@@ -19,8 +19,20 @@ public void ConfigureServices(IServiceCollection services)
public void Dispose() => _server.Dispose();
// This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
public void Configure(IApplicationBuilder app, IHostingEnvironment env)
public void Configure(IApplicationBuilder app, IHostingEnvironment env, IApplicationLifetime lifetime)
{
_server.Shutdown.ContinueWith((t, s) =>
{
try
{ // if the resp server is shutdown by a client: stop the kestrel server too
if (t.Result == RespServer.ShutdownReason.ClientInitiated)
{
((IApplicationLifetime)s).StopApplication();
}
}
catch { }
}, lifetime);
if (env.IsDevelopment()) app.UseDeveloperExceptionPage();
app.Run(context => context.Response.WriteAsync(_server.GetStats()));
}
......
StackExchange.Redis.Server/RedisServer.cs
View file @ f343388c
......@@ -277,7 +277,7 @@ protected virtual RedisResult Set(RedisClient client, RedisRequest request)
[RedisCommand(1)]
protected new virtual RedisResult Shutdown(RedisClient client, RedisRequest request)
{
DoShutdown();
DoShutdown(ShutdownReason.ClientInitiated);
return RedisResult.OK;
}
[RedisCommand(2)]
......@@ -378,8 +378,8 @@ StringBuilder AddHeader()
{
sb.Append("process:").Append(process.Id).AppendLine();
}
var port = TcpPort();
if (port >= 0) sb.Append("tcp_port:").Append(port).AppendLine();
//var port = TcpPort();
//if (port >= 0) sb.Append("tcp_port:").Append(port).AppendLine();
break;
case "Clients":
AddHeader().Append("connected_clients:").Append(ClientCount).AppendLine();
......
StackExchange.Redis.Server/RespServer.cs
View file @ f343388c
......@@ -5,8 +5,6 @@
using System.IO;
using System.IO.Pipelines;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Reflection;
using System.Text;
using System.Threading;
......@@ -17,9 +15,14 @@ namespace StackExchange.Redis.Server
{
public abstract partial class RespServer : IDisposable
{
public enum ShutdownReason
{
ServerDisposed,
ClientInitiated,
}
private readonly List<RedisClient> _clients = new List<RedisClient>();
private readonly TextWriter _output;
private Socket _listener;
public RespServer(TextWriter output = null)
{
_output = output;
......@@ -176,43 +179,7 @@ internal int NetArity()
}
}
delegate RedisResult RespOperation(RedisClient client, RedisRequest request);
protected int TcpPort()
{
var ep = _listener?.LocalEndPoint;
if (ep is IPEndPoint ip) return ip.Port;
if (ep is DnsEndPoint dns) return dns.Port;
return -1;
}
private Action<object> _runClientCallback;
// KeepAlive here just to make the compiler happy that we've done *something* with the task
private Action<object> RunClientCallback => _runClientCallback ??
(_runClientCallback = state => GC.KeepAlive(RunClientAsync((IDuplexPipe)state)));
public void Listen(
EndPoint endpoint,
AddressFamily addressFamily = AddressFamily.InterNetwork,
SocketType socketType = SocketType.Stream,
ProtocolType protocolType = ProtocolType.Tcp,
PipeOptions sendOptions = null, PipeOptions receiveOptions = null)
{
Socket listener = new Socket(addressFamily, socketType, protocolType);
listener.Bind(endpoint);
listener.Listen(20);
_listener = listener;
StartOnScheduler(receiveOptions?.ReaderScheduler, _ => ListenForConnections(
sendOptions ?? PipeOptions.Default, receiveOptions ?? PipeOptions.Default), null);
Log("Server is listening on " + Format.ToString(endpoint));
}
private static void StartOnScheduler(PipeScheduler scheduler, Action<object> callback, object state)
{
if (scheduler == PipeScheduler.Inline) scheduler = null;
(scheduler ?? PipeScheduler.ThreadPool).Schedule(callback, state);
}
// for extensibility, so that a subclass can get their own client type
// to be used via ListenForConnections
public virtual RedisClient CreateClient() => new RedisClient();
......@@ -244,33 +211,14 @@ public bool RemoveClient(RedisClient client)
return _clients.Remove(client);
}
}
private async void ListenForConnections(PipeOptions sendOptions, PipeOptions receiveOptions)
{
try
{
while (true)
{
var client = await _listener.AcceptAsync();
SocketConnection.SetRecommendedServerOptions(client);
var pipe = SocketConnection.Create(client, sendOptions, receiveOptions);
StartOnScheduler(receiveOptions.ReaderScheduler, RunClientCallback, pipe);
}
}
catch (NullReferenceException) { }
catch (ObjectDisposedException) { }
catch (Exception ex)
{
if (!_isShutdown) Log("Listener faulted: " + ex.Message);
}
}
private readonly TaskCompletionSource<int> _shutdown = new TaskCompletionSource<int>();
private readonly TaskCompletionSource<ShutdownReason> _shutdown = new TaskCompletionSource<ShutdownReason>(TaskCreationOptions.RunContinuationsAsynchronously);
private bool _isShutdown;
protected void ThrowIfShutdown()
{
if (_isShutdown) throw new InvalidOperationException("The server is shutting down");
}
protected void DoShutdown(PipeScheduler scheduler = null)
protected void DoShutdown(ShutdownReason reason, PipeScheduler scheduler = null)
{
if (_isShutdown) return;
Log("Server shutting down...");
......@@ -280,19 +228,13 @@ protected void DoShutdown(PipeScheduler scheduler = null)
foreach (var client in _clients) client.Dispose();
_clients.Clear();
}
StartOnScheduler(scheduler,
state => ((TaskCompletionSource<int>)state).TrySetResult(0), _shutdown);
_shutdown.TrySetResult(reason);
}
public Task Shutdown => _shutdown.Task;
public Task<ShutdownReason> Shutdown => _shutdown.Task;
public void Dispose() => Dispose(true);
protected virtual void Dispose(bool disposing)
{
DoShutdown();
var socket = _listener;
if (socket != null)
{
try { socket.Dispose(); } catch { }
}
DoShutdown(ShutdownReason.ServerDisposed);
}
public async Task RunClientAsync(IDuplexPipe pipe)
......@@ -335,7 +277,7 @@ public async Task RunClientAsync(IDuplexPipe pipe)
}
}
}
private void Log(string message)
public void Log(string message)
{
var output = _output;
if (output != null)
......
StackExchange.Redis.Server/RespSocketServer.cs 0 → 100644
View file @ f343388c
using System;
using System.Net;
using System.Threading.Tasks;
using Pipelines.Sockets.Unofficial;
namespace StackExchange.Redis.Server
{
public sealed class RespSocketServer : SocketServer
{
private readonly RespServer _server;
public RespSocketServer(RespServer server)
{
_server = server ?? throw new ArgumentNullException(nameof(server));
server.Shutdown.ContinueWith((t, o) => ((SocketServer)o).Dispose(), this);
}
protected override void OnStarted(EndPoint endPoint)
=> _server.Log("Server is listening on " + endPoint);
protected override Task OnClientConnectedAsync(in ClientConnection client)
=> _server.RunClientAsync(client.Transport);
protected override void Dispose(bool disposing)
{
if (disposing) _server.Dispose();
}
}
}
StackExchange.Redis/StackExchange.Redis.csproj
View file @ f343388c
......@@ -23,7 +23,7 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="Pipelines.Sockets.Unofficial" Version="0.2.1-alpha.68" />
<PackageReference Include="Pipelines.Sockets.Unofficial" Version="0.2.1-alpha.72" />
<PackageReference Include="System.Threading.Channels" Version="4.5.0" />
</ItemGroup>
</Project>
\ No newline at end of file
StackExchange.Redis/StackExchange/Redis/RedisResult.cs
View file @ f343388c
......@@ -35,12 +35,12 @@ public static RedisResult Create(RedisResult[] values)
/// <summary>
/// An empty array result
/// </summary>
public static RedisResult EmptyArray { get; } = new ArrayRedisResult(Array.Empty<RedisResult>());
internal static RedisResult EmptyArray { get; } = new ArrayRedisResult(Array.Empty<RedisResult>());
/// <summary>
/// A null array result
/// </summary>
public static RedisResult NullArray { get; } = new ArrayRedisResult(null);
internal static RedisResult NullArray { get; } = new ArrayRedisResult(null);
// internally, this is very similar to RawResult, except it is designed to be usable
// outside of the IO-processing pipeline: the buffers are standalone, etc
......@@ -96,16 +96,16 @@ internal static RedisResult TryCreate(PhysicalConnection connection, RawResult r
/// <summary>
/// An integer-zero result
/// </summary>
public static RedisResult Zero { get; } = Create(0, ResultType.Integer);
internal static RedisResult Zero { get; } = Create(0, ResultType.Integer);
/// <summary>
/// An integer-one result
/// </summary>
public static RedisResult One { get; } = Create(1, ResultType.Integer);
internal static RedisResult One { get; } = Create(1, ResultType.Integer);
/// <summary>
/// A null bulk-string result
/// </summary>
public static RedisResult Null { get; } = Create(RedisValue.Null, ResultType.BulkString);
internal static RedisResult Null { get; } = Create(RedisValue.Null, ResultType.BulkString);
/// <summary>
/// Interprets the result as a <see cref="string"/>.
......
TestConsole/Program.cs
View file @ f343388c
......@@ -7,10 +7,11 @@ static class Program
{
static async Task Main()
{
using (var server = new MemoryCacheRedisServer(Console.Out))
using (var resp = new MemoryCacheRedisServer(Console.Out))
using (var socket = new RespSocketServer(resp))
{
server.Listen(new IPEndPoint(IPAddress.Loopback, 6378));
await server.Shutdown;
socket.Listen(new IPEndPoint(IPAddress.Loopback, 6378));
await resp.Shutdown;
}
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment