Commit 33d50f32 authored by justcoding121's avatar justcoding121

Fix -ive server connection count

parent e662451f
......@@ -37,6 +37,7 @@ namespace Titanium.Web.Proxy
Task<TcpServerConnection> prefetchConnectionTask = null;
bool closeServerConnection = false;
bool calledRequestHandler = false;
try
{
......@@ -289,7 +290,7 @@ namespace Titanium.Web.Proxy
await tcpConnectionFactory.Release(connection, true);
}
}
calledRequestHandler = true;
// Now create the request
await handleHttpSessionRequest(endPoint, clientConnection, clientStream, clientStreamWriter,
cancellationTokenSource, connectHostname, connectArgs?.WebSession.ConnectRequest, prefetchConnectionTask);
......@@ -316,7 +317,8 @@ namespace Titanium.Web.Proxy
}
finally
{
if (prefetchConnectionTask != null)
if (!calledRequestHandler
&& prefetchConnectionTask != null)
{
var connection = await prefetchConnectionTask;
await tcpConnectionFactory.Release(connection, closeServerConnection);
......
......@@ -42,6 +42,29 @@ namespace Titanium.Web.Proxy.Network.Tcp
internal ProxyServer server { get; set; }
internal string GetConnectionCacheKey(string remoteHostName, int remotePort,
Version httpVersion, bool isHttps, List<SslApplicationProtocol> applicationProtocols, bool isConnect,
ProxyServer proxyServer, IPEndPoint upStreamEndPoint, ExternalProxy externalProxy)
{
var cacheKeyBuilder = new StringBuilder($"{remoteHostName}-{remotePort}" +
$"-{(httpVersion == null ? string.Empty : httpVersion.ToString())}" +
$"-{isHttps}-{isConnect}-");
if (applicationProtocols != null)
{
foreach (var protocol in applicationProtocols)
{
cacheKeyBuilder.Append($"{protocol}-");
}
}
cacheKeyBuilder.Append(upStreamEndPoint != null
? $"{upStreamEndPoint.Address}-{upStreamEndPoint.Port}-"
: string.Empty);
cacheKeyBuilder.Append(externalProxy != null ? $"{externalProxy.GetCacheKey()}-" : string.Empty);
return cacheKeyBuilder.ToString();
}
/// <summary>
/// Gets a TCP connection to server from connection pool.
/// </summary>
......@@ -61,23 +84,9 @@ namespace Titanium.Web.Proxy.Network.Tcp
ProxyServer proxyServer, IPEndPoint upStreamEndPoint, ExternalProxy externalProxy,
CancellationToken cancellationToken)
{
var cacheKeyBuilder = new StringBuilder($"{remoteHostName}-{remotePort}" +
$"-{(httpVersion == null ? string.Empty : httpVersion.ToString())}" +
$"-{isHttps}-{isConnect}-");
if (applicationProtocols != null)
{
foreach (var protocol in applicationProtocols)
{
cacheKeyBuilder.Append($"{protocol}-");
}
}
cacheKeyBuilder.Append(upStreamEndPoint != null
? $"{upStreamEndPoint.Address}-{upStreamEndPoint.Port}-"
: string.Empty);
cacheKeyBuilder.Append(externalProxy != null ? $"{externalProxy.GetCacheKey()}-" : string.Empty);
string cacheKey = cacheKeyBuilder.ToString();
var cacheKey = GetConnectionCacheKey(remoteHostName, remotePort,
httpVersion, isHttps, applicationProtocols, isConnect,
proxyServer, upStreamEndPoint, externalProxy);
if (proxyServer.EnableConnectionPool)
{
......
......@@ -189,11 +189,11 @@ namespace Titanium.Web.Proxy
newConnection = true;
}
// create a new connection if hostname/upstream end point changes
// create a new connection if cache key changes
if (serverConnection != null
&& (!serverConnection.HostName.EqualsIgnoreCase(request.RequestUri.Host)
|| args.WebSession.UpStreamEndPoint?.Equals(serverConnection.UpStreamEndPoint) ==
false))
&& (await getConnectionCacheKey(args, false,
clientConnection.NegotiatedApplicationProtocol)
!= serverConnection.CacheKey))
{
await tcpConnectionFactory.Release(serverConnection);
serverConnection = null;
......@@ -263,16 +263,14 @@ namespace Titanium.Web.Proxy
throw new Exception("Session was terminated by user.");
}
//TODO find why enabling this cause server connection count to go -ive
//TODO find why enabling this causes invalid httpCmd exception occationally
//With connection pool get connection for each HTTP session instead of per client connection.
//That will be more efficient especially when client is holding connection but not using it
//if (EnableConnectionPool)
//{
// await tcpConnectionFactory.Release(serverConnection);
// serverConnection = null;
// prefetchTask = null;
//}
if (EnableConnectionPool)
{
await tcpConnectionFactory.Release(serverConnection);
serverConnection = null;
prefetchTask = null;
}
}
catch (Exception e) when (!(e is ProxyHttpException))
......@@ -294,17 +292,10 @@ namespace Titanium.Web.Proxy
}
}
finally
{
//don't release prefetched connection here.
//it will be handled by the parent method which created it.
if (prefetchConnectionTask == null
|| serverConnection != await prefetchConnectionTask)
{
await tcpConnectionFactory.Release(serverConnection,
closeServerConnection);
}
}
}
/// <summary>
......@@ -386,54 +377,59 @@ namespace Titanium.Web.Proxy
}
/// <summary>
/// Create a server connection.
/// Handle upgrade to websocket
/// </summary>
/// <param name="args">The session event arguments.</param>
/// <param name="isConnect">Is this a CONNECT request.</param>
/// <param name="applicationProtocol"></param>
/// <param name="cancellationToken">The cancellation token for this async task.</param>
/// <returns></returns>
private Task<TcpServerConnection> getServerConnection(SessionEventArgsBase args, bool isConnect,
SslApplicationProtocol applicationProtocol, CancellationToken cancellationToken)
private async Task handleWebSocketUpgrade(string httpCmd,
SessionEventArgs args, Request request, Response response,
CustomBufferedStream clientStream, HttpResponseWriter clientStreamWriter,
TcpServerConnection serverConnection,
CancellationTokenSource cancellationTokenSource, CancellationToken cancellationToken)
{
List<SslApplicationProtocol> applicationProtocols = null;
if (applicationProtocol != default)
// prepare the prefix content
await serverConnection.StreamWriter.WriteLineAsync(httpCmd, cancellationToken);
await serverConnection.StreamWriter.WriteHeadersAsync(request.Headers,
cancellationToken: cancellationToken);
string httpStatus;
try
{
applicationProtocols = new List<SslApplicationProtocol> { applicationProtocol };
httpStatus = await serverConnection.Stream.ReadLineAsync(cancellationToken);
if (httpStatus == null)
{
throw new ServerConnectionException("Server connection was closed.");
}
return getServerConnection(args, isConnect, applicationProtocols, cancellationToken);
}
/// <summary>
/// Create a server connection.
/// </summary>
/// <param name="args">The session event arguments.</param>
/// <param name="isConnect">Is this a CONNECT request.</param>
/// <param name="applicationProtocols"></param>
/// <param name="cancellationToken">The cancellation token for this async task.</param>
/// <returns></returns>
private async Task<TcpServerConnection> getServerConnection(SessionEventArgsBase args, bool isConnect,
List<SslApplicationProtocol> applicationProtocols, CancellationToken cancellationToken)
catch (Exception e) when (!(e is ServerConnectionException))
{
ExternalProxy customUpStreamProxy = null;
throw new ServerConnectionException("Server connection was closed.", e);
}
bool isHttps = args.IsHttps;
if (GetCustomUpStreamProxyFunc != null)
Response.ParseResponseLine(httpStatus, out var responseVersion,
out int responseStatusCode,
out string responseStatusDescription);
response.HttpVersion = responseVersion;
response.StatusCode = responseStatusCode;
response.StatusDescription = responseStatusDescription;
await HeaderParser.ReadHeaders(serverConnection.Stream, response.Headers,
cancellationToken);
if (!args.IsTransparent)
{
customUpStreamProxy = await GetCustomUpStreamProxyFunc(args);
await clientStreamWriter.WriteResponseAsync(response,
cancellationToken: cancellationToken);
}
args.CustomUpStreamProxyUsed = customUpStreamProxy;
// If user requested call back then do it
if (!args.WebSession.Response.Locked)
{
await invokeBeforeResponse(args);
}
return await tcpConnectionFactory.GetClient(
args.WebSession.Request.RequestUri.Host,
args.WebSession.Request.RequestUri.Port,
args.WebSession.Request.HttpVersion,
isHttps, applicationProtocols, isConnect,
this, args.WebSession.UpStreamEndPoint ?? UpStreamEndPoint,
customUpStreamProxy ?? (isHttps ? UpStreamHttpsProxy : UpStreamHttpProxy),
cancellationToken);
await TcpHelper.SendRaw(clientStream, serverConnection.Stream, BufferPool, BufferSize,
(buffer, offset, count) => { args.OnDataSent(buffer, offset, count); },
(buffer, offset, count) => { args.OnDataReceived(buffer, offset, count); },
cancellationTokenSource, ExceptionFunc);
}
/// <summary>
......@@ -462,62 +458,93 @@ namespace Titanium.Web.Proxy
requestHeaders.FixProxyHeaders();
}
/// <summary>
/// Handle upgrade to websocket
/// Gets the connection cache key.
/// </summary>
private async Task handleWebSocketUpgrade(string httpCmd,
SessionEventArgs args, Request request, Response response,
CustomBufferedStream clientStream, HttpResponseWriter clientStreamWriter,
TcpServerConnection serverConnection,
CancellationTokenSource cancellationTokenSource, CancellationToken cancellationToken)
{
// prepare the prefix content
await serverConnection.StreamWriter.WriteLineAsync(httpCmd, cancellationToken);
await serverConnection.StreamWriter.WriteHeadersAsync(request.Headers,
cancellationToken: cancellationToken);
string httpStatus;
try
/// <param name="args">The session event arguments.</param>
/// <param name="isConnect">Is this a CONNECT request.</param>
/// <param name="applicationProtocol"></param>
/// <returns></returns>
private async Task<string> getConnectionCacheKey(SessionEventArgsBase args, bool isConnect,
SslApplicationProtocol applicationProtocol)
{
httpStatus = await serverConnection.Stream.ReadLineAsync(cancellationToken);
if (httpStatus == null)
List<SslApplicationProtocol> applicationProtocols = null;
if (applicationProtocol != default)
{
throw new ServerConnectionException("Server connection was closed.");
}
applicationProtocols = new List<SslApplicationProtocol> { applicationProtocol };
}
catch (Exception e) when (!(e is ServerConnectionException))
ExternalProxy customUpStreamProxy = null;
bool isHttps = args.IsHttps;
if (GetCustomUpStreamProxyFunc != null)
{
throw new ServerConnectionException("Server connection was closed.", e);
customUpStreamProxy = await GetCustomUpStreamProxyFunc(args);
}
Response.ParseResponseLine(httpStatus, out var responseVersion,
out int responseStatusCode,
out string responseStatusDescription);
response.HttpVersion = responseVersion;
response.StatusCode = responseStatusCode;
response.StatusDescription = responseStatusDescription;
args.CustomUpStreamProxyUsed = customUpStreamProxy;
await HeaderParser.ReadHeaders(serverConnection.Stream, response.Headers,
cancellationToken);
return tcpConnectionFactory.GetConnectionCacheKey(
args.WebSession.Request.RequestUri.Host,
args.WebSession.Request.RequestUri.Port,
args.WebSession.Request.HttpVersion,
isHttps, applicationProtocols, isConnect,
this, args.WebSession.UpStreamEndPoint ?? UpStreamEndPoint,
customUpStreamProxy ?? (isHttps ? UpStreamHttpsProxy : UpStreamHttpProxy));
}
if (!args.IsTransparent)
/// <summary>
/// Create a server connection.
/// </summary>
/// <param name="args">The session event arguments.</param>
/// <param name="isConnect">Is this a CONNECT request.</param>
/// <param name="applicationProtocol"></param>
/// <param name="cancellationToken">The cancellation token for this async task.</param>
/// <returns></returns>
private Task<TcpServerConnection> getServerConnection(SessionEventArgsBase args, bool isConnect,
SslApplicationProtocol applicationProtocol, CancellationToken cancellationToken)
{
await clientStreamWriter.WriteResponseAsync(response,
cancellationToken: cancellationToken);
List<SslApplicationProtocol> applicationProtocols = null;
if (applicationProtocol != default)
{
applicationProtocols = new List<SslApplicationProtocol> { applicationProtocol };
}
// If user requested call back then do it
if (!args.WebSession.Response.Locked)
{
await invokeBeforeResponse(args);
return getServerConnection(args, isConnect, applicationProtocols, cancellationToken);
}
await TcpHelper.SendRaw(clientStream, serverConnection.Stream, BufferPool, BufferSize,
(buffer, offset, count) => { args.OnDataSent(buffer, offset, count); },
(buffer, offset, count) => { args.OnDataReceived(buffer, offset, count); },
cancellationTokenSource, ExceptionFunc);
/// <summary>
/// Create a server connection.
/// </summary>
/// <param name="args">The session event arguments.</param>
/// <param name="isConnect">Is this a CONNECT request.</param>
/// <param name="applicationProtocols"></param>
/// <param name="cancellationToken">The cancellation token for this async task.</param>
/// <returns></returns>
private async Task<TcpServerConnection> getServerConnection(SessionEventArgsBase args, bool isConnect,
List<SslApplicationProtocol> applicationProtocols, CancellationToken cancellationToken)
{
ExternalProxy customUpStreamProxy = null;
bool isHttps = args.IsHttps;
if (GetCustomUpStreamProxyFunc != null)
{
customUpStreamProxy = await GetCustomUpStreamProxyFunc(args);
}
args.CustomUpStreamProxyUsed = customUpStreamProxy;
return await tcpConnectionFactory.GetClient(
args.WebSession.Request.RequestUri.Host,
args.WebSession.Request.RequestUri.Port,
args.WebSession.Request.HttpVersion,
isHttps, applicationProtocols, isConnect,
this, args.WebSession.UpStreamEndPoint ?? UpStreamEndPoint,
customUpStreamProxy ?? (isHttps ? UpStreamHttpsProxy : UpStreamHttpProxy),
cancellationToken);
}
/// <summary>
/// Invoke before request handler if it is set.
......
......@@ -35,6 +35,7 @@ namespace Titanium.Web.Proxy
Task<TcpServerConnection> prefetchConnectionTask = null;
bool closeServerConnection = false;
bool calledRequestHandler = false;
try
{
......@@ -127,7 +128,7 @@ namespace Titanium.Web.Proxy
}
}
calledRequestHandler = true;
// HTTPS server created - we can now decrypt the client's traffic
// Now create the request
await handleHttpSessionRequest(endPoint, clientConnection, clientStream, clientStreamWriter,
......@@ -155,7 +156,8 @@ namespace Titanium.Web.Proxy
}
finally
{
if (prefetchConnectionTask != null)
if (!calledRequestHandler
&& prefetchConnectionTask != null)
{
var connection = await prefetchConnectionTask;
await tcpConnectionFactory.Release(connection, closeServerConnection);
......
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