Skip to content

Commit f67d7fd

Browse files
cnblogs-duducnblogs-dudu
committed
Use new sockets api introduced in .net core 2.1
1 parent adb1e6b commit f67d7fd

File tree

9 files changed

+68
-1290
lines changed

9 files changed

+68
-1290
lines changed

‎Enyim.Caching/Memcached/AsyncPooledSocket.cs‎

Lines changed: 19 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,18 @@
11
//#define DEBUG_IO
2+
using Microsoft.Extensions.Logging;
23
using System;
3-
using System.Linq;
44
using System.Collections.Generic;
55
using System.Diagnostics;
66
using System.IO;
7+
using System.Linq;
78
using System.Net;
89
using System.Net.Sockets;
10+
using System.Reflection;
11+
using System.Runtime.CompilerServices;
12+
using System.Runtime.InteropServices;
913
using System.Text;
1014
using System.Threading;
1115
using System.Threading.Tasks;
12-
using Microsoft.Extensions.Logging;
13-
using Dawn.Net.Sockets;
14-
using System.Runtime.InteropServices;
15-
using System.Reflection;
16-
using System.Runtime.CompilerServices;
1716

1817
namespace Enyim.Caching.Memcached
1918
{
@@ -29,7 +28,7 @@ public partial class AsyncPooledSocket : IDisposable
2928
public AsyncPooledSocket(ILogger logger)
3029
{
3130
_logger = logger;
32-
_isAlive = true;
31+
_isAlive = true;
3332
}
3433

3534
private async Task CreateSocketAsync(DnsEndPoint endpoint, TimeSpan connectionTimeout, TimeSpan receiveTimeout)
@@ -65,7 +64,7 @@ private async Task CreateSocketAsync(DnsEndPoint endpoint, TimeSpan connectionTi
6564
_socket.NoDelay = true;
6665
_socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true);
6766

68-
_inputStream = new NetworkStream(_socket, ownsSocket: true);
67+
_inputStream = new NetworkStream(_socket, ownsSocket: true);
6968
}
7069

7170
//From https://github.com/dotnet/corefx/blob/master/src/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/ConnectHelper.cs
@@ -78,7 +77,7 @@ public void Initialize(CancellationToken cancellationToken)
7877
{
7978
CancellationToken = cancellationToken;
8079
var b = new AsyncTaskMethodBuilder();
81-
var ignored = b.Task;
80+
var ignored = b.Task;
8281
Builder = b;
8382
}
8483

@@ -104,7 +103,7 @@ protected override void OnCompleted(SocketAsyncEventArgs _)
104103
break;
105104
}
106105
}
107-
}
106+
}
108107

109108
public Action<AsyncPooledSocket> CleanupCallback { get; set; }
110109

@@ -230,12 +229,9 @@ public int ReadByte()
230229

231230
public async Task<byte[]> ReadBytesAsync(int count)
232231
{
233-
using (var awaitable = new SocketAwaitable())
234-
{
235-
awaitable.Buffer = new ArraySegment<byte>(new byte[count], 0, count);
236-
await _socket.ReceiveAsync(awaitable);
237-
return awaitable.Transferred.Array;
238-
}
232+
var buffer = new ArraySegment<byte>(new byte[count], 0, count);
233+
await _socket.ReceiveAsync(buffer, SocketFlags.None);
234+
return buffer.Array;
239235
}
240236

241237
/// <summary>
@@ -315,24 +311,14 @@ public void Write(IList<ArraySegment<byte>> buffers)
315311

316312
public async Task WriteSync(IList<ArraySegment<byte>> buffers)
317313
{
318-
using (var awaitable = new SocketAwaitable())
314+
try
319315
{
320-
awaitable.Arguments.BufferList = buffers;
321-
try
322-
{
323-
await _socket.SendAsync(awaitable);
324-
}
325-
catch
326-
{
327-
_isAlive = false;
328-
ThrowHelper.ThrowSocketWriteError(_socket.RemoteEndPoint, awaitable.Arguments.SocketError);
329-
}
330-
331-
if (awaitable.Arguments.SocketError != SocketError.Success)
332-
{
333-
_isAlive = false;
334-
ThrowHelper.ThrowSocketWriteError(_socket.RemoteEndPoint, awaitable.Arguments.SocketError);
335-
}
316+
await _socket.SendAsync(buffers, SocketFlags.None);
317+
}
318+
catch (Exception ex)
319+
{
320+
_isAlive = false;
321+
_logger.LogError(ex, nameof(PooledSocket.WriteSync));
336322
}
337323
}
338324

