refactor: Better ShadowsocksPipe

Author: HMBSbige

Shadowsocks.Protocol/ListenServices/TcpListenService.cs

@@ -1,7 +1,6 @@
 using Microsoft.Extensions.Logging;
 using Microsoft.VisualStudio.Threading;
 using Pipelines.Extensions;
-using Pipelines.Extensions.SocketPipe;
 using Shadowsocks.Protocol.LocalTcpServices;
 using System;
 using System.Collections.Generic;
@@ -11,6 +10,7 @@
 using System.Net.Sockets;
 using System.Threading;
 using System.Threading.Tasks;
+using static Shadowsocks.Protocol.ShadowsocksProtocolConstants;
 
 namespace Shadowsocks.Protocol.ListenServices
 {
@@ -24,8 +24,6 @@ public class TcpListenService : IListenService
 		private readonly CancellationTokenSource _cts;
 
 		private const string LoggerHeader = @"[TcpListenService]";
-		private const int FirstBufferSize = 8192;
-		private static readonly SocketPipeReaderOptions LocalPipeReaderOptions = new(sizeHint: FirstBufferSize);
 
 		public TcpListenService(ILogger<TcpListenService> logger, IPEndPoint local, IEnumerable<ILocalTcpService> services)
 		{
@@ -64,7 +62,7 @@ private async Task HandleAsync(Socket socket, CancellationToken token)
 			var remoteEndPoint = socket.RemoteEndPoint;
 			try
 			{
-				var pipe = socket.AsDuplexPipe(LocalPipeReaderOptions);
+				var pipe = socket.AsDuplexPipe(SocketPipeReaderOptions, SocketPipeWriterOptions);
 				var result = await pipe.Input.ReadAsync(token);
 				var buffer = result.Buffer;
 

Shadowsocks.Protocol/ShadowsocksProtocolConstants.cs

@@ -0,0 +1,24 @@
+using Pipelines.Extensions.SocketPipe;
+using System.IO.Pipelines;
+
+namespace Shadowsocks.Protocol
+{
+	internal static class ShadowsocksProtocolConstants
+	{
+		private const int BlockSize = 4096; // 4K
+		public const int ReceiveBufferSize = 20 * BlockSize; // 80K, must < 85000 LOH
+		public const int SendBufferSize = 5 * BlockSize; // 20K, must < 85000 LOH
+
+		private const int SegmentPoolSize = 16;
+		private const int MinimumSegmentSize = BlockSize;
+		private const int ResumeWriterThreshold = PauseWriterThreshold / 2;
+		private const int PauseWriterThreshold = MinimumSegmentSize * SegmentPoolSize;
+		public static readonly PipeOptions DefaultPipeOptions = new(
+			pauseWriterThreshold: PauseWriterThreshold,
+			resumeWriterThreshold: ResumeWriterThreshold,
+			minimumSegmentSize: MinimumSegmentSize);
+
+		public static readonly SocketPipeReaderOptions SocketPipeReaderOptions = new(DefaultPipeOptions, sizeHint: ReceiveBufferSize);
+		public static readonly SocketPipeWriterOptions SocketPipeWriterOptions = new(DefaultPipeOptions);
+	}
+}

Shadowsocks.Protocol/TcpClients/IPipeClient.cs

@@ -7,7 +7,7 @@ namespace Shadowsocks.Protocol.TcpClients
 {
 	public interface IPipeClient : IAsyncDisposable
 	{
-		ValueTask ConnectAsync(CancellationToken token);
+		ValueTask ConnectAsync(CancellationToken cancellationToken = default);
 
 		IDuplexPipe GetPipe(string targetAddress, ushort targetPort);
 	}

Shadowsocks.Protocol/TcpClients/ShadowsocksDuplexPipe.cs

@@ -1,8 +1,8 @@
+using Pipelines.Extensions;
 using Shadowsocks.Protocol.Models;
 using Socks5.Utils;
 using System.IO.Pipelines;
 using System.Runtime.CompilerServices;
-using System.Threading;
 
 namespace Shadowsocks.Protocol.TcpClients
 {
@@ -13,17 +13,14 @@ public static IDuplexPipe AsShadowsocksPipe(
 			this IDuplexPipe pipe,
 			ShadowsocksServerInfo serverInfo,
 			string targetAddress, ushort targetPort,
-			PipeOptions? pipeOptions = null,
-			CancellationToken cancellationToken = default)
+			PipeOptions? readerOptions = null,
+			PipeOptions? writerOptions = null)
 		{
-			return new ShadowsocksDuplexPipe(
-				pipe,
-				serverInfo,
-				targetAddress,
-				targetPort,
-				pipeOptions,
-				cancellationToken
-			);
+			var reader = pipe.Input.AsShadowsocksPipeReader(serverInfo, readerOptions);
+			var writer = pipe.Output.AsShadowsocksPipeWriter(serverInfo, writerOptions);
+			writer.WriteShadowsocksHeader(targetAddress, targetPort);
+
+			return DefaultDuplexPipe.Create(reader, writer);
 		}
 
 		[MethodImpl(MethodImplOptions.AggressiveInlining)]
@@ -35,5 +32,23 @@ public static void WriteShadowsocksHeader(
 			var addressLength = Pack.DestinationAddressAndPort(targetAddress, default, targetPort, span);
 			writer.Advance(addressLength);
 		}
+
+		[MethodImpl(MethodImplOptions.AggressiveInlining)]
+		public static PipeWriter AsShadowsocksPipeWriter(
+			this PipeWriter writer,
+			ShadowsocksServerInfo serverInfo,
+			PipeOptions? pipeOptions = null)
+		{
+			return new ShadowsocksPipeWriter(writer, serverInfo, pipeOptions);
+		}
+
+		[MethodImpl(MethodImplOptions.AggressiveInlining)]
+		public static PipeReader AsShadowsocksPipeReader(
+			this PipeReader reader,
+			ShadowsocksServerInfo serverInfo,
+			PipeOptions? pipeOptions = null)
+		{
+			return new ShadowsocksPipeReader(reader, serverInfo, pipeOptions);
+		}
 	}
 }