diff --git a/src/Core/Net/ElectrumClient.cs b/src/Core/Net/ElectrumClient.cs index 674e88b..c2dc210 100644 --- a/src/Core/Net/ElectrumClient.cs +++ b/src/Core/Net/ElectrumClient.cs @@ -1,8 +1,10 @@ +using System.Buffers; using System.Collections.Concurrent; +using System.IO.Pipelines; using System.Net.Security; using System.Net.Sockets; -using System.Text; using System.Text.Json; +using System.Threading.Channels; namespace PalladiumWallet.Core.Net; @@ -18,10 +20,25 @@ public sealed class ElectrumClient : IAsyncDisposable private readonly TcpClient _tcp; private readonly Stream _stream; - private readonly SemaphoreSlim _writeLock = new(1, 1); private readonly ConcurrentDictionary> _pending = new(); private readonly CancellationTokenSource _cts = new(); private readonly Task _readLoop; + private readonly Task _writeLoop; + + // Channel single-reader: le task scrivono i payload senza lock; + // il write loop drena tutto in un unico WriteAsync+FlushAsync — + // identico al buffered writer asyncio di Electrum. + private readonly Channel _outgoing = Channel.CreateUnbounded( + new UnboundedChannelOptions { SingleReader = true, AllowSynchronousContinuations = false }); + + // Tetto alle richieste in volo (non in coda di scrittura, ma in attesa di + // risposta sul server). Il write-loop batcha già l'invio in un solo segmento; + // questo gate evita di sommergere il server con migliaia di richieste + // simultanee su wallet grandi → niente -101/-102 a raffica né drop della + // connessione. Le scritture restano comunque pipelinate fino a questo grado. + private const int MaxInFlight = 32; + private readonly SemaphoreSlim _inFlight = new(MaxInFlight, MaxInFlight); + private long _nextId; public string Host { get; } @@ -29,10 +46,7 @@ public sealed class ElectrumClient : IAsyncDisposable public bool UseSsl { get; } public bool IsConnected => _tcp.Connected && !_cts.IsCancellationRequested; - /// (metodo, parametri) per le notifiche di subscription. public event Action? NotificationReceived; - - /// Scatta quando la connessione cade (errore di lettura o chiusura remota). public event Action? Disconnected; private ElectrumClient(TcpClient tcp, Stream stream, string host, int port, bool useSsl) @@ -42,18 +56,14 @@ public sealed class ElectrumClient : IAsyncDisposable Host = host; Port = port; UseSsl = useSsl; - _readLoop = Task.Run(ReadLoopAsync); + _readLoop = Task.Run(ReadLoopAsync); + _writeLoop = Task.Run(WriteLoopAsync); } - /// - /// Connette al server. Con la validazione del - /// certificato è TOFU tramite (§9): primo contatto - /// salva, contatti successivi confrontano; mismatch ⇒ . - /// public static async Task ConnectAsync(string host, int port, bool useSsl, CertificatePinStore? pins = null, CancellationToken ct = default) { - var tcp = new TcpClient(); + var tcp = new TcpClient { NoDelay = true }; try { await tcp.ConnectAsync(host, port, ct); @@ -64,10 +74,7 @@ public sealed class ElectrumClient : IAsyncDisposable var ssl = new SslStream(stream, leaveInnerStreamOpen: false, (_, cert, _, _) => { - // I server Electrum sono tipicamente self-signed: la - // fiducia è il pin TOFU, non la catena CA (§9). - if (cert is null) - return false; + if (cert is null) return false; pinOk = pins is null || pins.VerifyOrPin(host, port, cert); return pinOk; }); @@ -84,7 +91,6 @@ public sealed class ElectrumClient : IAsyncDisposable } var client = new ElectrumClient(tcp, stream, host, port, useSsl); - // Negoziazione obbligatoria prima di ogni altra richiesta. await client.RequestAsync("server.version", ct, ClientName, ProtocolVersion); return client; } @@ -98,75 +104,107 @@ public sealed class ElectrumClient : IAsyncDisposable public async Task RequestAsync(string method, CancellationToken ct = default, params object?[] parameters) { - var id = Interlocked.Increment(ref _nextId); - var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - _pending[id] = tcs; - - var payload = JsonSerializer.SerializeToUtf8Bytes(new - { - jsonrpc = "2.0", - id, - method, - @params = parameters, - }); - - await _writeLock.WaitAsync(ct); + // Gate prima di mettere in volo: oltre MaxInFlight richieste in attesa + // si attende che una risposta liberi uno slot, invece di sommergere il server. + await _inFlight.WaitAsync(ct); try { - await _stream.WriteAsync(payload, ct); - await _stream.WriteAsync("\n"u8.ToArray(), ct); - await _stream.FlushAsync(ct); + var id = Interlocked.Increment(ref _nextId); + var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _pending[id] = tcs; + + var payload = JsonSerializer.SerializeToUtf8Bytes(new + { + jsonrpc = "2.0", + id, + method, + @params = parameters, + }); + + _outgoing.Writer.TryWrite(payload); + + await using var registration = ct.Register(() => + { + _pending.TryRemove(id, out _); + tcs.TrySetCanceled(ct); + }); + return await tcs.Task; } finally { - _writeLock.Release(); + _inFlight.Release(); } - - await using var registration = ct.Register(() => tcs.TrySetCanceled(ct)); - return await tcs.Task; } - private async Task ReadLoopAsync() + /// + /// Drain loop: svuota il channel in un unico buffer → un solo WriteAsync+FlushAsync + /// per tutti i messaggi in coda. Quando N richieste sono in coda, vengono + /// trasmesse in un singolo segmento TCP invece di N flush seriali. + /// + private async Task WriteLoopAsync() { - Exception? failure = null; try { - using var reader = new StreamReader(_stream, Encoding.UTF8, leaveOpen: true); - while (!_cts.IsCancellationRequested) + while (await _outgoing.Reader.WaitToReadAsync(_cts.Token)) { - var line = await reader.ReadLineAsync(_cts.Token); - if (line is null) - break; // chiusura remota - if (string.IsNullOrWhiteSpace(line)) - continue; - - using var doc = JsonDocument.Parse(line); - var root = doc.RootElement; - - if (root.TryGetProperty("id", out var idEl) && idEl.ValueKind == JsonValueKind.Number) + using var ms = new MemoryStream(512); + while (_outgoing.Reader.TryRead(out var data)) { - if (!_pending.TryRemove(idEl.GetInt64(), out var tcs)) - continue; - if (root.TryGetProperty("error", out var error) && error.ValueKind != JsonValueKind.Null) - tcs.TrySetException(new ElectrumServerException(error.ToString())); - else - tcs.TrySetResult(root.GetProperty("result").Clone()); - } - else if (root.TryGetProperty("method", out var methodEl)) - { - NotificationReceived?.Invoke( - methodEl.GetString()!, - root.GetProperty("params").Clone()); + ms.Write(data); + ms.WriteByte((byte)'\n'); } + await _stream.WriteAsync(ms.GetBuffer().AsMemory(0, (int)ms.Length), _cts.Token); + await _stream.FlushAsync(_cts.Token); } } catch (OperationCanceledException) { } catch (Exception ex) + { + foreach (var (_, pending) in _pending) + pending.TrySetException(ex); + } + } + + /// + /// Read loop con PipeReader: buffer pooled, zero allocazioni per-risposta + /// (nessuna stringa intermedia), parsing JSON direttamente da byte span. + /// + private async Task ReadLoopAsync() + { + Exception? failure = null; + var pipe = PipeReader.Create(_stream, new StreamPipeReaderOptions(leaveOpen: true)); + try + { + while (true) + { + ReadResult result; + try { result = await pipe.ReadAsync(_cts.Token); } + catch (OperationCanceledException) { break; } + + var buffer = result.Buffer; + try + { + while (TrySliceLine(ref buffer, out var line)) + if (!line.IsEmpty) + DispatchLine(line); + } + finally + { + // consumed = tutto ciò che abbiamo consumato (fino all'ultimo \n) + // examined = tutto ciò che abbiamo guardato (fino alla fine del buffer) + pipe.AdvanceTo(buffer.Start, buffer.End); + } + + if (result.IsCompleted) break; + } + } + catch (Exception ex) { failure = ex; } finally { + await pipe.CompleteAsync(); foreach (var (_, tcs) in _pending) tcs.TrySetException(failure ?? new IOException("Connessione al server chiusa.")); _pending.Clear(); @@ -174,13 +212,67 @@ public sealed class ElectrumClient : IAsyncDisposable } } + private static bool TrySliceLine(ref ReadOnlySequence buffer, + out ReadOnlySequence line) + { + var pos = buffer.PositionOf((byte)'\n'); + if (pos is null) { line = default; return false; } + line = buffer.Slice(0, pos.Value); + buffer = buffer.Slice(buffer.GetPosition(1, pos.Value)); + return true; + } + + private void DispatchLine(ReadOnlySequence line) + { + if (line.IsSingleSegment) + { + DispatchSpan(line.FirstSpan); + return; + } + // Multi-segmento (risposta molto lunga): copia su ArrayPool poi parsa. + var len = (int)line.Length; + var buf = ArrayPool.Shared.Rent(len); + try + { + line.CopyTo(buf); + DispatchSpan(buf.AsSpan(0, len)); + } + finally { ArrayPool.Shared.Return(buf); } + } + + private void DispatchSpan(ReadOnlySpan utf8) + { + // JsonDocument.Parse via Utf8JsonReader: nessuna stringa intermedia, + // parsing direttamente dallo span pooled. + var reader = new Utf8JsonReader(utf8); + using var doc = JsonDocument.ParseValue(ref reader); + var root = doc.RootElement; + + if (root.TryGetProperty("id", out var idEl) && idEl.ValueKind == JsonValueKind.Number) + { + if (!_pending.TryRemove(idEl.GetInt64(), out var tcs)) return; + if (root.TryGetProperty("error", out var err) && err.ValueKind != JsonValueKind.Null) + tcs.TrySetException(new ElectrumServerException(err.ToString())); + else + tcs.TrySetResult(root.GetProperty("result").Clone()); + } + else if (root.TryGetProperty("method", out var methodEl)) + { + NotificationReceived?.Invoke( + methodEl.GetString()!, + root.GetProperty("params").Clone()); + } + } + public async ValueTask DisposeAsync() { await _cts.CancelAsync(); + _outgoing.Writer.Complete(); _tcp.Close(); - try { await _readLoop; } catch { /* in chiusura */ } + try { await _readLoop; } catch { } + try { await _writeLoop; } catch { } _cts.Dispose(); - _writeLock.Dispose(); + _inFlight.Dispose(); } }