diff --git a/src/Nonemm.App/AppSession.cs b/src/Nonemm.App/AppSession.cs index 94d523f..5f27f27 100644 --- a/src/Nonemm.App/AppSession.cs +++ b/src/Nonemm.App/AppSession.cs @@ -1,3 +1,4 @@ +using System.Collections.Concurrent; using Avalonia.Threading; using Nonemm.App.Configuration; using Nonemm.Contests; @@ -363,9 +364,7 @@ public sealed class AppSession : IDisposable /// multi-operator entry, following what the settings now say. public void ApplyNetworkSettings() { - announcing?.Stop(); network?.Dispose(); - link?.Dispose(); network = null; if (!Settings.NetworkEnabled) { @@ -378,8 +377,7 @@ public sealed class AppSession : IDisposable Settings.NetworkPeers); // the socket thread must not touch the log: every other change to it is // made where the windows read it - network.UpdateArrived += (_, update) => - Dispatcher.UIThread.Post(() => TakeFromNetwork(update)); + network.UpdateArrived += (_, update) => QueueFromNetwork(update); network.Start(); Changed?.Invoke(this, EventArgs.Empty); } @@ -395,6 +393,7 @@ public sealed class AppSession : IDisposable public void ApplyStationLinkSettings() { + announcing?.Stop(); link?.Dispose(); link = null; if (!Settings.StationLinkEnabled) @@ -412,8 +411,7 @@ public sealed class AppSession : IDisposable }; // the socket thread must not touch the log: every other change to it is // made where the windows read it - link.UpdateArrived += (_, update) => - Dispatcher.UIThread.Post(() => TakeFromNetwork(update)); + link.UpdateArrived += (_, update) => QueueFromNetwork(update); // the resync reads the log rather than being handed it, and another // station can ask for a rescore when its resync is done link.Log = new SessionContactLog(this); @@ -607,34 +605,75 @@ public sealed class AppSession : IDisposable host.Length > 1 && int.TryParse(host[1], out int given) ? given : Settings.StationLinkPort); } - /// What another station did to its log, applied to ours. Nothing goes back - /// out to the network, and the contact is scored here from the rules - /// instead of trusting what the sender put in the message. - private void TakeFromNetwork(ContactUpdate update) + /// Updates that have arrived from the network and not yet been applied. + /// They are applied together on the window thread, so a resync that sends + /// a few hundred contacts rescores the log and redraws the windows once + /// per batch rather than once per contact. + private readonly ConcurrentQueue fromNetwork = new(); + + private int networkDrainPosted; + + private void QueueFromNetwork(ContactUpdate update) + { + fromNetwork.Enqueue(update); + if (Interlocked.Exchange(ref networkDrainPosted, 1) == 0) + { + Dispatcher.UIThread.Post(TakeQueuedFromNetwork); + } + } + + private void TakeQueuedFromNetwork() + { + // cleared first: an update that arrives while this runs posts a new drain + Interlocked.Exchange(ref networkDrainPosted, 0); + List batch = []; + while (fromNetwork.TryDequeue(out ContactUpdate? update)) + { + batch.Add(update); + } + TakeFromNetwork(batch); + } + + /// What other stations did to their logs, applied to ours in the order it + /// arrived. Nothing goes back out to the network, and the contacts are + /// scored here from the rules instead of trusting what the sender put in + /// the message. + private void TakeFromNetwork(IReadOnlyList updates) { if (Logging is null) { return; } - if (!Settings.LogQsosAtAllStations && update is not ContactsCleared) + List logged = []; + foreach (ContactUpdate update in updates) { - return; - } - switch (update) - { - case ContactLogged logged: - Logging.AddFromNetwork(logged.Qso); - break; - case ContactReplaced replaced: - Logging.ReplaceFromNetwork(replaced.Qso, replaced.OldCall, replaced.OldTimestampUtc); - break; - case ContactDeleted deleted: - Logging.DeleteFromNetwork(deleted.Id, deleted.Call, deleted.TimestampUtc); - break; - case ContactsCleared cleared: - Logging.ClearFromNetwork(cleared.StationName, cleared.FromUtc, cleared.Length); - break; + if (!Settings.LogQsosAtAllStations && update is not ContactsCleared) + { + continue; + } + if (update is ContactLogged contact) + { + logged.Add(contact.Qso); + continue; + } + // a run of new contacts is applied before the update that follows + // it, because that update may be a clear or an edit of one of them + Logging.AddRangeFromNetwork(logged); + logged.Clear(); + switch (update) + { + case ContactReplaced replaced: + Logging.ReplaceFromNetwork(replaced.Qso, replaced.OldCall, replaced.OldTimestampUtc); + break; + case ContactDeleted deleted: + Logging.DeleteFromNetwork(deleted.Id, deleted.Call, deleted.TimestampUtc); + break; + case ContactsCleared cleared: + Logging.ClearFromNetwork(cleared.StationName, cleared.FromUtc, cleared.Length); + break; + } } + Logging.AddRangeFromNetwork(logged); } /// Starts, restarts or stops the keyer, following what the settings say. diff --git a/src/Nonemm.Contests/ContestLog.cs b/src/Nonemm.Contests/ContestLog.cs index a8e963f..5b890f7 100644 --- a/src/Nonemm.Contests/ContestLog.cs +++ b/src/Nonemm.Contests/ContestLog.cs @@ -127,12 +127,28 @@ public sealed class ContestLog Rebuild(); } + /// `Insert` for several contacts, scoring the log once instead of once per + /// contact. A resync sends a few hundred in a burst. + public void InsertRange(IEnumerable arrived) + { + qsos.AddRange(arrived); + Rebuild(); + } + public void Remove(string id) { qsos.RemoveAll(q => q.Id == id); Rebuild(); } + /// `Remove` for several contacts, scoring the log once. + public void RemoveRange(IReadOnlyCollection ids) + { + HashSet gone = [.. ids]; + qsos.RemoveAll(q => gone.Contains(q.Id)); + Rebuild(); + } + public void Replace(Qso qso) { int at = qsos.FindIndex(q => q.Id == qso.Id); diff --git a/src/Nonemm.Network/StationLink.cs b/src/Nonemm.Network/StationLink.cs index 8983e21..60a8232 100644 --- a/src/Nonemm.Network/StationLink.cs +++ b/src/Nonemm.Network/StationLink.cs @@ -54,6 +54,7 @@ public sealed class StationLink : IDisposable private readonly string computerName; private readonly string version; private readonly CancellationTokenSource stopping = new(); + private bool disposed; private readonly ConcurrentDictionary stations = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary writers = new(StringComparer.OrdinalIgnoreCase); private readonly NetworkedStation mine; @@ -496,6 +497,11 @@ public sealed class StationLink : IDisposable public void Dispose() { + if (disposed) + { + return; + } + disposed = true; stopping.Cancel(); foreach (TcpClient writer in writers.Values) { diff --git a/src/Nonemm.Session/ContestSession.cs b/src/Nonemm.Session/ContestSession.cs index 9f8e4dd..822a67f 100644 --- a/src/Nonemm.Session/ContestSession.cs +++ b/src/Nonemm.Session/ContestSession.cs @@ -131,10 +131,10 @@ public sealed class ContestSession foreach (Qso qso in gone) { store.Delete(qso.Id); - Log.Remove(qso.Id); } if (gone.Count > 0) { + Log.RemoveRange([.. gone.Select(q => q.Id)]); Changed?.Invoke(this, EventArgs.Empty); } return gone.Count; @@ -208,6 +208,34 @@ public sealed class ContestSession return true; } + /// `AddFromNetwork` for a burst of contacts, which is what a resync sends. + /// The log is scored and `Changed` is raised once for the lot, because + /// doing either per contact keeps the window thread busy for minutes on a + /// log of a few thousand. Returns how many were new. + public int AddRangeFromNetwork(IEnumerable arriving) + { + HashSet held = [.. Log.Qsos.Select(q => q.Id)]; + List stored = []; + foreach (Qso qso in arriving) + { + if (held.Add(qso.Id)) + { + stored.Add(store.Add(FromNetwork(qso))); + } + } + if (stored.Count == 0) + { + return 0; + } + Log.InsertRange(stored); + foreach (Qso qso in stored) + { + RecordNumber(qso); + } + Changed?.Invoke(this, EventArgs.Empty); + return stored.Count; + } + /// An edit made at another station. False when we do not hold the contact /// it changes. public bool ReplaceFromNetwork(Qso qso, string oldCall, DateTime oldTimestampUtc) diff --git a/tests/Nonemm.Session.Tests/NetworkedContactsTests.cs b/tests/Nonemm.Session.Tests/NetworkedContactsTests.cs index dac479f..f5cc3e0 100644 --- a/tests/Nonemm.Session.Tests/NetworkedContactsTests.cs +++ b/tests/Nonemm.Session.Tests/NetworkedContactsTests.cs @@ -166,6 +166,23 @@ public class NetworkedContactsTests Assert.Equal(1, session.Rescore(Start.AddMinutes(60))); } + /// A resync sends its contacts in a burst, and they are scored together. + [Fact] + public void ABurstOfContactsIsLoggedOnceAndRaisesOneChange() + { + (ContestSession session, FakeLogStore store) = Station(); + int changes = 0; + session.Changed += (_, _) => changes++; + + int added = session.AddRangeFromNetwork( + [Arriving("JA1XYZ", 1), Arriving("JA2XYZ", 2), Arriving("JA1XYZ", 1)]); + + Assert.Equal(2, added); + Assert.Equal(1, changes); + Assert.Equal(2, store.Qsos(session.Instance.ContestNumber).Count); + Assert.Equal(2, session.Log.Qsos.Count); + } + /// A resync drops the block it is about to be sent again, so the contacts /// are replaced rather than added to. [Fact]