You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
269 lines
16 KiB
269 lines
16 KiB
using System;
|
|
using System.Collections.Generic;
|
|
using System.IO;
|
|
using System.Net;
|
|
using System.Net.Sockets;
|
|
using System.Reflection;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
|
|
namespace DCT_test
|
|
{
|
|
public static class TransportRegression
|
|
{
|
|
private static void Check(bool value, string message) { if (!value) throw new Exception(message); }
|
|
private static DctPacket Receive(NetworkStream stream)
|
|
{
|
|
var frame = new List<byte>();
|
|
while (true)
|
|
{
|
|
int value = stream.ReadByte();
|
|
if (value < 0) throw new EndOfStreamException();
|
|
frame.Add((byte)value);
|
|
var parsed = DctPacket.ParseStream(frame);
|
|
if (parsed.Count > 0) return parsed[0];
|
|
}
|
|
}
|
|
private static DctPacket Request(NetworkStream stream, byte command, params byte[] payload)
|
|
{
|
|
byte[] bytes = DctPacket.Create(command, payload).Encode();
|
|
stream.Write(bytes, 0, bytes.Length);
|
|
return Receive(stream);
|
|
}
|
|
private static void VerifySlowReaderRecovery()
|
|
{
|
|
var listener = new TcpListener(IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
ClientSession session = null;
|
|
using (var peer = new TcpClient())
|
|
{
|
|
peer.ReceiveBufferSize = 65536;
|
|
try
|
|
{
|
|
peer.Connect(IPAddress.Loopback, ((IPEndPoint)listener.LocalEndpoint).Port);
|
|
var sender = listener.AcceptTcpClient();
|
|
sender.SendBufferSize = 4096;
|
|
session = new ClientSession(sender, "slow-reader");
|
|
int failures = 0;
|
|
session.SendFailed += (s, e) => { Interlocked.Increment(ref failures); Console.WriteLine(e); };
|
|
session.SetStreaming(true);
|
|
var bytes = new byte[4 * 1024 * 1024];
|
|
for (int i = 0; i < bytes.Length; i++) bytes[i] = (byte)(i % 251);
|
|
for (int chunk = 0; chunk < 16; chunk++)
|
|
Check(session.TryQueueTelemetry(bytes), "Queue slow-reader payload");
|
|
Thread.Sleep(1800);
|
|
Check(failures == 0 && session.SentPackets < 16, "A blocked write must survive the former 1s timeout; failures=" + failures + ", sent=" + session.SentPackets);
|
|
var stream = peer.GetStream();
|
|
stream.ReadTimeout = 5000;
|
|
var buffer = new byte[65536];
|
|
int received = 0;
|
|
while (received < bytes.Length * 16)
|
|
{
|
|
int n = stream.Read(buffer, 0, Math.Min(buffer.Length, bytes.Length * 16 - received));
|
|
Check(n > 0, "Connection must remain open after reader resumes");
|
|
for (int i = 0; i < n; i++) Check(buffer[i] == bytes[(received + i) % bytes.Length], "Byte order preserved after stall");
|
|
received += n;
|
|
}
|
|
Check(SpinWait.SpinUntil(() => session.SentPackets == 16, 2000), "Pending write must complete after recovery");
|
|
Check(session.SendControl(new byte[] { 0x20 }, 2000), "Control write after recovery");
|
|
Check(stream.ReadByte() == 0x20 && failures == 0, "Control traffic remains usable");
|
|
Console.WriteLine("PASS: 1.8s stalled reader recovers without disconnect, loss or duplication");
|
|
}
|
|
finally { if (session != null) session.Dispose(); listener.Stop(); }
|
|
}
|
|
}
|
|
|
|
private static void VerifyQueuedControlSurvivesStall()
|
|
{
|
|
var listener = new TcpListener(IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
ClientSession session = null;
|
|
try
|
|
{
|
|
using (var peer = new TcpClient())
|
|
{
|
|
peer.ReceiveBufferSize = 4096;
|
|
peer.Connect(IPAddress.Loopback, ((IPEndPoint)listener.LocalEndpoint).Port);
|
|
var sender = listener.AcceptTcpClient();
|
|
sender.SendBufferSize = 4096;
|
|
session = new ClientSession(sender, "control-stall");
|
|
session.SetStreaming(true);
|
|
var block = new byte[1024 * 1024];
|
|
session.TryQueueTelemetry(block);
|
|
Thread.Sleep(150);
|
|
var completed = new TaskCompletionSource<bool>();
|
|
Check(session.QueueControl(new byte[] { 0x77 }, sent => completed.TrySetResult(sent)), "Queue control");
|
|
Thread.Sleep(2800);
|
|
Check(!completed.Task.IsCompleted, "Control must remain pending behind stalled write");
|
|
var stream = peer.GetStream();
|
|
stream.ReadTimeout = 5000;
|
|
var buffer = new byte[65536];
|
|
int total = 0;
|
|
while (total < block.Length)
|
|
{
|
|
int count = stream.Read(buffer, 0, Math.Min(buffer.Length, block.Length - total));
|
|
Check(count > 0, "Stalled stream remains connected");
|
|
total += count;
|
|
}
|
|
Check(stream.ReadByte() == 0x77 && completed.Task.Wait(2000) && completed.Task.Result,
|
|
"Control must be delivered after the former 2.5s expiry");
|
|
int previousGeneration = session.TelemetryGeneration;
|
|
session.SetStreaming(false);
|
|
session.SetStreaming(true);
|
|
Check(!session.SendLogAsync(0xA3, new byte[] { 1 }, previousGeneration, CancellationToken.None).Result.HasValue,
|
|
"Previous run must not leak after restart");
|
|
session.Dispose();
|
|
Check(!session.SendLogAsync(0xA3, new byte[] { 1 }, session.TelemetryGeneration, CancellationToken.None).Result.HasValue,
|
|
"Disconnected file send must not wait forever");
|
|
}
|
|
}
|
|
finally { if (session != null) session.Dispose(); listener.Stop(); }
|
|
Console.WriteLine("PASS: delayed control delivery, old-run isolation, disconnect cancellation");
|
|
}
|
|
|
|
public static void Run()
|
|
{
|
|
VerifySlowReaderRecovery();
|
|
VerifyQueuedControlSurvivesStall();
|
|
var clock = new LogPlaybackClock();
|
|
clock.Reset(0);
|
|
clock.Advance(100, 1, false);
|
|
Check(clock.DelayUntil(300, 2) == 100, "2x playback interval");
|
|
clock.Advance(1100, 2, true);
|
|
Check(clock.PositionMilliseconds == 100, "Pause must freeze source time");
|
|
clock.Advance(1200, 2, false);
|
|
Check(clock.PositionMilliseconds == 300, "Resume at changed speed");
|
|
clock.Reset(1200);
|
|
Check(clock.PositionMilliseconds == 0, "Restart/loop must reset source time");
|
|
|
|
string fixture = Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString("N") + ".txt");
|
|
try
|
|
{
|
|
string hex = BitConverter.ToString(DctPacket.Create(0xA3, new byte[] { 1, 164, 1 }).Encode()).Replace('-', ' ');
|
|
File.WriteAllLines(fixture, new[] { "[Recv][23:59:59.900] " + hex,
|
|
"[Recv][00:00:00.100] " + hex, "[Recv][00:00:03.100] " + hex });
|
|
var log = new LogFileReader().Load(fixture);
|
|
Check(log.Success && log.Packets.Count == 3, "Midnight log parsing");
|
|
Check(log.Packets[1].LogElapsedMilliseconds == 200, "Midnight rollover");
|
|
Check(log.Packets[2].LogElapsedMilliseconds == 3200, "Original log gap preserved");
|
|
}
|
|
finally { File.Delete(fixture); }
|
|
|
|
var session = new ClientSession(new TcpClient(), "window-test");
|
|
session.SetStreaming(true);
|
|
for (int i = 0; i < ClientSession.TelemetryAckWindowSize; i++)
|
|
Check(session.ReserveTelemetry(0xA3), "Reserve window");
|
|
var blocked = Task.Run(() => session.ReserveTelemetry(0xA3));
|
|
Check(!blocked.Wait(50), "Full ACK window must backpressure producer");
|
|
session.Acknowledge(0x23);
|
|
Check(blocked.Wait(1000) && blocked.Result, "ACK must release producer");
|
|
blocked = Task.Run(() => session.ReserveTelemetry(0xA3));
|
|
session.SetStreaming(false);
|
|
Check(blocked.Wait(1000) && !blocked.Result, "STOP must release producer without sending");
|
|
session.Client.Close();
|
|
|
|
// A stalled ACK window must recover by retiring stale reservations;
|
|
// it is never allowed to throw and close the transport session.
|
|
var stalled = new ClientSession(new TcpClient(), "stall-test");
|
|
stalled.SetStreaming(true);
|
|
for (int i = 0; i < ClientSession.TelemetryAckWindowSize; i++)
|
|
Check(stalled.ReserveTelemetry(0xA3), "Stall window fill");
|
|
Thread.Sleep(3200);
|
|
Check(stalled.ReserveTelemetry(0xA3), "Stalled ACKs must not disconnect or throw");
|
|
stalled.SetStreaming(false);
|
|
stalled.ResetTelemetryAcks();
|
|
stalled.Client.Close();
|
|
|
|
var server = new DctTcpServer();
|
|
Check(server.Start(0), "Test protocol listener");
|
|
var listener = (TcpListener)typeof(DctTcpServer).GetField("_listener", BindingFlags.Instance | BindingFlags.NonPublic).GetValue(server);
|
|
try
|
|
{
|
|
using (var client = new TcpClient())
|
|
{
|
|
client.Connect(IPAddress.Loopback, ((IPEndPoint)listener.LocalEndpoint).Port);
|
|
var stream = client.GetStream();
|
|
stream.ReadTimeout = 3000;
|
|
Check(Request(stream, 0x20, 3).Payload[0] == 2, "START before handshake must fail");
|
|
Request(stream, 0x00); Request(stream, 0x01); Request(stream, 0x03); Request(stream, 0x04);
|
|
Check(Request(stream, 0x20, 3).Payload[0] == 2, "START before configuration must fail");
|
|
Request(stream, 0x09, 6, 0); Request(stream, 0x10, 0, 0);
|
|
Request(stream, 0x11, 0, 0); Request(stream, 0x13, 0, 0);
|
|
Check(Request(stream, 0x20, 3).Payload[0] == 1, "Configured START must succeed");
|
|
Check(SpinWait.SpinUntil(() => { foreach (var c in server.ActiveClients) if (c.Streaming) return true; return false; }, 1000), "Streaming activation");
|
|
server.BroadcastPacket(DctPacket.Create(0xA3, new byte[] { 1, 164, 7 }));
|
|
Check(Receive(stream).CommandId == 0xA3, "Telemetry after START");
|
|
byte[] ack = DctPacket.Create(0x23, new byte[] { 0 }).Encode(); // Valid NACK must retire pending frame too.
|
|
stream.Write(ack, 0, ack.Length);
|
|
// Lossless file replay must wait beyond the old 250 ms
|
|
// admission budget, not advance to the next record.
|
|
var replay = Task.Run(async () =>
|
|
{
|
|
for (int i = 0; i < 16; i++)
|
|
await server.BroadcastLogPacketAsync(DctPacket.Create(0xA3,
|
|
new byte[] { 1, 164, (byte)i }), i, CancellationToken.None);
|
|
});
|
|
Thread.Sleep(650);
|
|
Check(!replay.IsCompleted, "File replay must backpressure rather than skip records");
|
|
for (int i = 0; i < 16; i++)
|
|
{
|
|
var fileFrame = Receive(stream);
|
|
Check(fileFrame.CommandId == 0xA3 && fileFrame.Payload[2] == i,
|
|
"File record missing, duplicated or reordered: " + i);
|
|
stream.Write(ack, 0, ack.Length);
|
|
}
|
|
Check(replay.Wait(3000), "File replay resumes after ACKs");
|
|
Check(Request(stream, 0x21).CommandId == 0xA1, "STOP after backpressured file replay");
|
|
Check(Request(stream, 0x20, 1).Payload[0] == 1, "Restart after file replay");
|
|
|
|
// Do not ACK the next frame for longer than the old 3 s
|
|
// watchdog. The server must retire the stale reservation,
|
|
// send the next frame, and keep the TCP session alive.
|
|
server.BroadcastPacket(DctPacket.Create(0xA3, new byte[] { 1, 164, 9 }));
|
|
Check(Receive(stream).CommandId == 0xA3, "Telemetry before delayed ACK");
|
|
Thread.Sleep(3200);
|
|
server.BroadcastPacket(DctPacket.Create(0xA3, new byte[] { 1, 164, 10 }));
|
|
Check(Receive(stream).CommandId == 0xA3, "Telemetry after delayed ACK");
|
|
Check(client.Client.Connected, "Delayed ACK must not close TCP session");
|
|
stream.Write(ack, 0, ack.Length);
|
|
Check(Request(stream, 0x21).CommandId == 0xA1, "STOP ACK");
|
|
server.BroadcastPacket(DctPacket.Create(0xA3, new byte[] { 1, 164, 8 }));
|
|
Check(!stream.DataAvailable, "No telemetry after STOP ACK");
|
|
Check(Request(stream, 0x20, 3).Payload[0] == 1, "Restart without reconnect");
|
|
Check(SpinWait.SpinUntil(() => { foreach (var c in server.ActiveClients) if (c.Streaming) return true; return false; }, 1000),
|
|
"Streaming reactivation on the existing session");
|
|
server.BroadcastPacket(DctPacket.Create(0xA3, new byte[] { 1, 164, 11 }));
|
|
var restartedFrame = Receive(stream);
|
|
Check(restartedFrame.CommandId == 0xA3 && restartedFrame.Payload[2] == 11,
|
|
"Telemetry must resume on the same TCP connection after STOP then START");
|
|
stream.Write(ack, 0, ack.Length);
|
|
}
|
|
|
|
// PC A2L-only setup flow: no MDB 0x10/0x13 is sent.
|
|
// XCP USE -> IF_DATA -> MEASUREMENTS followed by common CAN
|
|
// configuration must therefore be sufficient for START.
|
|
using (var xcpClient = new TcpClient())
|
|
{
|
|
xcpClient.Connect(IPAddress.Loopback, ((IPEndPoint)listener.LocalEndpoint).Port);
|
|
var stream = xcpClient.GetStream();
|
|
stream.ReadTimeout = 3000;
|
|
Request(stream, 0x00); Request(stream, 0x01); Request(stream, 0x03); Request(stream, 0x04);
|
|
Check(Request(stream, 0x40, 0x01).Payload[0] == 1, "XCP USE must succeed");
|
|
Check(Request(stream, 0x42, 0x00, 0x01, 0x02, 0x00, 0x10, 0x00, 0x70).Payload[0] == 0,
|
|
"XCP measurements before IF_DATA must fail");
|
|
Check(Request(stream, 0x41, 0x00, 0x90, 0x06, 0x00, 0x00, 0x93, 0x06, 0x00, 0x00).Payload[0] == 1,
|
|
"XCP IF_DATA must succeed");
|
|
Check(Request(stream, 0x42, 0x00, 0x01, 0x02, 0x00, 0x10, 0x00, 0x70).Payload[0] == 1,
|
|
"XCP measurements must succeed after IF_DATA");
|
|
Request(stream, 0x09, 0x06, 0x00);
|
|
Request(stream, 0x11, 0x00, 0x00);
|
|
Check(Request(stream, 0x20, 0x01).Payload[0] == 1,
|
|
"PC A2L-only configuration must allow START without MDB commands");
|
|
Check(Request(stream, 0x21).CommandId == 0xA1, "XCP STOP ACK");
|
|
}
|
|
}
|
|
finally { server.Stop(); }
|
|
Console.WriteLine("PASS: log timestamps/midnight/pause/speed/reset, ACK backpressure, MDB and PC A2L-only start/stop TCP exchange");
|
|
}
|
|
}
|
|
}
|
|
|