huahaofeng
/
aspnetcore
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request #145 from aspnet/lodejard/multiloop 

Multiple threads processing connections
This commit is contained in:
Louis DeJardin 2015-07-27 11:49:08 -07:00
parent a457f6da25 5e6e5fec01
commit 166ec72484
26 changed files with 850 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/Kestrel/IKestrelServerInformation.cs Normal file
View File

@ -0,0 +1,10 @@
// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
namespace Kestrel
{
public interface IKestrelServerInformation
{
int ThreadCount { get; set; }
}
}

2
src/Kestrel/ServerFactory.cs
View File

@ -38,7 +38,7 @@ namespace Kestrel
var disposables = new List<IDisposable>();
var information = (ServerInformation)serverInformation;
var engine = new KestrelEngine(_libraryManager, _appShutdownService);
engine.Start(1);
engine.Start(information.ThreadCount == 0 ? 1 : information.ThreadCount);
foreach (var address in information.Addresses)
{
disposables.Add(engine.CreateServer(

6
src/Kestrel/ServerInformation.cs
View File

@ -8,7 +8,7 @@ using Microsoft.Framework.Configuration;
namespace Kestrel
{
public class ServerInformation : IServerInformation
public class ServerInformation : IServerInformation, IKestrelServerInformation
{
public ServerInformation()
{
@ -25,7 +25,7 @@ namespace Kestrel
foreach (var url in urls.Split(new[] { ';' }, StringSplitOptions.RemoveEmptyEntries))
{
var address = ServerAddress.FromUrl(url);
if(address != null)
if (address != null)
{
Addresses.Add(address);
}
@ -41,5 +41,7 @@ namespace Kestrel
}
public IList<ServerAddress> Addresses { get; private set; }
public int ThreadCount { get; set; }
}
}

2
src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs
View File

@ -24,7 +24,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
private readonly UvStreamHandle _socket;
private Frame _frame;
long _connectionId;
long _connectionId = 0;
public Connection(ListenerContext context, UvStreamHandle socket) : base(context)
{

10
src/Microsoft.AspNet.Server.Kestrel/Http/Listener.cs
View File

@ -1,6 +1,7 @@
// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
using Microsoft.AspNet.Server.Kestrel.Infrastructure;
using Microsoft.AspNet.Server.Kestrel.Networking;
using System;
using System.Diagnostics;
@ -53,7 +54,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
ListenSocket = new UvTcpHandle();
ListenSocket.Init(Thread.Loop, Thread.QueueCloseHandle);
ListenSocket.Bind(new IPEndPoint(IPAddress.Any, port));
ListenSocket.Listen(10, _connectionCallback, this);
ListenSocket.Listen(Constants.ListenBacklog, _connectionCallback, this);
tcs.SetResult(0);
}
catch (Exception ex)
@ -70,7 +71,12 @@ namespace Microsoft.AspNet.Server.Kestrel.Http
acceptSocket.Init(Thread.Loop, Thread.QueueCloseHandle);
listenSocket.Accept(acceptSocket);
var connection = new Connection(this, acceptSocket);
DispatchConnection(acceptSocket);
}
protected virtual void DispatchConnection(UvTcpHandle socket)
{
var connection = new Connection(this, socket);
connection.Start();
}

89
src/Microsoft.AspNet.Server.Kestrel/Http/ListenerPrimary.cs Normal file
View File

@ -0,0 +1,89 @@
// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
using Microsoft.AspNet.Server.Kestrel.Infrastructure;
using Microsoft.AspNet.Server.Kestrel.Networking;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
namespace Microsoft.AspNet.Server.Kestrel.Http
{
public class ListenerPrimary : Listener
{
UvPipeHandle ListenPipe { get; set; }
List<UvPipeHandle> _dispatchPipes = new List<UvPipeHandle>();
int _dispatchIndex;
ArraySegment<ArraySegment<byte>> _1234 = new ArraySegment<ArraySegment<byte>>(new[] { new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 }) });
public ListenerPrimary(IMemoryPool memory) : base(memory)
{
}
public async Task StartAsync(
string pipeName,
string scheme,
string host,
int port,
KestrelThread thread,
Func<Frame, Task> application)
{
await StartAsync(scheme, host, port, thread, application);
await Thread.PostAsync(_ =>
{
ListenPipe = new UvPipeHandle();
ListenPipe.Init(Thread.Loop, false);
ListenPipe.Bind(pipeName);
ListenPipe.Listen(Constants.ListenBacklog, OnListenPipe, null);
}, null);
}
private void OnListenPipe(UvStreamHandle pipe, int status, Exception error, object state)
{
if (status < 0)
{
return;
}
var dispatchPipe = new UvPipeHandle();
dispatchPipe.Init(Thread.Loop, true);
try
{
pipe.Accept(dispatchPipe);
}
catch (Exception)
{
dispatchPipe.Dispose();
return;
}
_dispatchPipes.Add(dispatchPipe);
}
protected override void DispatchConnection(UvTcpHandle socket)
{
var index = _dispatchIndex++ % (_dispatchPipes.Count + 1);
if (index == _dispatchPipes.Count)
{
base.DispatchConnection(socket);
}
else
{
var dispatchPipe = _dispatchPipes[index];
var write = new UvWriteReq();
write.Init(Thread.Loop);
write.Write2(
dispatchPipe,
_1234,
socket,
(write2, status, error, state) =>
{
write2.Dispose();
((UvTcpHandle)state).Dispose();
},
socket);
}
}
}
}

117
src/Microsoft.AspNet.Server.Kestrel/Http/ListenerSecondary.cs Normal file
View File

@ -0,0 +1,117 @@
// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
using Microsoft.AspNet.Server.Kestrel.Networking;
using System;
using System.Diagnostics;
using System.Runtime.InteropServices;
using System.Threading.Tasks;
namespace Microsoft.AspNet.Server.Kestrel.Http
{
public class ListenerSecondary : ListenerContext, IDisposable
{
UvPipeHandle DispatchPipe { get; set; }
public ListenerSecondary(IMemoryPool memory)
{
Memory = memory;
}
public Task StartAsync(
string pipeName,
KestrelThread thread,
Func<Frame, Task> application)
{
Thread = thread;
Application = application;
DispatchPipe = new UvPipeHandle();
var tcs = new TaskCompletionSource<int>();
Thread.Post(_ =>
{
try
{
DispatchPipe.Init(Thread.Loop, true);
var connect = new UvConnectRequest();
connect.Init(Thread.Loop);
connect.Connect(
DispatchPipe,
pipeName,
(connect2, status, error, state) =>
{
connect.Dispose();
if (error != null)
{
tcs.SetException(error);
return;
}
try
{
var ptr = Marshal.AllocHGlobal(16);
var buf = Thread.Loop.Libuv.buf_init(ptr, 16);
DispatchPipe.ReadStart(
(_1, _2, _3) => buf,
(_1, status2, error2, state2) =>
{
if (status2 == 0)
{
DispatchPipe.Dispose();
Marshal.FreeHGlobal(ptr);
return;
}
var acceptSocket = new UvTcpHandle();
acceptSocket.Init(Thread.Loop, Thread.QueueCloseHandle);
try
{
DispatchPipe.Accept(acceptSocket);
}
catch (Exception ex)
{
Trace.WriteLine("DispatchPipe.Accept " + ex.Message);
acceptSocket.Dispose();
return;
}
var connection = new Connection(this, acceptSocket);
connection.Start();
},
null);
tcs.SetResult(0);
}
catch (Exception ex)
{
DispatchPipe.Dispose();
tcs.SetException(ex);
}
},
null);
}
catch (Exception ex)
{
DispatchPipe.Dispose();
tcs.SetException(ex);
}
}, null);
return tcs.Task;
}
public void Dispose()
{
// Ensure the event loop is still running.
// If the event loop isn't running and we try to wait on this Post
// to complete, then KestrelEngine will never be disposed and
// the exception that stopped the event loop will never be surfaced.
if (Thread.FatalError == null)
{
Thread.Send(_ => DispatchPipe.Dispose(), null);
}
}
}
}

1
src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs
View File

@ -4,7 +4,6 @@
using Microsoft.AspNet.Server.Kestrel.Networking;
using System;
using System.Threading;
using System.Threading;
namespace Microsoft.AspNet.Server.Kestrel.Http
{

12
src/Microsoft.AspNet.Server.Kestrel/Infrastructure/Constants.cs Normal file
View File

@ -0,0 +1,12 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
namespace Microsoft.AspNet.Server.Kestrel.Infrastructure
{
internal class Constants
{
public const int ListenBacklog = 128;
}
}

12
src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs
View File

@ -118,6 +118,18 @@ namespace Microsoft.AspNet.Server.Kestrel
return tcs.Task;
}
public void Send(Action<object> callback, object state)
{
if (_loop.ThreadId == Thread.CurrentThread.ManagedThreadId)
{
callback.Invoke(state);
}
else
{
PostAsync(callback, state).Wait();
}
}
private void PostCloseHandle(Action<IntPtr> callback, IntPtr handle)
{
lock (_workSync)

33
src/Microsoft.AspNet.Server.Kestrel/KestrelEngine.cs
View File

@ -38,10 +38,10 @@ namespace Microsoft.AspNet.Server.Kestrel
: "amd64";
libraryPath = Path.Combine(
libraryPath,
libraryPath,
"native",
"windows",
architecture,
architecture,
"libuv.dll");
}
else if (Libuv.IsDarwin)
@ -91,16 +91,37 @@ namespace Microsoft.AspNet.Server.Kestrel
public IDisposable CreateServer(string scheme, string host, int port, Func<Frame, Task> application)
{
var listeners = new List<Listener>();
var listeners = new List<IDisposable>();
try
{
var pipeName = (Libuv.IsWindows ? @"\\.\pipe\kestrel_" : "/tmp/kestrel_") + Guid.NewGuid().ToString("n");
var single = Threads.Count == 1;
var first = true;
foreach (var thread in Threads)
{
var listener = new Listener(Memory);
if (single)
{
var listener = new Listener(Memory);
listeners.Add(listener);
listener.StartAsync(scheme, host, port, thread, application).Wait();
}
else if (first)
{
var listener = new ListenerPrimary(Memory);
listeners.Add(listener);
listener.StartAsync(pipeName, scheme, host, port, thread, application).Wait();
}
else
{
var listener = new ListenerSecondary(Memory);
listeners.Add(listener);
listener.StartAsync(pipeName, thread, application).Wait();
}
listeners.Add(listener);
listener.StartAsync(scheme, host, port, thread, application).Wait();
first = false;
}
return new Disposable(() =>
{

108
src/Microsoft.AspNet.Server.Kestrel/Networking/Libuv.cs
View File

@ -85,7 +85,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_loop_init(UvLoopHandle a0);
uv_loop_init _uv_loop_init;
uv_loop_init _uv_loop_init = default(uv_loop_init);
public void loop_init(UvLoopHandle handle)
{
Check(_uv_loop_init(handle));
@ -93,7 +93,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_loop_close(IntPtr a0);
uv_loop_close _uv_loop_close;
uv_loop_close _uv_loop_close = default(uv_loop_close);
public void loop_close(UvLoopHandle handle)
{
handle.Validate(closed: true);
@ -102,7 +102,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_run(UvLoopHandle handle, int mode);
uv_run _uv_run;
uv_run _uv_run = default(uv_run);
public int run(UvLoopHandle handle, int mode)
{
handle.Validate();
@ -111,7 +111,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate void uv_stop(UvLoopHandle handle);
uv_stop _uv_stop;
uv_stop _uv_stop = default(uv_stop);
public void stop(UvLoopHandle handle)
{
handle.Validate();
@ -120,7 +120,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate void uv_ref(UvHandle handle);
uv_ref _uv_ref;
uv_ref _uv_ref = default(uv_ref);
public void @ref(UvHandle handle)
{
handle.Validate();
@ -129,7 +129,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate void uv_unref(UvHandle handle);
uv_unref _uv_unref;
uv_unref _uv_unref = default(uv_unref);
public void unref(UvHandle handle)
{
handle.Validate();
@ -141,7 +141,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
public delegate void uv_close_cb(IntPtr handle);
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate void uv_close(IntPtr handle, uv_close_cb close_cb);
uv_close _uv_close;
uv_close _uv_close = default(uv_close);
public void close(UvHandle handle, uv_close_cb close_cb)
{
handle.Validate(closed: true);
@ -156,7 +156,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
public delegate void uv_async_cb(IntPtr handle);
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_async_init(UvLoopHandle loop, UvAsyncHandle handle, uv_async_cb cb);
uv_async_init _uv_async_init;
uv_async_init _uv_async_init = default(uv_async_init);
public void async_init(UvLoopHandle loop, UvAsyncHandle handle, uv_async_cb cb)
{
loop.Validate();
@ -166,7 +166,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_async_send(UvAsyncHandle handle);
uv_async_send _uv_async_send;
uv_async_send _uv_async_send = default(uv_async_send);
public void async_send(UvAsyncHandle handle)
{
Check(_uv_async_send(handle));
@ -174,7 +174,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_tcp_init(UvLoopHandle loop, UvTcpHandle handle);
uv_tcp_init _uv_tcp_init;
uv_tcp_init _uv_tcp_init = default(uv_tcp_init);
public void tcp_init(UvLoopHandle loop, UvTcpHandle handle)
{
loop.Validate();
@ -184,18 +184,46 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_tcp_bind(UvTcpHandle handle, ref sockaddr addr, int flags);
uv_tcp_bind _uv_tcp_bind;
uv_tcp_bind _uv_tcp_bind = default(uv_tcp_bind);
public void tcp_bind(UvTcpHandle handle, ref sockaddr addr, int flags)
{
handle.Validate();
Check(_uv_tcp_bind(handle, ref addr, flags));
}
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_tcp_open(UvTcpHandle handle, IntPtr hSocket);
uv_tcp_open _uv_tcp_open = default(uv_tcp_open);
public void tcp_open(UvTcpHandle handle, IntPtr hSocket)
{
handle.Validate();
Check(_uv_tcp_open(handle, hSocket));
}
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_pipe_init(UvLoopHandle loop, UvPipeHandle handle, int ipc);
uv_pipe_init _uv_pipe_init = default(uv_pipe_init);
public void pipe_init(UvLoopHandle loop, UvPipeHandle handle, bool ipc)
{
loop.Validate();
handle.Validate();
Check(_uv_pipe_init(loop, handle, ipc ? -1 : 0));
}
[UnmanagedFunctionPointer(CallingConvention.Cdecl, CharSet = CharSet.Ansi)]
delegate int uv_pipe_bind(UvPipeHandle loop, string name);
uv_pipe_bind _uv_pipe_bind = default(uv_pipe_bind);
public void pipe_bind(UvPipeHandle handle, string name)
{
handle.Validate();
Check(_uv_pipe_bind(handle, name));
}
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
public delegate void uv_connection_cb(IntPtr server, int status);
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_listen(UvStreamHandle handle, int backlog, uv_connection_cb cb);
uv_listen _uv_listen;
uv_listen _uv_listen = default(uv_listen);
public void listen(UvStreamHandle handle, int backlog, uv_connection_cb cb)
{
handle.Validate();
@ -204,7 +232,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_accept(UvStreamHandle server, UvStreamHandle client);
uv_accept _uv_accept;
uv_accept _uv_accept = default(uv_accept);
public void accept(UvStreamHandle server, UvStreamHandle client)
{
server.Validate();
@ -212,13 +240,25 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
Check(_uv_accept(server, client));
}
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
public delegate void uv_connect_cb(IntPtr req, int status);
[UnmanagedFunctionPointer(CallingConvention.Cdecl, CharSet = CharSet.Ansi)]
unsafe delegate int uv_pipe_connect(UvConnectRequest req, UvPipeHandle handle, string name, uv_connect_cb cb);
uv_pipe_connect _uv_pipe_connect = default(uv_pipe_connect);
unsafe public void pipe_connect(UvConnectRequest req, UvPipeHandle handle, string name, uv_connect_cb cb)
{
req.Validate();
handle.Validate();
Check(_uv_pipe_connect(req, handle, name, cb));
}
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
public delegate void uv_alloc_cb(IntPtr server, int suggested_size, out uv_buf_t buf);
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
public delegate void uv_read_cb(IntPtr server, int nread, ref uv_buf_t buf);
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_read_start(UvStreamHandle handle, uv_alloc_cb alloc_cb, uv_read_cb read_cb);
uv_read_start _uv_read_start;
uv_read_start _uv_read_start = default(uv_read_start);
public void read_start(UvStreamHandle handle, uv_alloc_cb alloc_cb, uv_read_cb read_cb)
{
handle.Validate();
@ -227,7 +267,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_read_stop(UvStreamHandle handle);
uv_read_stop _uv_read_stop;
uv_read_stop _uv_read_stop = default(uv_read_stop);
public void read_stop(UvStreamHandle handle)
{
handle.Validate();
@ -236,7 +276,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_try_write(UvStreamHandle handle, Libuv.uv_buf_t[] bufs, int nbufs);
uv_try_write _uv_try_write;
uv_try_write _uv_try_write = default(uv_try_write);
public int try_write(UvStreamHandle handle, Libuv.uv_buf_t[] bufs, int nbufs)
{
handle.Validate();
@ -246,20 +286,30 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
public delegate void uv_write_cb(IntPtr req, int status);
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
unsafe delegate int uv_write(UvWriteReq req, UvStreamHandle handle, Libuv.uv_buf_t* bufs, int nbufs, uv_write_cb cb);
uv_write _uv_write;
unsafe public void write(UvWriteReq req, UvStreamHandle handle, Libuv.uv_buf_t* bufs, int nbufs, uv_write_cb cb)
unsafe delegate int uv_write(UvRequest req, UvStreamHandle handle, Libuv.uv_buf_t* bufs, int nbufs, uv_write_cb cb);
uv_write _uv_write = default(uv_write);
unsafe public void write(UvRequest req, UvStreamHandle handle, Libuv.uv_buf_t* bufs, int nbufs, uv_write_cb cb)
{
req.Validate();
handle.Validate();
Check(_uv_write(req, handle, bufs, nbufs, cb));
}
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
unsafe delegate int uv_write2(UvRequest req, UvStreamHandle handle, Libuv.uv_buf_t* bufs, int nbufs, UvStreamHandle sendHandle, uv_write_cb cb);
uv_write2 _uv_write2 = default(uv_write2);
unsafe public void write2(UvRequest req, UvStreamHandle handle, Libuv.uv_buf_t* bufs, int nbufs, UvStreamHandle sendHandle, uv_write_cb cb)
{
req.Validate();
handle.Validate();
Check(_uv_write2(req, handle, bufs, nbufs, sendHandle, cb));
}
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
public delegate void uv_shutdown_cb(IntPtr req, int status);
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_shutdown(UvShutdownReq req, UvStreamHandle handle, uv_shutdown_cb cb);
uv_shutdown _uv_shutdown;
uv_shutdown _uv_shutdown = default(uv_shutdown);
public void shutdown(UvShutdownReq req, UvStreamHandle handle, uv_shutdown_cb cb)
{
req.Validate();
@ -269,7 +319,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate IntPtr uv_err_name(int err);
uv_err_name _uv_err_name;
uv_err_name _uv_err_name = default(uv_err_name);
public unsafe String err_name(int err)
{
IntPtr ptr = _uv_err_name(err);
@ -278,7 +328,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate IntPtr uv_strerror(int err);
uv_strerror _uv_strerror;
uv_strerror _uv_strerror = default(uv_strerror);
public unsafe String strerror(int err)
{
IntPtr ptr = _uv_strerror(err);
@ -287,7 +337,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_loop_size();
uv_loop_size _uv_loop_size;
uv_loop_size _uv_loop_size = default(uv_loop_size);
public int loop_size()
{
return _uv_loop_size();
@ -295,7 +345,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_handle_size(HandleType handleType);
uv_handle_size _uv_handle_size;
uv_handle_size _uv_handle_size = default(uv_handle_size);
public int handle_size(HandleType handleType)
{
return _uv_handle_size(handleType);
@ -303,7 +353,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_req_size(RequestType reqType);
uv_req_size _uv_req_size;
uv_req_size _uv_req_size = default(uv_req_size);
public int req_size(RequestType reqType)
{
return _uv_req_size(reqType);
@ -312,7 +362,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_ip4_addr(string ip, int port, out sockaddr addr);
uv_ip4_addr _uv_ip4_addr;
uv_ip4_addr _uv_ip4_addr = default(uv_ip4_addr);
public int ip4_addr(string ip, int port, out sockaddr addr, out Exception error)
{
return Check(_uv_ip4_addr(ip, port, out addr), out error);
@ -321,7 +371,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
delegate int uv_ip6_addr(string ip, int port, out sockaddr addr);
uv_ip6_addr _uv_ip6_addr;
uv_ip6_addr _uv_ip6_addr = default(uv_ip6_addr);
public int ip6_addr(string ip, int port, out sockaddr addr, out Exception error)
{
return Check(_uv_ip6_addr(ip, port, out addr), out error);
@ -331,7 +381,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
public delegate void uv_walk_cb(IntPtr handle, IntPtr arg);
[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
unsafe delegate int uv_walk(UvLoopHandle loop, uv_walk_cb walk_cb, IntPtr arg);
uv_walk _uv_walk;
uv_walk _uv_walk = default(uv_walk);
unsafe public void walk(UvLoopHandle loop, uv_walk_cb walk_cb, IntPtr arg)
{
loop.Validate();
@ -345,6 +395,8 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
public struct sockaddr
{
public sockaddr(long ignored) { x3 = x0 = x1 = x2 = x3 = 0; }
long x0;
long x1;
long x2;

70
src/Microsoft.AspNet.Server.Kestrel/Networking/UvConnectRequest.cs Normal file
View File

@ -0,0 +1,70 @@
// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Runtime.InteropServices;
namespace Microsoft.AspNet.Server.Kestrel.Networking
{
/// <summary>
/// Summary description for UvWriteRequest
/// </summary>
public class UvConnectRequest : UvRequest
{
private readonly static Libuv.uv_connect_cb _uv_connect_cb = UvConnectCb;
Action<UvConnectRequest, int, Exception, object> _callback;
object _state;
public void Init(UvLoopHandle loop)
{
var requestSize = loop.Libuv.req_size(Libuv.RequestType.CONNECT);
CreateMemory(
loop.Libuv,
loop.ThreadId,
requestSize);
}
public void Connect(
UvPipeHandle pipe,
string name,
Action<UvConnectRequest, int, Exception, object> callback,
object state)
{
_callback = callback;
_state = state;
Pin();
Libuv.pipe_connect(this, pipe, name, _uv_connect_cb);
}
private static void UvConnectCb(IntPtr ptr, int status)
{
var req = FromIntPtr<UvConnectRequest>(ptr);
req.Unpin();
var callback = req._callback;
req._callback = null;
var state = req._state;
req._state = null;
Exception error = null;
if (status < 0)
{
req.Libuv.Check(status, out error);
}
try
{
callback(req, status, error, state);
}
catch (Exception ex)
{
Trace.WriteLine("UvConnectRequest " + ex.ToString());
}
}
}
}

36
src/Microsoft.AspNet.Server.Kestrel/Networking/UvPipeHandle.cs Normal file
View File

@ -0,0 +1,36 @@
// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
using System;
using System.Net;
namespace Microsoft.AspNet.Server.Kestrel.Networking
{
public class UvPipeHandle : UvStreamHandle
{
public void Init(UvLoopHandle loop, bool ipc)
{
CreateMemory(
loop.Libuv,
loop.ThreadId,
loop.Libuv.handle_size(Libuv.HandleType.NAMED_PIPE));
_uv.pipe_init(loop, this, ipc);
}
public void Init(UvLoopHandle loop, Action<Action<IntPtr>, IntPtr> queueCloseHandle)
{
CreateHandle(
loop.Libuv,
loop.ThreadId,
loop.Libuv.handle_size(Libuv.HandleType.TCP), queueCloseHandle);
_uv.pipe_init(loop, this, false);
}
public void Bind(string name)
{
_uv.pipe_bind(this, name);
}
}
}

17
src/Microsoft.AspNet.Server.Kestrel/Networking/UvReq.cs
View File

@ -1,17 +0,0 @@
// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
using System;
namespace Microsoft.AspNet.Server.Kestrel.Networking
{
public abstract class UvReq : UvMemory
{
protected override bool ReleaseHandle()
{
DestroyMemory(handle);
handle = IntPtr.Zero;
return true;
}
}
}

31
src/Microsoft.AspNet.Server.Kestrel/Networking/UvRequest.cs Normal file
View File

@ -0,0 +1,31 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.InteropServices;
using System.Threading.Tasks;
namespace Microsoft.AspNet.Server.Kestrel.Networking
{
public class UvRequest : UvMemory
{
GCHandle _pin;
protected override bool ReleaseHandle()
{
DestroyMemory(handle);
handle = IntPtr.Zero;
return true;
}
public virtual void Pin()
{
_pin = GCHandle.Alloc(this, GCHandleType.Normal);
}
public virtual void Unpin()
{
_pin.Free();
}
}
}

7
src/Microsoft.AspNet.Server.Kestrel/Networking/UvShutdownReq.cs
View File

@ -9,13 +9,12 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
/// <summary>
/// Summary description for UvShutdownRequest
/// </summary>
public class UvShutdownReq : UvReq
public class UvShutdownReq : UvRequest
{
private readonly static Libuv.uv_shutdown_cb _uv_shutdown_cb = UvShutdownCb;
Action<UvShutdownReq, int, object> _callback;
object _state;
GCHandle _pin;
public void Init(UvLoopHandle loop)
{
@ -29,14 +28,14 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
{
_callback = callback;
_state = state;
_pin = GCHandle.Alloc(this, GCHandleType.Normal);
Pin();
_uv.shutdown(this, handle, _uv_shutdown_cb);
}
private static void UvShutdownCb(IntPtr ptr, int status)
{
var req = FromIntPtr<UvShutdownReq>(ptr);
req._pin.Free();
req.Unpin();
req._callback(req, status, req._state);
req._callback = null;
req._state = null;

2
src/Microsoft.AspNet.Server.Kestrel/Networking/UvStreamHandle.cs
View File

@ -46,7 +46,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
_listenCallback = callback;
_listenState = state;
_listenVitality = GCHandle.Alloc(this, GCHandleType.Normal);
_uv.listen(this, 10, _uv_connection_cb);
_uv.listen(this, backlog, _uv_connection_cb);
}
catch
{

5
src/Microsoft.AspNet.Server.Kestrel/Networking/UvTcpHandle.cs
View File

@ -48,5 +48,10 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
_uv.tcp_bind(this, ref addr, 0);
}
public void Open(IntPtr hSocket)
{
_uv.tcp_open(this, hSocket);
}
}
}

52
src/Microsoft.AspNet.Server.Kestrel/Networking/UvWriteReq.cs
View File

@ -11,7 +11,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
/// <summary>
/// Summary description for UvWriteRequest
/// </summary>
public class UvWriteReq : UvReq
public class UvWriteReq : UvRequest
{
private readonly static Libuv.uv_write_cb _uv_write_cb = UvWriteCb;
@ -26,7 +26,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
public void Init(UvLoopHandle loop)
{
var requestSize = loop.Libuv.req_size(Libuv.RequestType.WRITE);
var bufferSize = Marshal.SizeOf(typeof(Libuv.uv_buf_t)) * BUFFER_COUNT;
var bufferSize = Marshal.SizeOf<Libuv.uv_buf_t>() * BUFFER_COUNT;
CreateMemory(
loop.Libuv,
loop.ThreadId,
@ -81,6 +81,54 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking
}
}
public unsafe void Write2(
UvStreamHandle handle,
ArraySegment<ArraySegment<byte>> bufs,
UvStreamHandle sendHandle,
Action<UvWriteReq, int, Exception, object> callback,
object state)
{
try
{
// add GCHandle to keeps this SafeHandle alive while request processing
_pins.Add(GCHandle.Alloc(this, GCHandleType.Normal));
var pBuffers = (Libuv.uv_buf_t*)_bufs;
var nBuffers = bufs.Count;
if (nBuffers > BUFFER_COUNT)
{
// create and pin buffer array when it's larger than the pre-allocated one
var bufArray = new Libuv.uv_buf_t[nBuffers];
var gcHandle = GCHandle.Alloc(bufArray, GCHandleType.Pinned);
_pins.Add(gcHandle);
pBuffers = (Libuv.uv_buf_t*)gcHandle.AddrOfPinnedObject();
}
for (var index = 0; index != nBuffers; ++index)
{
// create and pin each segment being written
var buf = bufs.Array[bufs.Offset + index];
var gcHandle = GCHandle.Alloc(buf.Array, GCHandleType.Pinned);
_pins.Add(gcHandle);
pBuffers[index] = Libuv.buf_init(
gcHandle.AddrOfPinnedObject() + buf.Offset,
buf.Count);
}
_callback = callback;
_state = state;
_uv.write2(this, handle, pBuffers, nBuffers, sendHandle, _uv_write_cb);
}
catch
{
_callback = null;
_state = null;
Unpin(this);
throw;
}
}
private static void Unpin(UvWriteReq req)
{
foreach (var pin in req._pins)

BIN
src/Microsoft.AspNet.Server.Kestrel/native/windows/amd64/libuv.dll
View File

Binary file not shown.

BIN
src/Microsoft.AspNet.Server.Kestrel/native/windows/x86/libuv.dll
View File

Binary file not shown.

2
src/Microsoft.AspNet.Server.Kestrel/project.json
View File

@ -6,7 +6,7 @@
"url": "git://github.com/aspnet/kestrelhttpserver"
},
"dependencies": {
"Microsoft.Framework.Runtime.Abstractions": "1.0.0-*"
"Microsoft.Framework.Runtime.Abstractions": "1.0.0-beta7-*"
},
"frameworks": {
"dnx451": { },

271
test/Microsoft.AspNet.Server.KestrelTests/MultipleLoopTests.cs Normal file
View File

@ -0,0 +1,271 @@
using Microsoft.AspNet.Server.Kestrel;
using Microsoft.AspNet.Server.Kestrel.Networking;
using Microsoft.Framework.Runtime;
using Microsoft.Framework.Runtime.Infrastructure;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
using Xunit;
namespace Microsoft.AspNet.Server.KestrelTests
{
public class MultipleLoopTests
{
Libuv _uv;
public MultipleLoopTests()
{
var engine = new KestrelEngine(LibraryManager, new ShutdownNotImplemented());
_uv = engine.Libuv;
}
ILibraryManager LibraryManager
{
get
{
var locator = CallContextServiceLocator.Locator;
if (locator == null)
{
return null;
}
var services = locator.ServiceProvider;
if (services == null)
{
return null;
}
return (ILibraryManager)services.GetService(typeof(ILibraryManager));
}
}
[Fact]
public void InitAndCloseServerPipe()
{
var loop = new UvLoopHandle();
var pipe = new UvPipeHandle();
loop.Init(_uv);
pipe.Init(loop, true);
pipe.Bind(@"\\.\pipe\InitAndCloseServerPipe");
pipe.Dispose();
loop.Run();
pipe.Dispose();
loop.Dispose();
}
[Fact]
public void ServerPipeListenForConnections()
{
var loop = new UvLoopHandle();
var serverListenPipe = new UvPipeHandle();
loop.Init(_uv);
serverListenPipe.Init(loop, false);
serverListenPipe.Bind(@"\\.\pipe\ServerPipeListenForConnections");
serverListenPipe.Listen(128, (_1, status, error, _2) =>
{
var serverConnectionPipe = new UvPipeHandle();
serverConnectionPipe.Init(loop, true);
try
{
serverListenPipe.Accept(serverConnectionPipe);
}
catch (Exception)
{
serverConnectionPipe.Dispose();
return;
}
var writeRequest = new UvWriteReq();
writeRequest.Init(loop);
writeRequest.Write(
serverConnectionPipe,
new ArraySegment<ArraySegment<byte>>(new ArraySegment<byte>[] { new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 }) }),
(_3, status2, error2, _4) =>
{
writeRequest.Dispose();
serverConnectionPipe.Dispose();
serverListenPipe.Dispose();
},
null);
}, null);
var worker = new Thread(() =>
{
var loop2 = new UvLoopHandle();
var clientConnectionPipe = new UvPipeHandle();
var connect = new UvConnectRequest();
loop2.Init(_uv);
clientConnectionPipe.Init(loop2, true);
connect.Init(loop2);
connect.Connect(clientConnectionPipe, @"\\.\pipe\ServerPipeListenForConnections", (_1, status, error, _2) =>
{
var buf = loop2.Libuv.buf_init(Marshal.AllocHGlobal(8192), 8192);
connect.Dispose();
clientConnectionPipe.ReadStart(
(_3, cb, _4) => buf,
(_3, status2, error2, _4) =>
{
if (status2 == 0)
{
clientConnectionPipe.Dispose();
}
},
null);
}, null);
loop2.Run();
loop2.Dispose();
});
worker.Start();
loop.Run();
loop.Dispose();
worker.Join();
}
[Fact]
public void ServerPipeDispatchConnections()
{
var pipeName = @"\\.\pipe\ServerPipeDispatchConnections" + Guid.NewGuid().ToString("n");
var loop = new UvLoopHandle();
loop.Init(_uv);
var serverConnectionPipe = default(UvPipeHandle);
var serverConnectionPipeAcceptedEvent = new ManualResetEvent(false);
var serverConnectionTcpDisposedEvent = new ManualResetEvent(false);
var serverListenPipe = new UvPipeHandle();
serverListenPipe.Init(loop, false);
serverListenPipe.Bind(pipeName);
serverListenPipe.Listen(128, (_1, status, error, _2) =>
{
serverConnectionPipe = new UvPipeHandle();
serverConnectionPipe.Init(loop, true);
try
{
serverListenPipe.Accept(serverConnectionPipe);
serverConnectionPipeAcceptedEvent.Set();
}
catch (Exception ex)
{
Console.WriteLine(ex);
serverConnectionPipe.Dispose();
serverConnectionPipe = null;
}
}, null);
var serverListenTcp = new UvTcpHandle();
serverListenTcp.Init(loop);
serverListenTcp.Bind(new IPEndPoint(0, 54321));
serverListenTcp.Listen(128, (_1, status, error, _2) =>
{
var serverConnectionTcp = new UvTcpHandle();
serverConnectionTcp.Init(loop);
serverListenTcp.Accept(serverConnectionTcp);
serverConnectionPipeAcceptedEvent.WaitOne();
var writeRequest = new UvWriteReq();
writeRequest.Init(loop);
writeRequest.Write2(
serverConnectionPipe,
new ArraySegment<ArraySegment<byte>>(new ArraySegment<byte>[] { new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 }) }),
serverConnectionTcp,
(_3, status2, error2, _4) =>
{
writeRequest.Dispose();
serverConnectionTcp.Dispose();
serverConnectionTcpDisposedEvent.Set();
serverConnectionPipe.Dispose();
serverListenPipe.Dispose();
serverListenTcp.Dispose();
},
null);
}, null);
var worker = new Thread(() =>
{
var loop2 = new UvLoopHandle();
var clientConnectionPipe = new UvPipeHandle();
var connect = new UvConnectRequest();
loop2.Init(_uv);
clientConnectionPipe.Init(loop2, true);
connect.Init(loop2);
connect.Connect(clientConnectionPipe, pipeName, (_1, status, error, _2) =>
{
connect.Dispose();
var buf = loop2.Libuv.buf_init(Marshal.AllocHGlobal(64), 64);
serverConnectionTcpDisposedEvent.WaitOne();
clientConnectionPipe.ReadStart(
(_3, cb, _4) => buf,
(_3, status2, error2, _4) =>
{
if (status2 == 0)
{
clientConnectionPipe.Dispose();
return;
}
var clientConnectionTcp = new UvTcpHandle();
clientConnectionTcp.Init(loop2);
clientConnectionPipe.Accept(clientConnectionTcp);
var buf2 = loop2.Libuv.buf_init(Marshal.AllocHGlobal(64), 64);
clientConnectionTcp.ReadStart(
(_5, cb, _6) => buf2,
(_5, status3, error3, _6) =>
{
if (status3 == 0)
{
clientConnectionTcp.Dispose();
}
},
null);
},
null);
}, null);
loop2.Run();
loop2.Dispose();
});
var worker2 = new Thread(() =>
{
try
{
serverConnectionPipeAcceptedEvent.WaitOne();
var socket = new Socket(SocketType.Stream, ProtocolType.IP);
socket.Connect(IPAddress.Loopback, 54321);
socket.Send(new byte[] { 6, 7, 8, 9 });
socket.Shutdown(SocketShutdown.Send);
var cb = socket.Receive(new byte[64]);
socket.Dispose();
}
catch(Exception ex)
{
Console.WriteLine(ex);
}
});
worker.Start();
worker2.Start();
loop.Run();
loop.Dispose();
worker.Join();
worker2.Join();
}
}
}

19
test/Microsoft.AspNet.Server.KestrelTests/Program.cs
View File

@ -1,6 +1,9 @@
// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
using Microsoft.Framework.Runtime;
using System;
namespace Microsoft.AspNet.Server.KestrelTests
{
/// <summary>
@ -8,9 +11,21 @@ namespace Microsoft.AspNet.Server.KestrelTests
/// </summary>
public class Program
{
public void Main()
private readonly IApplicationEnvironment env;
private readonly IServiceProvider sp;
public Program(IApplicationEnvironment env, IServiceProvider sp)
{
new EngineTests().DisconnectingClient().Wait();
this.env = env;
this.sp = sp;
}
public int Main()
{
return new Xunit.Runner.AspNet.Program(env, sp).Main(new string[] {
"-class",
typeof(MultipleLoopTests).FullName
});
}
}
}

5
test/Microsoft.AspNet.Server.KestrelTests/project.json
View File

@ -14,8 +14,11 @@
}
}
},
"compilationOptions": {
"allowUnsafe": true
},
"commands": {
"run": "xunit.runner.aspnet",
"run": "Microsoft.AspNet.Server.KestrelTests",
"test": "xunit.runner.aspnet"
}
}