ew8bak/ewsdr
1
0
Fork 0
mirror of https://git.vladimir.cc/vladimir/ewsdr.git synced 2026-08-25 17:27:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(tci): ревизия — потоки, валидация, арбитраж и синхронизация клиентов

Browse Source 
Разбор семи проходов ревью ветки. Ниже — по сути, а не по списку.

Потоки. Сетевые потоки больше не читают модель контроллера напрямую. Слайсы
снимаются в потоке контроллера (RefreshSlices → FSliceSnap, на событиях
rfSliceFreq/rfSliceState/rfDevice/…), железо — тоже (RefreshDev → TTCIDevSnap:
имя платы, границы, число панов, HasTX). Копия TCtrlSlice из чужого потока
портила счётчик ссылок managed-строк, а BackendCaps и BoardDisplayName смотрят
в FNetwork, который UI освобождает на смене устройства. По той же причине
ActiveTXFreqHz переведён на GetSliceView. Sync-методы читают живую таблицу: они
уже в потоке контроллера.

Жизненный цикл. Stop ждёт выхода клиентских потоков БЕЗ таймаута, прокачивая
очередь Synchronize: выйти по таймауту нельзя — следом освобождаются и клиенты,
и сам сервер. OnDisconnect зовётся и при остановке (иначе захваты параметров
ушедших клиентов доживали до следующего запуска). Отправка переехала на поток
самого клиента (recv с TCI_POLL_MS): общий поток задерживал всех на таймаут
записи в один медленный сокет. WebUtils.SockSend шлёт с MSG_NOSIGNAL — SIGPIPE
убивал headless-процесс.

Транспорт. Слот протокола выдаётся только после Upgrade, а сокет до него живёт
по таймауту handshake: восемь молчащих соединений закрывали дверь настоящим
клиентам. Handshake с заголовком Origin получает 403 — авторизации в TCI нет, и
без этого открытая вкладка браузера дотягивалась до TRX и VFO. Заголовки
разбираются построчно, текстовые кадры проверяются на UTF-8, close длиной один
байт отвергается, на close отвечаем close.

Валидация. Все установки ходят через TCITryArg* — «vfo^0~0~abc» больше не
превращается в честный ноль. Частота проверяется дважды: в потоке клиента по
снимку и в SyncSetVfo/SyncSetCenter по живым границам (устройство успевают
сменить между разбором и исполнением). Границы теперь из ОДНОГО источника
(FreqLimits поверх VisibleFreqBounds) — тот же, что уходит в VFO_LIMITS; сами
VFO_LIMITS переобъявляются при смене железа, и их кэш ведётся независимо от
того, подключён ли кто-то. Слайс двигается только TuneSliceInBand, как у CAT:
прямой SetSliceTarget уводил TX-слайс в DUC на чужой диапазон без антенн и
фильтров. Параметры потоков сверяются со списками спецификации, а IQ_START и
прочие запуски честно отвечают ошибкой вместо молчания.

Синхронизация клиентов (§3.5). Появился захват параметра на 200 мс: два логгера
больше не перетягивают частоту. Пачка инициализации уходит под FClientLock —
изменение между строкой снимка и READY терялось навсегда. Глобальные величины
(tune_drive, cw_macros_*, split_enable, mon_volume) рассылаются всем, а правки
оператора приходят событиями: rfTXProfile, rfActiveVfo, rfMonVolume и новый
rfCWSettings. Создание и удаление слайса рассылается по rfDevice (сравнение
расстановки), у живого пана без слайсов канал A показывает центр — иначе клиент
навсегда оставался с частотой удалённого слайса.

Прочее. SliceFreqChanged переехал внутрь SetSliceTarget — один путь для мыши,
CAT и TCI (перетаскивание флага мимо клиентов проходило молча). VOLUME и
MON_VOLUME развели: SetVolume правит АКТИВНУЮ громкость, поэтому команда на
DUP-передаче уезжала в монитор — добавлен адресный SetRxVolume. Настройки
сохраняются только после успешного применения, при отказе поднимается прежний
слушатель. Время спота — UTC. Подписки на измерители читаются и пишутся под
локом клиента.

