flywithu/CryptoExchange.Net
1
0
Fork 0
mirror of https://github.com/JKorf/CryptoExchange.Net.git synced 2026-08-17 11:23:00 +00:00
Code Issues Releases Wiki Activity

Ratelimiting for socket requests

Browse Source
This commit is contained in:
JKorf
2023-08-24 20:51:17 +02:00
parent 468cd5e48e
commit be25a68c9c
15 changed files with 291 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CryptoExchange.Net/Sockets/CryptoExchangeWebSocketClient.cs
+47 -62
View File
@@ -7,6 +7,7 @@ using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Http;
using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks;
@@ -30,9 +31,8 @@ namespace CryptoExchange.Net.Sockets
private static readonly object _streamIdLock = new();
private readonly AsyncResetEvent _sendEvent;
private readonly ConcurrentQueue<byte[]> _sendBuffer;
private readonly ConcurrentQueue<SendItem> _sendBuffer;
private readonly SemaphoreSlim _closeSem;
private readonly List<DateTime> _outgoingMessages;
private ClientWebSocket _socket;
private CancellationTokenSource _ctsSource;
@@ -103,6 +103,9 @@ namespace CryptoExchange.Net.Sockets
/// <inheritdoc />
public event Action<string>? OnMessage;
/// <inheritdoc />
public event Action<int>? OnRequestSent;
/// <inheritdoc />
public event Action<Exception>? OnError;
@@ -128,10 +131,9 @@ namespace CryptoExchange.Net.Sockets
_logger = logger;
Parameters = websocketParameters;
_outgoingMessages = new List<DateTime>();
_receivedMessages = new List<ReceiveItem>();
_sendEvent = new AsyncResetEvent();
_sendBuffer = new ConcurrentQueue<byte[]>();
_sendBuffer = new ConcurrentQueue<SendItem>();
_ctsSource = new CancellationTokenSource();
_receivedMessagesLock = new object();
@@ -270,14 +272,14 @@ namespace CryptoExchange.Net.Sockets
}
/// <inheritdoc />
public virtual void Send(string data)
public virtual void Send(int id, string data, int weight)
{
if (_ctsSource.IsCancellationRequested)
return;
var bytes = Parameters.Encoding.GetBytes(data);
_logger.Log(LogLevel.Trace, $"Socket {Id} Adding {bytes.Length} to sent buffer");
_sendBuffer.Enqueue(bytes);
_logger.Log(LogLevel.Trace, $"Socket {Id} - msg {id} - Adding {bytes.Length} to send buffer");
_sendBuffer.Enqueue(new SendItem { Id = id, Weight = weight, Bytes = bytes });
_sendEvent.Set();
}
@@ -392,6 +394,7 @@ namespace CryptoExchange.Net.Sockets
{
try
{
var limitKey = Uri.ToString() + "/" + Id.ToString();
while (true)
{
if (_ctsSource.IsCancellationRequested)
@@ -404,25 +407,24 @@ namespace CryptoExchange.Net.Sockets
while (_sendBuffer.TryDequeue(out var data))
{
if (Parameters.RatelimitPerSecond != null)
if (Parameters.RateLimiters != null)
{
// Wait for rate limit
DateTime? start = null;
while (MessagesSentLastSecond() >= Parameters.RatelimitPerSecond)
foreach(var ratelimiter in Parameters.RateLimiters)
{
start ??= DateTime.UtcNow;
await Task.Delay(50).ConfigureAwait(false);
var limitResult = await ratelimiter.LimitRequestAsync(_logger, limitKey, HttpMethod.Get, false, null, RateLimitingBehaviour.Wait, data.Weight, _ctsSource.Token).ConfigureAwait(false);
if (limitResult.Success)
{
if (limitResult.Data > 0)
_logger.Log(LogLevel.Debug, $"Socket {Id} - msg {data.Id} - send delayed {limitResult.Data}ms because of rate limit");
}
}
if (start != null)
_logger.Log(LogLevel.Debug, $"Socket {Id} sent delayed {Math.Round((DateTime.UtcNow - start.Value).TotalMilliseconds)}ms because of rate limit");
}
try
{
await _socket.SendAsync(new ArraySegment<byte>(data, 0, data.Length), WebSocketMessageType.Text, true, _ctsSource.Token).ConfigureAwait(false);
_outgoingMessages.Add(DateTime.UtcNow);
_logger.Log(LogLevel.Trace, $"Socket {Id} sent {data.Length} bytes");
await _socket.SendAsync(new ArraySegment<byte>(data.Bytes, 0, data.Bytes.Length), WebSocketMessageType.Text, true, _ctsSource.Token).ConfigureAwait(false);
OnRequestSent?.Invoke(data.Id);
_logger.Log(LogLevel.Trace, $"Socket {Id} - msg {data.Id} - sent {data.Bytes.Length} bytes");
}
catch (OperationCanceledException)
{
@@ -630,42 +632,6 @@ namespace CryptoExchange.Net.Sockets
}
}
/// <summary>
/// Trigger the OnMessage event
/// </summary>
/// <param name="data"></param>
protected void TriggerOnMessage(string data)
{
LastActionTime = DateTime.UtcNow;
OnMessage?.Invoke(data);
}
/// <summary>
/// Trigger the OnError event
/// </summary>
/// <param name="ex"></param>
protected void TriggerOnError(Exception ex) => OnError?.Invoke(ex);
/// <summary>
/// Trigger the OnError event
/// </summary>
protected void TriggerOnOpen() => OnOpen?.Invoke();
/// <summary>
/// Trigger the OnError event
/// </summary>
protected void TriggerOnClose() => OnClose?.Invoke();
/// <summary>
/// Trigger the OnReconnecting event
/// </summary>
protected void TriggerOnReconnecting() => OnReconnecting?.Invoke();
/// <summary>
/// Trigger the OnReconnected event
/// </summary>
protected void TriggerOnReconnected() => OnReconnected?.Invoke();
/// <summary>
/// Checks if there is no data received for a period longer than the specified timeout
/// </summary>
@@ -721,13 +687,6 @@ namespace CryptoExchange.Net.Sockets
}
}
private int MessagesSentLastSecond()
{
var testTime = DateTime.UtcNow;
_outgoingMessages.RemoveAll(r => testTime - r > TimeSpan.FromSeconds(1));
return _outgoingMessages.Count;
}
/// <summary>
/// Update the received messages list, removing messages received longer than 3s ago
/// </summary>
@@ -769,6 +728,32 @@ namespace CryptoExchange.Net.Sockets
}
}
/// <summary>
/// Message info
/// </summary>
public struct SendItem
{
/// <summary>
/// The request id
/// </summary>
public int Id { get; set; }
/// <summary>
/// The request id
/// </summary>
public int Weight { get; set; }
/// <summary>
/// Timestamp the request was sent
/// </summary>
public DateTime SendTime { get; set; }
/// <summary>
/// The bytes to send
/// </summary>
public byte[] Bytes { get; set; }
}
/// <summary>
/// Received message info
/// </summary>
CryptoExchange.Net/Sockets/PendingRequest.cs
+9 -3
View File
@@ -7,6 +7,7 @@ namespace CryptoExchange.Net.Sockets
{
internal class PendingRequest
{
public int Id { get; set; }
public Func<JToken, bool> Handler { get; }
public JToken? Result { get; private set; }
public bool Completed { get; private set; }
@@ -15,17 +16,22 @@ namespace CryptoExchange.Net.Sockets
public TimeSpan Timeout { get; }
public SocketSubscription? Subscription { get; }
private CancellationTokenSource _cts;
private CancellationTokenSource? _cts;
public PendingRequest(Func<JToken, bool> handler, TimeSpan timeout, SocketSubscription? subscription)
public PendingRequest(int id, Func<JToken, bool> handler, TimeSpan timeout, SocketSubscription? subscription)
{
Id = id;
Handler = handler;
Event = new AsyncResetEvent(false, false);
Timeout = timeout;
RequestTimestamp = DateTime.UtcNow;
Subscription = subscription;
}
_cts = new CancellationTokenSource(timeout);
public void IsSend()
{
// Start timeout countdown
_cts = new CancellationTokenSource(Timeout);
_cts.Token.Register(Fail, false);
}
CryptoExchange.Net/Sockets/SocketConnection.cs
+55 -14
View File
@@ -182,6 +182,7 @@ namespace CryptoExchange.Net.Sockets
_socket = socket;
_socket.OnMessage += HandleMessage;
_socket.OnRequestSent += HandleRequestSent;
_socket.OnOpen += HandleOpen;
_socket.OnClose += HandleClose;
_socket.OnReconnecting += HandleReconnecting;
@@ -284,6 +285,22 @@ namespace CryptoExchange.Net.Sockets
_logger.Log(LogLevel.Warning, $"Socket {SocketId} error: " + e.ToLogString());
}
/// <summary>
/// Handler for whenever a request is sent over the websocket
/// </summary>
/// <param name="requestId">Id of the request sent</param>
protected virtual void HandleRequestSent(int requestId)
{
var pendingRequest = _pendingRequests.SingleOrDefault(p => p.Id == requestId);
if (pendingRequest == null)
{
_logger.Log(LogLevel.Debug, $"Socket {SocketId} - msg {requestId} - message sent, but not pending");
return;
}
pendingRequest.IsSend();
}
/// <summary>
/// Process a message received by the socket
/// </summary>
@@ -318,7 +335,6 @@ namespace CryptoExchange.Net.Sockets
// Check if this message is an answer on any pending requests
foreach (var pendingRequest in requests)
{
if (pendingRequest.CheckData(tokenData))
{
lock (_pendingRequests)
@@ -329,12 +345,13 @@ namespace CryptoExchange.Net.Sockets
// Answer to a timed out request, unsub if it is a subscription request
if (pendingRequest.Subscription != null)
{
_logger.Log(LogLevel.Warning, "Received subscription info after request timed out; unsubscribing. Consider increasing the SocketResponseTimout");
_logger.Log(LogLevel.Warning, "Received subscription info after request timed out; unsubscribing. Consider increasing the RequestTimeout");
_ = ApiClient.UnsubscribeAsync(this, pendingRequest.Subscription).ConfigureAwait(false);
}
}
else
{
_logger.Log(LogLevel.Trace, $"Socket {SocketId} - msg {pendingRequest.Id} - received data matched to pending request");
pendingRequest.Succeed(tokenData);
}
@@ -570,45 +587,69 @@ namespace CryptoExchange.Net.Sockets
/// <param name="timeout">The timeout for response</param>
/// <param name="subscription">Subscription if this is a subscribe request</param>
/// <param name="handler">The response handler, should return true if the received JToken was the response to the request</param>
/// <param name="weight">The weight of the message</param>
/// <returns></returns>
public virtual Task SendAndWaitAsync<T>(T obj, TimeSpan timeout, SocketSubscription? subscription, Func<JToken, bool> handler)
public virtual async Task SendAndWaitAsync<T>(T obj, TimeSpan timeout, SocketSubscription? subscription, int weight, Func<JToken, bool> handler)
{
var pending = new PendingRequest(handler, timeout, subscription);
var pending = new PendingRequest(ExchangeHelpers.NextId(), handler, timeout, subscription);
lock (_pendingRequests)
{
_pendingRequests.Add(pending);
}
var sendOk = Send(obj);
if(!sendOk)
pending.Fail();
return pending.Event.WaitAsync(timeout);
var sendOk = Send(pending.Id, obj, weight);
if (!sendOk)
{
pending.Fail();
return;
}
while (true)
{
if(!_socket.IsOpen)
{
pending.Fail();
return;
}
if (pending.Completed)
return;
await pending.Event.WaitAsync(TimeSpan.FromMilliseconds(500)).ConfigureAwait(false);
if (pending.Completed)
return;
}
}
/// <summary>
/// Send data over the websocket connection
/// </summary>
/// <typeparam name="T">The type of the object to send</typeparam>
/// <param name="requestId">The request id</param>
/// <param name="obj">The object to send</param>
/// <param name="nullValueHandling">How null values should be serialized</param>
public virtual bool Send<T>(T obj, NullValueHandling nullValueHandling = NullValueHandling.Ignore)
/// <param name="weight">The weight of the message</param>
public virtual bool Send<T>(int requestId, T obj, int weight, NullValueHandling nullValueHandling = NullValueHandling.Ignore)
{
if(obj is string str)
return Send(str);
return Send(requestId, str, weight);
else
return Send(JsonConvert.SerializeObject(obj, Formatting.None, new JsonSerializerSettings { NullValueHandling = nullValueHandling }));
return Send(requestId, JsonConvert.SerializeObject(obj, Formatting.None, new JsonSerializerSettings { NullValueHandling = nullValueHandling }), weight);
}
/// <summary>
/// Send string data over the websocket connection
/// </summary>
/// <param name="data">The data to send</param>
public virtual bool Send(string data)
/// <param name="weight">The weight of the message</param>
/// <param name="requestId">The id of the request</param>
public virtual bool Send(int requestId, string data, int weight)
{
_logger.Log(LogLevel.Trace, $"Socket {SocketId} sending data: {data}");
_logger.Log(LogLevel.Trace, $"Socket {SocketId} - msg {requestId} - sending messsage: {data}");
try
{
_socket.Send(data);
_socket.Send(requestId, data, weight);
return true;
}
catch(Exception)
CryptoExchange.Net/Sockets/WebSocketParameters.cs
+4 -3
View File
@@ -1,4 +1,5 @@
using CryptoExchange.Net.Objects;
using CryptoExchange.Net.Interfaces;
using CryptoExchange.Net.Objects;
using System;
using System.Collections.Generic;
using System.Text;
@@ -52,9 +53,9 @@ namespace CryptoExchange.Net.Sockets
public TimeSpan? KeepAliveInterval { get; set; }
/// <summary>
/// The max amount of messages to send per second
/// The rate limiters for the socket connection
/// </summary>
public int? RatelimitPerSecond { get; set; }
public IEnumerable<IRateLimiter>? RateLimiters { get; set; }
/// <summary>
/// Origin header value to send in the connection handshake