‎Enyim.Caching/Memcached/MemcachedNode.cs‎

Lines changed: 25 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,18 @@
1+
using Enyim.Caching.Configuration;
2+
using Enyim.Caching.Memcached.Protocol.Binary;
3+
using Enyim.Caching.Memcached.Results;
4+
using Enyim.Caching.Memcached.Results.Extensions;
5+
using Enyim.Collections;
6+
using Microsoft.Extensions.Logging;
17
using System;
28
using System.Collections.Generic;
39
using System.Diagnostics;
10+
using System.IO;
411
using System.Net;
5-
using System.Threading;
6-
using Enyim.Caching.Configuration;
7-
using Enyim.Collections;
8-
using System.Security;
9-
using Enyim.Caching.Memcached.Protocol.Binary;
1012
using System.Runtime.Serialization;
11-
using System.IO;
12-
using Enyim.Caching.Memcached.Results;
13-
using Enyim.Caching.Memcached.Results.Extensions;
13+
using System.Security;
14+
using System.Threading;
1415
using System.Threading.Tasks;
15-
using Microsoft.Extensions.Logging;
1616

1717
namespace Enyim.Caching.Memcached
1818
{
@@ -27,13 +27,13 @@ public class MemcachedNode : IMemcachedNode
2727

2828
private bool isDisposed;
2929

30-
private DnsEndPoint endPoint;
31-
private ISocketPoolConfiguration config;
30+
private readonly DnsEndPoint endPoint;
31+
private readonly ISocketPoolConfiguration config;
3232
private InternalPoolImpl internalPoolImpl;
3333
private bool isInitialized;
3434

3535
public MemcachedNode(
36-
DnsEndPoint endpoint,
36+
DnsEndPoint endpoint,
3737
ISocketPoolConfiguration socketPoolConfig,
3838
ILogger logger)
3939
{
@@ -186,7 +186,7 @@ void IDisposable.Dispose()
186186
private class InternalPoolImpl : IDisposable
187187
{
188188
private readonly ILogger _logger;
189-
private bool _isDebugEnabled;
189+
private readonly bool _isDebugEnabled;
190190

191191
/// <summary>
192192
/// A list of already connected but free to use sockets
@@ -197,18 +197,18 @@ private class InternalPoolImpl : IDisposable
197197
private bool isAlive;
198198
private DateTime markedAsDeadUtc;
199199

200-
private int minItems;
201-
private int maxItems;
200+
private readonly int minItems;
201+
private readonly int maxItems;
202202

203203
private MemcachedNode ownerNode;
204-
private EndPoint endPoint;
205-
private TimeSpan queueTimeout;
204+
private readonly EndPoint endPoint;
205+
private readonly TimeSpan queueTimeout;
206206
private Semaphore semaphore;
207207

208-
private object initLock = new Object();
208+
private readonly object initLock = new Object();
209209

210210
internal InternalPoolImpl(
211-
MemcachedNode ownerNode,
211+
MemcachedNode ownerNode,
212212
ISocketPoolConfiguration config,
213213
ILogger logger)
214214
{
@@ -519,9 +519,9 @@ protected internal virtual PooledSocket CreateSocket()
519519
{
520520
return new PooledSocket(this.endPoint, this.config.ConnectionTimeout, this.config.ReceiveTimeout, _logger);
521521
}
522-
catch(Exception ex)
522+
catch (Exception ex)
523523
{
524-
_logger.LogError(new EventId (this.GetHashCode(), nameof(MemcachedNode) ), ex, $"Create {nameof(PooledSocket)}");
524+
_logger.LogError(new EventId(this.GetHashCode(), nameof(MemcachedNode)), ex, $"Create {nameof(PooledSocket)}");
525525
throw;
526526
}
527527
}
@@ -582,12 +582,16 @@ protected virtual IPooledSocketResult ExecuteOperation(IOperation op)
582582

583583
protected async virtual Task<IPooledSocketResult> ExecuteOperationAsync(IOperation op)
584584
{
585+
_logger.LogDebug($"ExecuteOperationAsync({op})");
586+
585587
var result = this.Acquire();
586588
if (result.Success && result.HasValue)
587589
{
588590
try
589591
{
590592
var pooledSocket = result.Value;
593+
594+
591595
//if Get, call BinaryRequest.CreateBuffer()
592596
var b = op.GetBuffer();
593597

‎Enyim.Caching/Memcached/PooledSocket.cs‎

Lines changed: 20 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,16 @@
1-
//#define DEBUG_IO
1+
using Microsoft.Extensions.Logging;
22
using System;
3-
using System.Linq;
43
using System.Collections.Generic;
54
using System.Diagnostics;
65
using System.IO;
6+
using System.Linq;
77
using System.Net;
88
using System.Net.Sockets;
9+
using System.Reflection;
10+
using System.Runtime.InteropServices;
911
using System.Text;
1012
using System.Threading;
1113
using System.Threading.Tasks;
12-
using Microsoft.Extensions.Logging;
13-
using Dawn.Net.Sockets;
14-
using System.Runtime.InteropServices;
15-
using System.Reflection;
1614

1715
namespace Enyim.Caching.Memcached
1816
{
@@ -34,7 +32,7 @@ public PooledSocket(DnsEndPoint endpoint, TimeSpan connectionTimeout, TimeSpan r
3432

3533
this.isAlive = true;
3634

37-
var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
35+
var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
3836
socket.NoDelay = true;
3937

4038
var timeout = connectionTimeout == TimeSpan.MaxValue
@@ -46,14 +44,14 @@ public PooledSocket(DnsEndPoint endpoint, TimeSpan connectionTimeout, TimeSpan r
4644
: (int)receiveTimeout.TotalMilliseconds;
4745

4846
socket.ReceiveTimeout = rcv;
49-
socket.SendTimeout = rcv;
47+
socket.SendTimeout = rcv;
5048

5149
ConnectWithTimeout(socket, endpoint, timeout);
5250

5351
this.socket = socket;
5452
this.endpoint = endpoint;
5553

56-
this.inputStream = new NetworkStream(socket);
54+
this.inputStream = new NetworkStream(socket);
5755
}
5856

5957
private void ConnectWithTimeout(Socket socket, DnsEndPoint endpoint, int timeout)
@@ -80,13 +78,13 @@ private void ConnectWithTimeout(Socket socket, DnsEndPoint endpoint, int timeout
8078
args.Completed += (s, e) => mres.Set();
8179
if (socket.ConnectAsync(args))
8280
{
83-
if(!mres.Wait(timeout))
81+
if (!mres.Wait(timeout))
8482
{
8583
throw new TimeoutException("Could not connect to " + endpoint);
8684
}
8785
}
88-
}
89-
}
86+
}
87+
}
9088

9189
public Action<PooledSocket> CleanupCallback { get; set; }
9290

@@ -212,12 +210,9 @@ public int ReadByte()
212210

213211
public async Task<byte[]> ReadBytesAsync(int count)
214212
{
215-
using (var awaitable = new SocketAwaitable())
216-
{
217-
awaitable.Buffer = new ArraySegment<byte>(new byte[count], 0, count);
218-
await this.socket.ReceiveAsync(awaitable);
219-
return awaitable.Transferred.Array;
220-
}
213+
var buffer = new ArraySegment<byte>(new byte[count], 0, count);
214+
await this.socket.ReceiveAsync(buffer, SocketFlags.None);
215+
return buffer.Array;
221216
}
222217

223218
/// <summary>
@@ -297,24 +292,14 @@ public void Write(IList<ArraySegment<byte>> buffers)
297292

298293
public async Task WriteSync(IList<ArraySegment<byte>> buffers)
299294
{
300-
using (var awaitable = new SocketAwaitable())
295+
try
301296
{
302-
awaitable.Arguments.BufferList = buffers;
303-
try
304-
{
305-
await this.socket.SendAsync(awaitable);
306-
}
307-
catch
308-
{
309-
this.isAlive = false;
310-
ThrowHelper.ThrowSocketWriteError(this.endpoint, awaitable.Arguments.SocketError);
311-
}
312-
313-
if (awaitable.Arguments.SocketError != SocketError.Success)
314-
{
315-
this.isAlive = false;
316-
ThrowHelper.ThrowSocketWriteError(this.endpoint, awaitable.Arguments.SocketError);
317-
}
297+
await socket.SendAsync(buffers, SocketFlags.None);
298+
}
299+
catch (Exception ex)
300+
{
301+
isAlive = false;
302+
_logger.LogError(ex, nameof(PooledSocket.WriteSync));
318303
}
319304
}
320305

0 commit comments

Comments
 (0)