Проверено стендом (сырой WS-клиент + живой TRadioController без железа):
73 проверки, включая изоляцию медленного клиента, остановку под Synchronize,
арбитраж до и после 200 мс, отбраковку по живым границам и переобъявление
VFO_LIMITS. На реальном железе и с реальным клиентом по-прежнему не гонялось.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
ew8bak
2026-08-17 22:53:32 +03:00
co-authored by Claude Opus 5
parent 84c9e60b93
commit 4165cbe9a5
7 changed files with 1738 additions and 418 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TCIServer.pas
+424 -92
View File
@@ -16,14 +16,22 @@ unit TCIServer;
измерителей с индивидуальным для каждого клиента периодом.
Отправка НИКОГДА не блокирует того, кто зовёт Send/Broadcast: строка кладётся
в очередь клиента, а в сокет её пишет тик-поток (FlushClients). Иначе
медленный клиент останавливал бы UI-поток на секунду за раз — уведомления
рождаются в OnState, то есть внутри Changed() контроллера.
в очередь клиента, а в сокет её пишет ЕГО СОБСТВЕННЫЙ поток (Flush в цикле
HandleClient, recv просыпается каждые TCI_POLL_MS). Иначе медленный клиент
останавливал бы UI-поток на секунду за раз — уведомления рождаются в OnState,
то есть внутри Changed() контроллера, — а общий поток отправки задерживал бы
на его таймаут ещё и всех остальных клиентов.
Владение объектом клиента: создаёт accept-поток, освобождает ТОЛЬКО тик-поток
(ReapClients) и только после того, как клиентский поток честно вышел. Никто
больше клиентов не освобождает — поэтому указатель, взятый под FClientLock,
остаётся валидным, пока тик-поток не сделает следующий проход.
остаётся валидным, пока тик-поток не сделает следующий проход. На остановке
освобождает Stop, но лишь дождавшись выхода ВСЕХ клиентских потоков.
Два счёта соединений. Слот из TCI_MAX_CLIENTS занимает только клиент,
прошедший handshake (Up); сокет до handshake живёт в общем массиве
(TCI_MAX_SOCKETS) и убивается по таймауту TCI_HANDSHAKE_MS. Иначе восемь
молчащих TCP-соединений навсегда закрывали дверь настоящим клиентам.
Бинарные фреймы (потоки IQ/аудио, §3.4) пока не обрабатываются: этап 2,
см. doc/TCI.md. Приходящие от клиента binary-фреймы молча отбрасываются.
@@ -34,6 +42,8 @@ unit TCIServer;
Авторизации у TCI нет by design. Порт слушается там, где сказано в
настройках; умолчание — 127.0.0.1, чтобы наружу он не торчал без спроса.
Отсюда же отказ браузерным клиентам (заголовок Origin): страница, открытая
в браузере, иначе дотянулась бы до петлевого порта и до передатчика.
}
{$IFDEF FPC}
@@ -50,14 +60,17 @@ uses
SyncObjs; // ← после платформенных юнитов (конфликт идентификатора Create)
const
TCI_MAX_CLIENTS = 8;
TCI_MAX_CLIENTS = 8; // прошедших handshake (слоты протокола)
TCI_MAX_SOCKETS = 32; // всего сокетов, включая ещё не поднявшиеся
TCI_TICK_MS = 20; // период OnTick (сенсоры троттлятся адаптером)
TCI_WS_GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';
TCI_SEND_TIMEOUT = 300; // мс на SockSend, иначе клиент считается мёртвым
TCI_HANDSHAKE_MS = 5000; // мс на HTTP-запрос от подключившегося
TCI_POLL_MS = 20; // на столько recv клиента засыпает между кадрами
TCI_OUT_MAX = 4000; // потолок очереди отправки на клиента (строк)
TCI_OUT_CHUNK = 3800; // склейка очереди в один фрейм, символов
TCI_MSG_MAX = 65536; // потолок собираемого из фрагментов сообщения
TCI_STOP_WAIT_MS = 10000; // сколько ждём выхода клиентских потоков в Stop
TCI_STOP_KILL_MS = 500; // как часто добиваем клиентов, ожидая их выхода
type
TTCIServer = class;
@@ -65,12 +78,17 @@ type
{ Один подключённый клиент: WS-сокет, его личные подписки и очередь
отправки. Подписки на сенсоры в TCI индивидуальны (RX_SENSORS_ENABLE
«отправляется только клиентом»), поэтому живут здесь, а не в адаптере.
Параметры потоков (§4.3) — тоже клиентские, их держит адаптер по ссылке
на этот объект. }
Параметры потоков (§4.3) — тоже клиентские.
Подписки и параметры потоков пишет поток клиента, а читает тик-поток,
поэтому и те и другие ходят через FStateLock: набор «включено + период +
последняя отправка» обязан меняться и читаться целиком. }
TTCIClient = class
private
FWs: TWsClient;
FUp: Boolean; // handshake прошёл: клиент занимает слот
FReady: Boolean; // пачка инициализации отправлена
FStateLock: TCriticalSection;
FRxSensors: Boolean;
FRxSensorsMs: Integer;
FRxSensorsAt: QWord; // тик последней отправки
@@ -92,30 +110,46 @@ type
FDead: Boolean; // сокет уже не пишется — гасим соединение
FKilled: Boolean; // shutdown сокета уже сделан
FClosed: Boolean; // клиентский поток вышел (можно освобождать)
function GetReady: Boolean;
procedure SetReady(V: Boolean);
function GetIQRate: Integer; procedure SetIQRate(V: Integer);
function GetAudioRate: Integer; procedure SetAudioRate(V: Integer);
function GetAudioSamples: Integer; procedure SetAudioSamples(V: Integer);
function GetAudioChannels: Integer; procedure SetAudioChannels(V: Integer);
function GetAudioSampleType: string; procedure SetAudioSampleType(const V: string);
function GetTxBuffering: Integer; procedure SetTxBuffering(V: Integer);
public
constructor Create(AWs: TWsClient);
destructor Destroy; override;
{ Строку в очередь клиенту. False — соединение уже мертво. Не блокирует. }
function Send(const S: string): Boolean;
{ Слить очередь в сокет. Зовёт только тик-поток. False — клиент умер. }
{ Слить очередь в сокет. Зовёт ТОЛЬКО собственный поток клиента: запись
может ждать до TCI_SEND_TIMEOUT, и общий поток на этом задерживал бы
всех остальных. False — клиент умер. }
function Flush: Boolean;
{ Пометить мёртвым и разбудить его поток (shutdown сокета). }
procedure Kill;
{ Подписки на измерители — целиком под локом. }
procedure SetRxSensors(On_: Boolean);
procedure SetRxSensorsMs(Ms: Integer);
procedure SetTxSensors(On_: Boolean);
procedure SetTxSensorsMs(Ms: Integer);
{ Пора ли слать измеритель: проверка периода и отметка отправки — один
атомарный шаг, иначе тик-поток и клиентский расходятся в наборе. }
function DueRxSensors(Now_: QWord): Boolean;
function DueTxSensors(Now_: QWord): Boolean;
property Ws: TWsClient read FWs;
property Ready: Boolean read FReady write FReady;
property Up: Boolean read FUp;
property Ready: Boolean read GetReady write SetReady;
property Dead: Boolean read FDead;
property RxSensors: Boolean read FRxSensors write FRxSensors;
property RxSensorsMs: Integer read FRxSensorsMs write FRxSensorsMs;
property RxSensorsAt: QWord read FRxSensorsAt write FRxSensorsAt;
property TxSensors: Boolean read FTxSensors write FTxSensors;
property TxSensorsMs: Integer read FTxSensorsMs write FTxSensorsMs;
property TxSensorsAt: QWord read FTxSensorsAt write FTxSensorsAt;
property IQRate: Integer read FIQRate write FIQRate;
property AudioRate: Integer read FAudioRate write FAudioRate;
property AudioSamples: Integer read FAudioSamples write FAudioSamples;
property AudioChannels: Integer read FAudioChannels write FAudioChannels;
property AudioSampleType: string read FAudioSampleType write FAudioSampleType;
property TxBuffering: Integer read FTxBuffering write FTxBuffering;
property IQRate: Integer read GetIQRate write SetIQRate;
property AudioRate: Integer read GetAudioRate write SetAudioRate;
property AudioSamples: Integer read GetAudioSamples write SetAudioSamples;
property AudioChannels: Integer read GetAudioChannels write SetAudioChannels;
property AudioSampleType: string read GetAudioSampleType write SetAudioSampleType;
property TxBuffering: Integer read GetTxBuffering write SetTxBuffering;
end;
TTCIClientEvent = procedure(Client: TTCIClient) of object;
@@ -124,8 +158,9 @@ type
TTCIServer = class
private
FListenSock: TSocket;
FClients: array[0..TCI_MAX_CLIENTS-1] of TTCIClient;
FClientCount: Integer;
FClients: array[0..TCI_MAX_SOCKETS-1] of TTCIClient;
FClientCount: Integer; // всего сокетов в массиве (с не поднявшимися)
FUpCount: LongInt; // прошедших handshake (Interlocked*)
FClientLock: TCriticalSection;
FAcceptThread: TThread;
FTickThread: TThread;
@@ -140,11 +175,16 @@ type
FOnTick: TThreadMethod;
function InitListen: Boolean;
procedure ReapClients; // освободить клиентов, чьи потоки вышли
procedure FlushClients; // слить очереди в сокеты (вне FClientLock)
procedure KillAll;
procedure Disconnected(Client: TTCIClient); // OnDisconnect, единая точка
public
constructor Create;
destructor Destroy; override;
{ Проверка настроек без побочных эффектов: можно ли вообще открыть такой
слушатель. Зовётся ДО остановки работающего сервера. }
class function ValidSettings(APort: Word; const ABindIP: string): Boolean;
{ Настройка слушателя. Применяется при следующем Start.
False — адрес не разобран (порт не откроется). }
function Configure(APort: Word; const ABindIP: string): Boolean;
@@ -168,6 +208,7 @@ type
procedure AcceptLoop;
procedure TickLoop;
procedure HandleClient(Client: TTCIClient);
function Promote(Client: TTCIClient): Boolean; // handshake прошёл
procedure ThreadDone; // клиентский поток отработал
property Port: Word read FPort;
@@ -187,6 +228,16 @@ type
«ошибка разбора = слушаем всё» в протоколе без авторизации недопустима. }
function TCIParseIPv4(const S: string; out Addr: LongWord): Boolean;
{ Значение HTTP-заголовка (Name — в нижнем регистре, без ':'). Разбор
построчный: точное сравнение подстроки «upgrade: websocket» отвергало
валидные запросы с табуляцией или без пробела после двоеточия. }
function TCIHttpHeader(const Header, Name: string): string;
{ Проверка UTF-8: текстовые кадры WebSocket обязаны быть корректным UTF-8
(RFC 6455 §5.6). Отвергает и оборванные последовательности, и избыточно
длинные формы, и суррогаты, и всё выше U+10FFFF. }
function TCIValidUTF8(const S: string): Boolean;
implementation
type
@@ -251,7 +302,8 @@ begin
end;
finally
// Освобождать себя нельзя: объект переиспользуется рассылкой из чужих
// потоков. Помечаем «поток вышел» — освободит тик-поток (ReapClients).
// потоков. Помечаем «поток вышел» — освободит тик-поток (ReapClients)
// или Stop, который ждёт именно этого.
FClient.FClosed := True;
FServer.ThreadDone;
end;
@@ -290,6 +342,57 @@ begin
Result := True;
end;
function TCIHttpHeader(const Header, Name: string): string;
var
i, Start, P: Integer;
Line, LName: string;
begin
Result := '';
Start := 1;
for i := 1 to Length(Header) + 1 do
if (i > Length(Header)) or (Header[i] = #10) then
begin
Line := Trim(Copy(Header, Start, i - Start)); // Trim снимет и #13
Start := i + 1;
P := Pos(':', Line);
if P <= 1 then Continue;
LName := LowerCase(Trim(Copy(Line, 1, P - 1)));
if LName = Name then
Exit(Trim(Copy(Line, P + 1, MaxInt)));
end;
end;
function TCIValidUTF8(const S: string): Boolean;
var
i, k, N, Len: Integer;
B: Byte;
Cp: LongWord;
begin
i := 1;
Len := Length(S);
while i <= Len do
begin
B := Byte(S[i]);
if B < $80 then begin Inc(i); Continue; end
else if (B >= $C2) and (B <= $DF) then begin N := 1; Cp := B and $1F; end
else if (B >= $E0) and (B <= $EF) then begin N := 2; Cp := B and $0F; end
else if (B >= $F0) and (B <= $F4) then begin N := 3; Cp := B and $07; end
else Exit(False); // $80..$C1 и $F5.. началом последовательности не бывают
if i + N > Len then Exit(False); // оборвано на середине символа
for k := 1 to N do
begin
if (Byte(S[i + k]) and $C0) <> $80 then Exit(False);
Cp := (Cp shl 6) or (Byte(S[i + k]) and $3F);
end;
// Избыточно длинная форма, суррогатная пара и выход за U+10FFFF.
if ((N = 2) and (Cp < $800)) or ((N = 3) and (Cp < $10000)) or
((Cp >= $D800) and (Cp <= $DFFF)) or (Cp > $10FFFF) then Exit(False);
Inc(i, N + 1);
end;
Result := True;
end;
{ ═══════════════════════════════════════════════════════════════════════════
TTCIClient
═══════════════════════════════════════════════════════════════════════════ }
@@ -298,7 +401,9 @@ constructor TTCIClient.Create(AWs: TWsClient);
begin
inherited Create;
FWs := AWs;
FUp := False;
FReady := False;
FStateLock := TCriticalSection.Create;
FRxSensors := False;
FRxSensorsMs := 200;
FTxSensors := False;
@@ -318,9 +423,150 @@ end;
destructor TTCIClient.Destroy;
begin
FOutLock.Free;
FStateLock.Free;
inherited;
end;
function TTCIClient.GetReady: Boolean;
begin
FStateLock.Enter;
try Result := FReady; finally FStateLock.Leave; end;
end;
procedure TTCIClient.SetReady(V: Boolean);
begin
FStateLock.Enter;
try FReady := V; finally FStateLock.Leave; end;
end;
function TTCIClient.GetIQRate: Integer;
begin
FStateLock.Enter;
try Result := FIQRate; finally FStateLock.Leave; end;
end;
procedure TTCIClient.SetIQRate(V: Integer);
begin
FStateLock.Enter;
try FIQRate := V; finally FStateLock.Leave; end;
end;
function TTCIClient.GetAudioRate: Integer;
begin
FStateLock.Enter;
try Result := FAudioRate; finally FStateLock.Leave; end;
end;
procedure TTCIClient.SetAudioRate(V: Integer);
begin
FStateLock.Enter;
try FAudioRate := V; finally FStateLock.Leave; end;
end;
function TTCIClient.GetAudioSamples: Integer;
begin
FStateLock.Enter;
try Result := FAudioSamples; finally FStateLock.Leave; end;
end;
procedure TTCIClient.SetAudioSamples(V: Integer);
begin
FStateLock.Enter;
try FAudioSamples := V; finally FStateLock.Leave; end;
end;
function TTCIClient.GetAudioChannels: Integer;
begin
FStateLock.Enter;
try Result := FAudioChannels; finally FStateLock.Leave; end;
end;
procedure TTCIClient.SetAudioChannels(V: Integer);
begin
FStateLock.Enter;
try FAudioChannels := V; finally FStateLock.Leave; end;
end;
function TTCIClient.GetAudioSampleType: string;
begin
FStateLock.Enter;
try Result := FAudioSampleType; finally FStateLock.Leave; end;
end;
procedure TTCIClient.SetAudioSampleType(const V: string);
begin
FStateLock.Enter;
try FAudioSampleType := V; finally FStateLock.Leave; end;
end;
function TTCIClient.GetTxBuffering: Integer;
begin
FStateLock.Enter;
try Result := FTxBuffering; finally FStateLock.Leave; end;
end;
procedure TTCIClient.SetTxBuffering(V: Integer);
begin
FStateLock.Enter;
try FTxBuffering := V; finally FStateLock.Leave; end;
end;
procedure TTCIClient.SetRxSensors(On_: Boolean);
begin
FStateLock.Enter;
try
FRxSensors := On_;
if On_ then FRxSensorsAt := 0; // первую посылку не ждём период
finally
FStateLock.Leave;
end;
end;
procedure TTCIClient.SetRxSensorsMs(Ms: Integer);
begin
FStateLock.Enter;
try FRxSensorsMs := Ms; finally FStateLock.Leave; end;
end;
procedure TTCIClient.SetTxSensors(On_: Boolean);
begin
FStateLock.Enter;
try
FTxSensors := On_;
if On_ then FTxSensorsAt := 0;
finally
FStateLock.Leave;
end;
end;
procedure TTCIClient.SetTxSensorsMs(Ms: Integer);
begin
FStateLock.Enter;
try FTxSensorsMs := Ms; finally FStateLock.Leave; end;
end;
function TTCIClient.DueRxSensors(Now_: QWord): Boolean;
begin
FStateLock.Enter;
try
Result := FRxSensors and (Now_ - FRxSensorsAt >= QWord(FRxSensorsMs));
if Result then FRxSensorsAt := Now_;
finally
FStateLock.Leave;
end;
end;
function TTCIClient.DueTxSensors(Now_: QWord): Boolean;
begin
FStateLock.Enter;
try
Result := FTxSensors and (Now_ - FTxSensorsAt >= QWord(FTxSensorsMs));
if Result then FTxSensorsAt := Now_;
finally
FStateLock.Leave;
end;
end;
function TTCIClient.Send(const S: string): Boolean;
begin
Result := False;
@@ -423,6 +669,7 @@ begin
{$ENDIF}
FListenSock := SOCK_INVALID;
FClientCount := 0;
FUpCount := 0;
FClientLock := TCriticalSection.Create;
FPort := TCI_DEFAULT_PORT;
FBindIP := '127.0.0.1';
@@ -438,12 +685,17 @@ begin
inherited;
end;
function TTCIServer.Configure(APort: Word; const ABindIP: string): Boolean;
class function TTCIServer.ValidSettings(APort: Word; const ABindIP: string): Boolean;
var Dummy: LongWord;
begin
Result := (APort <> 0) and TCIParseIPv4(ABindIP, Dummy);
end;
function TTCIServer.Configure(APort: Word; const ABindIP: string): Boolean;
begin
FPort := APort;
FBindIP := ABindIP;
Result := (APort <> 0) and TCIParseIPv4(ABindIP, Dummy);
Result := ValidSettings(APort, ABindIP);
end;
function TTCIServer.Running: Boolean;
@@ -513,6 +765,18 @@ begin
Result := True;
end;
procedure TTCIServer.KillAll;
var i: Integer;
begin
FClientLock.Enter;
try
for i := 0 to FClientCount - 1 do
if FClients[i] <> nil then FClients[i].Kill;
finally
FClientLock.Leave;
end;
end;
procedure TTCIServer.Stop;
var i, Waited: Integer;
begin
@@ -529,42 +793,40 @@ begin
FListenSock := SOCK_INVALID;
end;
// Шаг 2: будим клиентские потоки, висящие в recv.
FClientLock.Enter;
try
for i := 0 to FClientCount - 1 do
if FClients[i] <> nil then FClients[i].Kill;
finally
FClientLock.Leave;
end;
// Шаг 3: свои потоки (клиентские — FreeOnTerminate, ждём их отдельно).
// Шаг 2: дожидаемся accept-потока — после него новых клиентов не появится.
if FAcceptThread <> nil then begin FAcceptThread.WaitFor; FreeAndNil(FAcceptThread); end;
if FTickThread <> nil then begin FTickThread.WaitFor; FreeAndNil(FTickThread); end;
// Шаг 4: ждём выхода клиентских потоков. Прокачивая очередь Synchronize:
// Stop зовёт поток контроллера (UI), а клиентский поток может как раз в нём
// висеть на FController.Invoke. Без прокачки это гарантированный взаимный
// клин, а по его истечении — освобождение объекта из-под живого потока.
// Шаг 3: будим клиентские потоки, висящие в recv, и останавливаем тик.
KillAll;
if FTickThread <> nil then begin FTickThread.WaitFor; FreeAndNil(FTickThread); end;
// Шаг 4: ждём выхода клиентских потоков — БЕЗ таймаута. Прокачивая очередь
// Synchronize: Stop зовёт поток контроллера (UI), а клиентский поток может
// как раз в нём висеть на FController.Invoke; без прокачки это взаимный
// клин. Выйти отсюда по таймауту нельзя: следом освобождаются и клиенты, и
// сам сервер с адаптером, а живой поток вернулся бы в эту память.
Waited := 0;
while (FThreadCount > 0) and (Waited < TCI_STOP_WAIT_MS) do
while FThreadCount > 0 do
begin
if GetCurrentThreadId = MainThreadID then CheckSynchronize(5) else Sleep(5);
Inc(Waited, 5);
// Повторный shutdown: клиент мог быть принят между шагом 2 и шагом 3
// (accept уже вернул сокет, поток стартовал позже) и Kill его не застал.
if (Waited mod TCI_STOP_KILL_MS) = 0 then KillAll;
end;
// Шаг 5: зачистка. Если поток всё же не вышел (не должно случаться: сокеты
// закрыты, очередь прокачана), объект НЕ освобождаем — утечка на выходе
// несравнимо дешевле обращения к освобождённой памяти из живого потока.
// Шаг 5: зачистка. Потоков больше нет — освобождать безопасно.
FClientLock.Enter;
try
for i := 0 to FClientCount - 1 do
if (FClients[i] <> nil) and FClients[i].FClosed then
if FClients[i] <> nil then
begin
Disconnected(FClients[i]); // и на остановке тоже: захваты снимаются
FClients[i].Ws.Free;
FreeAndNil(FClients[i]);
end;
FClientCount := 0;
FUpCount := 0;
finally
FClientLock.Leave;
end;
@@ -603,12 +865,16 @@ begin
end;
SockSetSndTimeout(CSock, TCI_SEND_TIMEOUT);
// До конца handshake сокет не должен молчать вечно: иначе горстка пустых
// соединений держала бы место, ничего не сказав.
SockSetRcvTimeout(CSock, TCI_HANDSHAKE_MS);
Client := TTCIClient.Create(TWsClient.Create(CSock));
Full := False;
FClientLock.Enter;
try
// Слот берём под локом: место в массиве освобождает тик-поток.
if FClientCount >= TCI_MAX_CLIENTS then Full := True
// Место в массиве освобождает тик-поток; слот протокола (Up) клиент
// получит позже — после успешного Upgrade (см. Promote).
if FClientCount >= TCI_MAX_SOCKETS then Full := True
else
begin
FClients[FClientCount] := Client;
@@ -630,13 +896,34 @@ begin
end;
end;
function TTCIServer.Promote(Client: TTCIClient): Boolean;
// Слот протокола выдаётся ТОЛЬКО тут — после разбора HTTP-запроса и до ответа
// 101. Считаем поднявшихся: молчащие сокеты слотов не занимают.
var i, N: Integer;
begin
Result := False;
if not FRunning then Exit;
FClientLock.Enter;
try
N := 0;
for i := 0 to FClientCount - 1 do
if (FClients[i] <> nil) and FClients[i].FUp then Inc(N);
if N >= TCI_MAX_CLIENTS then Exit;
Client.FUp := True;
InterLockedIncrement(FUpCount);
Result := True;
finally
FClientLock.Leave;
end;
end;
procedure TTCIServer.ReapClients;
// Освобождение клиентов — единственное место во всей программе. Зовёт только
// Освобождение клиентов — единственное место, кроме Stop. Зовёт только
// тик-поток, поэтому указатель, взятый кем угодно под FClientLock, живёт до
// следующего прохода тика (а вне лока указателей никто не держит).
var
i, j, N: Integer;
Doomed: array[0..TCI_MAX_CLIENTS-1] of TTCIClient;
Doomed: array[0..TCI_MAX_SOCKETS-1] of TTCIClient;
begin
N := 0;
FClientLock.Enter;
@@ -645,6 +932,7 @@ begin
while i < FClientCount do
if (FClients[i] <> nil) and FClients[i].FClosed then
begin
if FClients[i].FUp then InterLockedDecrement(FUpCount);
Doomed[N] := FClients[i];
Inc(N);
for j := i to FClientCount - 2 do FClients[j] := FClients[j + 1];
@@ -659,33 +947,18 @@ begin
for i := 0 to N - 1 do
begin
if Assigned(FOnDisconnect) then FOnDisconnect(Doomed[i]);
Disconnected(Doomed[i]);
Doomed[i].Ws.Free; // закрывает сокет
Doomed[i].Free;
end;
end;
procedure TTCIServer.FlushClients;
// Запись в сокеты — вне FClientLock: медленный клиент не должен держать лок,
// иначе Broadcast из потока контроллера снова начнёт ждать сеть.
var
Snap: array[0..TCI_MAX_CLIENTS-1] of TTCIClient;
i, N: Integer;
procedure TTCIServer.Disconnected(Client: TTCIClient);
// Единственное место, где наверх уходит «клиент ушёл»: и обычное отключение
// (ReapClients), и остановка сервера. Иначе после Stop у адаптера оставались
// висеть захваты параметров ушедших клиентов (§3.5).
begin
N := 0;
FClientLock.Enter;
try
for i := 0 to FClientCount - 1 do
if (FClients[i] <> nil) and not FClients[i].FClosed then
begin
Snap[N] := FClients[i];
Inc(N);
end;
finally
FClientLock.Leave;
end;
for i := 0 to N - 1 do
Snap[i].Flush;
if (Client <> nil) and Assigned(FOnDisconnect) then FOnDisconnect(Client);
end;
{ ═══════════════════════════════════════════════════════════════════════════
@@ -696,12 +969,12 @@ procedure TTCIServer.HandleClient(Client: TTCIClient);
var
Ws: TWsClient;
R, HeaderEnd: Integer;
Header, HeaderLC, Key, AcceptKey, Response, Text: string;
Header, Key, AcceptKey, Response, Text: string;
Raw: array[0..4095] of Byte;
RawLen, Rest: Integer;
B0, B1: Byte;
Masked, Fin, Pending: Boolean;
PayLen, Need, i, j, Consumed, KPos, KEnd: Integer;
PayLen, Need, i, j, Consumed: Integer;
Hi32: LongWord;
Mask: array[0..3] of Byte;
Payload: array of Byte;
@@ -717,6 +990,8 @@ begin
HeaderEnd := 0;
repeat
R := SockRecv(Ws.Socket, @Raw[RawLen], SizeOf(Raw) - RawLen, 0);
// R <= 0 здесь — это и разрыв, и истёкший TCI_HANDSHAKE_MS: молчащее
// соединение уходит само, не занимая место.
if R <= 0 then begin Ws.State := wsClosed; Break; end;
Inc(RawLen, R);
SetLength(Header, RawLen);
@@ -728,10 +1003,10 @@ begin
Consumed := HeaderEnd + 3; // длина заголовков вместе с CRLFCRLF
Header := Copy(Header, 1, Consumed);
HeaderLC := LowerCase(Header);
// Путь не проверяем: клиенты ходят на '/', но протокол его не оговаривает.
if System.Pos('upgrade: websocket', HeaderLC) = 0 then
if (Pos('websocket', LowerCase(TCIHttpHeader(Header, 'upgrade'))) = 0) or
(Pos('upgrade', LowerCase(TCIHttpHeader(Header, 'connection'))) = 0) then
begin
Response := 'HTTP/1.1 426 Upgrade Required'#13#10 +
'Content-Length: 0'#13#10'Connection: close'#13#10#13#10;
@@ -739,15 +1014,20 @@ begin
Exit;
end;
Key := '';
KPos := System.Pos('sec-websocket-key: ', HeaderLC);
if KPos > 0 then
// Браузерный клиент. Origin шлют только браузеры, и он — единственный
// признак, отличающий страницу от нативной программы. Авторизации в TCI
// нет: без этой проверки открытая вкладка с чужого сайта дотянулась бы по
// ws://127.0.0.1:40001 до TRX/TUNE/VFO. Своим web-страницам нужен явный
// прокси, а не дыра по умолчанию.
if TCIHttpHeader(Header, 'origin') <> '' then
begin
Key := Copy(Header, KPos + 19, 100);
KEnd := System.Pos(#13, Key);
if KEnd > 0 then Key := Copy(Key, 1, KEnd - 1);
Key := Trim(Key);
Response := 'HTTP/1.1 403 Forbidden'#13#10 +
'Content-Length: 0'#13#10'Connection: close'#13#10#13#10;
Ws.SendRaw(Response[1], Length(Response));
Exit;
end;
Key := TCIHttpHeader(Header, 'sec-websocket-key');
// Пустой ключ = не WebSocket-клиент (или сломанный): Accept без ключа
// формально считается валидным, и такое «соединение» потом молча висит.
if Key = '' then
@@ -758,6 +1038,16 @@ begin
Exit;
end;
// Слот протокола — до ответа 101: отказать после «Switching Protocols» уже
// некрасиво, клиент считал бы себя подключённым.
if not Promote(Client) then
begin
Response := 'HTTP/1.1 503 Service Unavailable'#13#10 +
'Content-Length: 0'#13#10'Connection: close'#13#10#13#10;
Ws.SendRaw(Response[1], Length(Response));
Exit;
end;
AcceptKey := Base64EncodeBytes(SHA1(Key + TCI_WS_GUID), 20);
Response := 'HTTP/1.1 101 Switching Protocols'#13#10 +
'Upgrade: websocket'#13#10 +
@@ -765,6 +1055,9 @@ begin
'Sec-WebSocket-Accept: ' + AcceptKey + #13#10#13#10;
if not Ws.SendRaw(Response[1], Length(Response)) then Exit;
Ws.State := wsOpen;
// Дальше клиент вправе молчать сколько угодно, но просыпаться нам надо:
// на этом же потоке уходит его очередь отправки (Flush).
SockSetRcvTimeout(Ws.Socket, TCI_POLL_MS);
// Хвост первого пакета: клиент вправе прислать первый WS-фрейм в том же
// сегменте, что и заголовки. Выбросить его — потерять первую команду.
@@ -772,8 +1065,22 @@ begin
if Rest > 0 then Move(Raw[Consumed], Ws.BufData[0], Rest);
Ws.BufLen := Rest;
// Пачка инициализации + текущее состояние (§3.1) — дело адаптера.
if Assigned(FOnConnect) then FOnConnect(Client);
// Пачка инициализации + текущее состояние (§3.1) — дело адаптера. Под
// FClientLock: пока она набирается, рассылка обязана ждать. Иначе изменение,
// случившееся после строки снимка, но до Ready=True, пропадало навсегда —
// Broadcast пропускает не-Ready клиента, и тот оставался со старым значением,
// считая инициализацию завершённой. Лок держится только на укладку строк в
// очередь (сеть тут не пишется), но обработчик OnConnect по этой же причине
// НЕ имеет права звать Invoke в поток контроллера: тот может ждать этот лок.
if Assigned(FOnConnect) then
begin
FClientLock.Enter;
try
FOnConnect(Client);
finally
FClientLock.Leave;
end;
end;
// ── Цикл WS-сообщений ────────────────────────────────────────────────────
Cmds := TStringList.Create;
@@ -786,7 +1093,9 @@ begin
if not Pending then
begin
R := Ws.Recv;
if R <= 0 then Break;
// R <= 0 — либо разрыв, либо просто истёк TCI_POLL_MS. Второе штатно:
// просыпаемся, чтобы отдать накопившуюся очередь.
if (R <= 0) and not SockRecvTimedOut then Break;
end;
Pending := False;
@@ -902,6 +1211,14 @@ begin
if Fin then
begin
// Текстовое сообщение обязано быть валидным UTF-8 (§5.6);
// битую последовательность RFC велит закрывать, а не молча
// скармливать разбору команд.
if (MsgOp = $01) and not TCIValidUTF8(Frag) then
begin
Ws.State := wsClosed;
Break;
end;
if (MsgOp = $01) and Assigned(FOnCommand) and (Frag <> '') then
begin
TCISplit(Frag, Cmds);
@@ -912,8 +1229,15 @@ begin
Frag := '';
end;
end;
$08: // close
$08: // close: RFC 6455 §5.5.1 требует ответить своим close-кадром
begin
// Полезная нагрузка close — либо пустая, либо код (2 байта) плюс
// причина. Ровно один байт невалиден: отвечать на такое нечем.
if PayLen = 1 then begin Ws.State := wsClosed; Break; end;
// В ответе — только код: причину повторять не обязаны (§5.5.1),
// а чужой текст мы наружу не пересылаем.
if PayLen >= 2 then Ws.SendWsFrame($08, Payload[0], 2)
else Ws.SendWsFrame($08, PayLen, 0);
Ws.State := wsClosed;
Break;
end;
@@ -927,6 +1251,11 @@ begin
Break;
end;
end;
// Очередь — в сокет здесь же, на потоке этого клиента: ответы на только
// что разобранные команды уходят сразу, а медленный клиент задерживает
// только себя (тик-поток очереди лишь наполняет).
if not Client.Flush then Break;
end;
finally
Cmds.Free;
@@ -960,7 +1289,8 @@ begin
FClientLock.Enter;
try
for i := 0 to FClientCount - 1 do
if (FClients[i] <> nil) and not FClients[i].FClosed then Proc(FClients[i]);
if (FClients[i] <> nil) and FClients[i].FUp and not FClients[i].FClosed then
Proc(FClients[i]);
finally
FClientLock.Leave;
end;
@@ -973,7 +1303,7 @@ end;
function TTCIServer.ClientCount: Integer;
begin
Result := FClientCount;
Result := FUpCount;
end;
procedure TTCIServer.TickLoop;
@@ -983,8 +1313,10 @@ begin
Sleep(TCI_TICK_MS);
if not FRunning then Break;
ReapClients; // отключившиеся — освобождаем только здесь
if (FClientCount > 0) and Assigned(FOnTick) then FOnTick;
FlushClients; // очереди → сокеты
if (FUpCount > 0) and Assigned(FOnTick) then FOnTick;
// В сокеты пишет каждый клиент сам, на своём потоке (см. HandleClient):
// общий поток отправки означал бы, что один медленный клиент задерживает
// очередь всех остальных на свой таймаут записи.
end;
end;