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

fix(tci): писатель WAV — один поток с очередью, файл публикуется атомарно

Browse Source 
SAVE запускал поток на каждую запись с FreeOnTerminate: его никто не держал
и никто не ждал. Замерено отдельным процессом — при штатном выходе сразу
после сохранения от ожидаемых 100000044 байт на диске оставалось 40960, а
заголовок заявлял полную длину; на медленном каталоге идущие подряд SAVE
плодили сотни потоков, чья память стеков в 128-МБ бюджет не входила.

Теперь писатель один и принадлежит адаптеру (лениво на первом SAVE), очередь
ограничена TCI_RECORD_MAX_JOBS = 16 (сверх — клиенту writer busy и возврат
резерва), а деструктор адаптера гасит его через TCIStopWriter: Close →
WaitDrained (без срока) → Free. Срока здесь нет намеренно: поток, стоящий в
write/fsync, изнутри процесса не останавливается (Terminate не указ, Free
обязан WaitFor, бросить живой TThread нельзя — он ходит в общий бюджет),
поэтому срок не ограничивал выход, а только терял подтверждённые клиенту
записи. Ограниченный выход = писатель отдельным процессом, одним TThread не
делается; это записано в коде и в доке.

Результат записи больше не игнорируется: FileWrite возвращает число байт и
при ошибке даёт 0/-1 без исключения, поэтому на полном диске файл спокойно
дописывался до конца огрызком. TCIWriteAll — цикл с проверкой каждого вызова,
плюс FileFlush перед публикацией (на ext4 с отложенным размещением ENOSPC
приходит именно там).

Файл появляется под целевым именем целиком или не появляется вовсе: данные
пишутся во временный файл рядом (эксклюзивно, не по симлинку), а публикует
их TCIPublishFile — renameat2(RENAME_NOREPLACE) напрямую через Do_SysCall,
если его нет — link + unlink, если нет и ссылок (FAT/exFAT, часть CIFS/SMB и
FUSE) — отказ с сохранением данных в .part. FileExists + rename не делается
нигде: это тот самый TOCTOU. На Windows — MoveFileW без REPLACE_EXISTING.

doc/TCI.md: §2.5 переписан (писатель-очередь, остановка, публикация); заодно
исправлено устаревшее описание склейки кусков в Take (её нет с 7aae0fd).

Стенд test/tci: 214 проверок (было 199), все зелёные. Новое — временный файл
убирается после удачи, неудачная запись не оставляет ни файла, ни .part, за
всё время записи 32 МБ целевое имя ни разу не видно незаконченным, отказ
сверх потолка заданий, после Close заданий не берут, TCIStopWriter дожидается
и самой записи, и хвоста очереди за ней. Все новые гарантии прогнаны
негативным контролем; путь link проверен сборкой с выключенным renameat2,
сам renameat2 — под strace.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
ew8bak
2026-08-19 19:40:31 +03:00
co-authored by Claude Opus 5
parent 7aae0fdfcd
commit 3dc7035f9e
4 changed files with 734 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TCIAdapter.pas
+34 -2
View File
@@ -146,6 +146,7 @@ type
FStreamLock: TCriticalSection; FStreamLock: TCriticalSection;
FStreams: array of TTCIStreamOut; FStreams: array of TTCIStreamOut;
FRec: array[0..TCI_MAX_RX-1] of TTCIRecorder; FRec: array[0..TCI_MAX_RX-1] of TTCIRecorder;
FWriter: TTCIWavWriter; // писатель WAV: один поток с очередью
FTapsOn: Boolean; // тапы навешены на контроллер/движок FTapsOn: Boolean; // тапы навешены на контроллер/движок
// ── TX-аудио от клиента (§3.4) ── // ── TX-аудио от клиента (§3.4) ──
@@ -236,6 +237,8 @@ type
procedure PushTxChrono; // тик: маркеры времени клиенту procedure PushTxChrono; // тик: маркеры времени клиенту
procedure CmdStream(Client: TTCIClient; const M: TTCIMessage); procedure CmdStream(Client: TTCIClient; const M: TTCIMessage);
procedure CmdRecorder(Client: TTCIClient; const M: TTCIMessage); procedure CmdRecorder(Client: TTCIClient; const M: TTCIMessage);
function EnqueueWav(const APath: string; const R: TTCIRecTake;
ARate: Integer): Boolean;
procedure DropClientRecorders(C: TTCIClient); // клиент ушёл — и запись с ним procedure DropClientRecorders(C: TTCIClient); // клиент ушёл — и запись с ним
procedure SweepRecorders; // тик: истёкшие окна записи procedure SweepRecorders; // тик: истёкшие окна записи
function RecordDir: string; // каталог, куда пишем WAV function RecordDir: string; // каталог, куда пишем WAV
@@ -390,6 +393,7 @@ begin
FStreamLock := TCriticalSection.Create; FStreamLock := TCriticalSection.Create;
FTxLock := TCriticalSection.Create; FTxLock := TCriticalSection.Create;
FWriter := nil; // заводится на первом SAVE (см. EnqueueWav)
FTapsOn := False; FTapsOn := False;
SetLength(FTxRaw, TCI_STREAM_DATA_MAX div 2); // худший случай: int16 SetLength(FTxRaw, TCI_STREAM_DATA_MAX div 2); // худший случай: int16
SetLength(FTxMono, TCI_STREAM_DATA_MAX div 2); SetLength(FTxMono, TCI_STREAM_DATA_MAX div 2);
@@ -426,6 +430,11 @@ begin
end; end;
// Потоки — уже после остановки сервера: их объекты ссылаются на клиентов. // Потоки — уже после остановки сервера: их объекты ссылаются на клиентов.
StopAllStreams; StopAllStreams;
// ★Писателя гасим ЗДЕСЬ и с ожиданием: раньше SAVE запускал ничей поток с
// FreeOnTerminate, и штатный выход из программы обрывал запись на полуслове
// (из 100000044 байт на диске оставалось 40960). Ждём ВСЮ очередь без срока:
// за каждую запись в ней клиенту сказано «сохранено» (см. TCIStopWriter).
TCIStopWriter(FWriter);
FreeAndNil(FTxInterp); FreeAndNil(FTxInterp);
FLock.Free; FLock.Free;
FEchoLock.Free; FEchoLock.Free;
@@ -3115,8 +3124,31 @@ begin
Exit; Exit;
end; end;
// Пишет отдельный поток: файл может быть в десятки мегабайт, а мы сейчас в // Пишет отдельный поток: файл может быть в десятки мегабайт, а мы сейчас в
// потоке клиента, который в это время не читает свой сокет. // потоке клиента, который в это время не читает свой сокет. ★Поток ОДИН и
TTCIWavWriter.Create(Path, Rec, Rate); // принадлежит адаптеру: очередь заданий ограничена, а при закрытии
// программы её дожидаются (см. TTCIWavWriter).
if not EnqueueWav(Path, Rec, Rate) then
begin
// Очередь полна (медленный диск) — запись отдать некому, значит и её
// место в бюджете держать больше незачем.
TCIRecBudgetFree(Rec.Reserved);
Reply(Client, TCIBuild('tci_error', [LowerCase(M.Name), 'writer busy']));
end;
end;
function TTCIAdapter.EnqueueWav(const APath: string; const R: TTCIRecTake;
ARate: Integer): Boolean;
// Писателя заводим на первом сохранении: большинству операторов рекордер не
// нужен вовсе, и держать ради них спящий поток незачем. Зовут из потока
// клиента, поэтому создание — под FStreamLock.
begin
FStreamLock.Enter;
try
if FWriter = nil then FWriter := TTCIWavWriter.Create;
finally
FStreamLock.Leave;
end;
Result := FWriter.Enqueue(APath, R, ARate);
end; end;
procedure TTCIAdapter.DropClientRecorders(C: TTCIClient); procedure TTCIAdapter.DropClientRecorders(C: TTCIClient);
TCIStreams.pas
+410 -75
View File
@@ -37,6 +37,8 @@ uses
// ловилось в DX-кластере. Порядок здесь и есть лечение. // ловилось в DX-кластере. Порядок здесь и есть лечение.
{$IFDEF WINDOWS}Windows,{$ENDIF} {$IFDEF WINDOWS}Windows,{$ENDIF}
{$IFDEF UNIX}BaseUnix,{$ENDIF} {$IFDEF UNIX}BaseUnix,{$ENDIF}
// renameat2 обёртки в RTL нет, а нужен именно он (см. TCIPublishFile).
{$IFDEF LINUX}syscall,{$ENDIF}
Classes, SysUtils, Math, SyncObjs, TCIProtocol, TCIServer; Classes, SysUtils, Math, SyncObjs, TCIProtocol, TCIServer;
const const
@@ -48,6 +50,12 @@ const
// заведены под него в конструкторе: SetLength в DSP-потоке на каждый блок // заведены под него в конструкторе: SetLength в DSP-потоке на каждый блок
// аудио — это тысячи обращений к куче в секунду на ровном месте. // аудио — это тысячи обращений к куче в секунду на ровном месте.
TCI_FEED_CHUNK = 4096; TCI_FEED_CHUNK = 4096;
// ★Очередь писателя WAV. Заданий больше этого числа не копим: клиент
// услышит отказ, а не будет молча наполнять память сохранениями, которые
// медленный диск разгребёт неизвестно когда. Каждое задание держит своё
// место в общем бюджете (TCI_RECORD_MAX_BYTES), так что памятью очередь
// ограничена и без счётчика — этот потолок про сами задания.
TCI_RECORD_MAX_JOBS = 16;
type type
{ Накопленная запись: 16-битный PCM с чередованием L/R. } { Накопленная запись: 16-битный PCM с чередованием L/R. }
@@ -206,8 +214,9 @@ type
молчащий, мёртвый или замьюченный приёмник не занимает ничего, а растущая молчащий, мёртвый или замьюченный приёмник не занимает ничего, а растущая
запись спрашивает разрешения на каждый кусок у общего бюджета запись спрашивает разрешения на каждый кусок у общего бюджета
TCI_RECORD_MAX_BYTES. Куски не перевыделяются (никакого realloc в TCI_RECORD_MAX_BYTES. Куски не перевыделяются (никакого realloc в
DSP-потоке) и склеиваются один раз, в Take — а он идёт уже после того, как DSP-потоке) и НЕ склеиваются вовсе: Take отдаёт их писателю как есть, а тот
рекордер вынут из таблицы, то есть без DSP-потока на плечах. } пишет их в файл подряд — сплошная копия удваивала бы пик памяти ровно
там, где её меньше всего. }
TTCIRecorder = class TTCIRecorder = class
private private
FLock: TCriticalSection; FLock: TCriticalSection;
@@ -243,23 +252,63 @@ type
property Owner: TObject read FOwner; property Owner: TObject read FOwner;
end; end;
{ Писатель WAV в своём потоке: файл до 60 МБ, а зовут сохранение из тика { Одно задание писателю: куда писать и что. }
сервера — блокировать его на секунду диска нельзя. Данные забирает себе TTCIRecJob = record
КУСКАМИ, вместе с их местом в общем бюджете (TTCIRecTake.Reserved), и Path: string;
отпускает его, только когда они больше не нужны. Пока писатель ждёт Rec: TTCIRecTake;
медленный диск, эти байты остаются занятыми — так потолок и держит Rate: Integer;
очередь сохранений, а не только сами записи. } end;
{ Писатель WAV: ОДИН поток с очередью, которым владеет адаптер. Пишет не
вызывающий (файл до 60 МБ, а зовут сохранение из потока клиента), но и не
кто попало.
★Раньше на каждый SAVE заводился отдельный поток с FreeOnTerminate: его
никто не держал и никто не ждал. Отсюда две беды.
(1) Штатный выход из программы обрывал запись на полуслове — замерено: из
ожидаемых 100000044 байт на диске оставалось 40960 (то, что успел
сбросить буфер ФС), а заголовок при этом заявлял полную длину.
(2) На медленном или зависшем каталоге идущие подряд SAVE плодили сотни
потоков, и память их стеков в 128-МБ бюджет не входила вовсе.
Теперь очередь ограничена (TCI_RECORD_MAX_JOBS — дальше клиент слышит
отказ), поток один, а адаптер при своей гибели дожидается ВСЕЙ очереди и
делает WaitFor (TCIStopWriter) — ни брошенных потоков, ни выброшенных
записей после нас не остаётся. Данные каждого задания держат своё место в
общем бюджете (TTCIRecTake.Reserved) до конца записи — так потолок держит
и очередь сохранений, а не только сами записи. }
TTCIWavWriter = class(TThread) TTCIWavWriter = class(TThread)
private private
FPath: string; FLock: TCriticalSection;
FRec: TTCIRecTake; FWake: PRTLEvent;
FRate: Integer; FJobs: array of TTCIRecJob; // очередь, FIFO
procedure ReleaseData; FBusy: Boolean; // задание на руках у Execute
FClosed: Boolean; // идёт остановка: новых не принимаем
function Pop(out J: TTCIRecJob): Boolean;
procedure Done;
procedure WriteJob(const J: TTCIRecJob);
class procedure ReleaseJob(var J: TTCIRecJob);
protected protected
procedure Execute; override; procedure Execute; override;
public public
constructor Create(const APath: string; const ARec: TTCIRecTake; constructor Create;
ARateHz: Integer); destructor Destroy; override;
{ Поставить запись в очередь. False — очередь полна или писатель гасится;
тогда куски и их место в бюджете остаются на вызывающем (он обязан
вернуть Reserved сам). }
function Enqueue(const APath: string; const ARec: TTCIRecTake;
ARateHz: Integer): Boolean;
{ Сколько заданий ещё не дописано (вместе с тем, что пишется сейчас). }
function Pending: Integer;
{ Больше не принимать заданий (первый шаг остановки). }
procedure Close;
{ Ждать, пока очередь опустеет — БЕЗ СРОКА. Всякий срок здесь оказался
обманом: остановить поток, стоящий в write/fsync, изнутри процесса всё
равно нечем (Terminate ему не указ, а Free обязан сделать WaitFor),
поэтому срок не спасал от зависания — он только терял хвост очереди на
медленном каталоге, который потом оживал. За каждую запись в очереди
клиенту уже сказано «сохранено», так что ждём столько, сколько нужно. }
procedure WaitDrained;
end; end;
{ Коэффициент прореживания SrcRate → WantRate: наибольший целый делитель, { Коэффициент прореживания SrcRate → WantRate: наибольший целый делитель,
@@ -304,6 +353,13 @@ function TCIRecordPath(const BaseDir, Req: string): string;
имя приходит из сети. THandle(-1) — не вышло. } имя приходит из сети. THandle(-1) — не вышло. }
function TCICreateNewFile(const Path: string): THandle; function TCICreateNewFile(const Path: string): THandle;
{ Погасить писателя: закрыть приём заданий, дождаться ВСЕЙ очереди и
освободить объект (Destroy = Terminate + WaitFor). Зовут при гибели
адаптера, ради этого писатель и стал управляемым. W обнуляется в любом
случае.
★Срока здесь нет намеренно — см. WaitDrained и комментарий в реализации. }
procedure TCIStopWriter(var W: TTCIWavWriter);
implementation implementation
{ ═══════════════════════════════════════════════════════════════════════════ { ═══════════════════════════════════════════════════════════════════════════
@@ -903,38 +959,282 @@ end;
WAV WAV
═══════════════════════════════════════════════════════════════════════════ } ═══════════════════════════════════════════════════════════════════════════ }
constructor TTCIWavWriter.Create(const APath: string; { ─── вспомогательное для записи файла ──────────────────────────────────── }
const ARec: TTCIRecTake; ARateHz: Integer);
var
RecTmpSeq: LongInt = 0;
function TCIRecTempName(const Path: string): string;
// Имя временного файла. Уникальное: застрявший от прошлого падения «.part» не
// должен запирать сохранение под тем же именем навсегда.
begin
Result := Format('%s.%d-%d.part',
[Path, Integer(GetProcessID), InterLockedIncrement(RecTmpSeq)]);
end;
function TCIWriteAll(H: THandle; const Buf; Count: Integer): Boolean;
// ★Результат записи проверяем, и не «= Count», а циклом. FileWrite (как и
// THandleStream.Write под ним) возвращает ЧИСЛО записанных байт и при ошибке
// отдаёт 0 или -1, не поднимая исключения: на полном диске прежний код
// спокойно дописывал WAV до конца, оставляя огрызок с заголовком на полную
// длину. Короткая запись без ошибки тоже законна (сигнал, лимит ФС) — её
// дописываем, а не считаем провалом.
var
P: PByte;
n: Integer;
begin
P := @Buf;
while Count > 0 do
begin
n := FileWrite(H, P^, Count);
if n <= 0 then Exit(False);
Inc(P, n);
Dec(Count, n);
end;
Result := True;
end;
{ Чем кончилась публикация: легло под целевым именем / имя занято /
публиковать нечем — на этой ФС нет ни одной безопасной операции. }
type
TTCIPubResult = (tpubDone, tpubTaken, tpubNoWay);
{$IF DEFINED(LINUX) and (DEFINED(CPUX86_64) or DEFINED(CPUI386) or
DEFINED(CPUAARCH64) or DEFINED(CPUARM))}
{$DEFINE TCI_HAS_RENAMEAT2}
{$IFEND}
{$IFDEF TCI_HAS_RENAMEAT2}
const
// Обёртки в RTL нет, зовём напрямую. Номера — стабильная часть ABI ядра.
{$IFDEF CPUX86_64} TCI_SYS_RENAMEAT2 = 316; {$ENDIF}
{$IFDEF CPUI386} TCI_SYS_RENAMEAT2 = 353; {$ENDIF}
{$IFDEF CPUAARCH64} TCI_SYS_RENAMEAT2 = 276; {$ENDIF}
{$IFDEF CPUARM} TCI_SYS_RENAMEAT2 = 382; {$ENDIF}
TCI_AT_FDCWD = -100;
TCI_RENAME_NOREPLACE = 1;
{$ENDIF}
{$IFDEF UNIX}
function TCIRenameNoReplace(const Src, Dst: string; out Err: Integer): Boolean;
// «Переименовать, если имя свободно» — одним вызовом ядра. Нет такого вызова
// (старое ядро, другая архитектура, ФС не умеет флаг) — Err = ENOSYS/EINVAL, и
// вызывающий идёт дальше по списку.
begin
Result := False;
Err := ESysENOSYS;
{$IFDEF TCI_HAS_RENAMEAT2}
if Do_SysCall(TSysParam(TCI_SYS_RENAMEAT2),
TSysParam(TCI_AT_FDCWD), TSysParam(PtrUInt(PChar(Src))),
TSysParam(TCI_AT_FDCWD), TSysParam(PtrUInt(PChar(Dst))),
TSysParam(TCI_RENAME_NOREPLACE)) = 0 then
Exit(True);
Err := fpGetErrno;
{$ENDIF}
end;
{$ENDIF}
function TCIPublishFile(const Src, Dst: string): TTCIPubResult;
// ★Публикация готового файла: целевое имя появляется ОДНИМ вызовом ядра, сразу
// с полным содержимым, и только если оно свободно. Замены нет ни в каком виде.
//
// Портируемо получить сразу три свойства — атомарное появление, запрет замены
// и работу на любой ФС — нельзя, поэтому идём по списку и на последнем шаге
// честно отказываемся:
// 1) renameat2(RENAME_NOREPLACE) — то, что нужно, одним вызовом;
// 2) нет его — link(2) + unlink: новое имя обязано не существовать (EEXIST),
// на симлинк по этому имени link тоже не пойдёт. Каталог у Src и Dst один
// (записи лежат в tci.record_dir), так что EXDEV тут не бывает;
// 3) нет и жёстких ссылок (FAT/exFAT, часть CIFS/SMB и FUSE) — публиковать
// нечем. ★FileExists + rename здесь НЕ годится: rename затирает то, что
// лежит по имени сейчас, а между проверкой и переносом туда может попасть
// что угодно — это ровно то окно (TOCTOU), ради закрытия которого всё и
// затевалось. Отвечаем tpubNoWay; данные при этом не пропадают — писатель
// оставляет их во временном файле (см. WriteJob).
// На Windows MoveFileW без MOVEFILE_REPLACE_EXISTING уже обладает нужной
// семантикой: существующее имя = отказ.
{$IFDEF UNIX}
var Err: Integer;
{$ENDIF}
begin
{$IFDEF UNIX}
if TCIRenameNoReplace(Src, Dst, Err) then Exit(tpubDone);
if Err = ESysEEXIST then Exit(tpubTaken);
if FpLink(PChar(Src), PChar(Dst)) = 0 then
begin
// Временное имя убираем: данные уже живут под целевым (это тот же inode).
FpUnlink(PChar(Src));
Exit(tpubDone);
end;
if fpGetErrno = ESysEEXIST then Exit(tpubTaken);
Result := tpubNoWay;
{$ELSE}
// WINBOOL — не Boolean: сравнение делает приведение явным.
if MoveFileW(PWideChar(UnicodeString(Src)),
PWideChar(UnicodeString(Dst))) <> False then
Exit(tpubDone);
if (GetLastError = ERROR_ALREADY_EXISTS) or
(GetLastError = ERROR_FILE_EXISTS) then
Exit(tpubTaken);
Result := tpubNoWay;
{$ENDIF}
end;
{ ─── писатель ─────────────────────────────────────────────────────────── }
constructor TTCIWavWriter.Create;
begin begin
inherited Create(True); inherited Create(True);
FreeOnTerminate := True; FLock := TCriticalSection.Create;
FPath := APath; FWake := RTLEventCreate;
FRec := ARec;
FRate := ARateHz;
Start; Start;
end; end;
procedure TTCIWavWriter.ReleaseData; destructor TTCIWavWriter.Destroy;
var J: TTCIRecJob;
begin
Terminate;
RTLEventSetEvent(FWake);
WaitFor;
// Недописанное (гасят, не дождавшись очереди) — данные всё равно вернуть в
// бюджет: объект умирает, а счётчик общий и переживёт нас.
while Pop(J) do ReleaseJob(J);
Done;
RTLEventDestroy(FWake);
FLock.Free;
inherited;
end;
class procedure TTCIWavWriter.ReleaseJob(var J: TTCIRecJob);
// Данные и их место в бюджете уходят вместе и ровно один раз — сколько бы // Данные и их место в бюджете уходят вместе и ровно один раз — сколько бы
// путей выхода ни было у Execute. // путей выхода ни было у записи.
var i: Integer; var i: Integer;
begin begin
for i := 0 to High(FRec.Chunks) do FRec.Chunks[i] := nil; for i := 0 to High(J.Rec.Chunks) do J.Rec.Chunks[i] := nil;
FRec.Chunks := nil; J.Rec.Chunks := nil;
TCIRecBudgetFree(FRec.Reserved); TCIRecBudgetFree(J.Rec.Reserved);
FRec.Reserved := 0; J.Rec.Reserved := 0;
J.Rec.Count := 0;
end;
function TTCIWavWriter.Enqueue(const APath: string; const ARec: TTCIRecTake;
ARateHz: Integer): Boolean;
var n: Integer;
begin
Result := False;
FLock.Enter;
try
if FClosed or Terminated then Exit;
n := Length(FJobs);
if n >= TCI_RECORD_MAX_JOBS then Exit;
SetLength(FJobs, n + 1);
FJobs[n].Path := APath;
FJobs[n].Rec := ARec;
FJobs[n].Rate := ARateHz;
Result := True;
finally
FLock.Leave;
end;
RTLEventSetEvent(FWake);
end;
function TTCIWavWriter.Pop(out J: TTCIRecJob): Boolean;
var i: Integer;
begin
Result := False;
FillChar(J.Rec, SizeOf(J.Rec), 0);
J.Path := '';
J.Rate := 0;
FLock.Enter;
try
if Length(FJobs) = 0 then Exit;
J := FJobs[0];
for i := 1 to High(FJobs) do FJobs[i - 1] := FJobs[i];
FJobs[High(FJobs)].Path := ''; // строку из хвоста отпускаем
FJobs[High(FJobs)].Rec.Chunks := nil;
SetLength(FJobs, Length(FJobs) - 1);
FBusy := True;
Result := True;
finally
FLock.Leave;
end;
end;
procedure TTCIWavWriter.Done;
begin
FLock.Enter;
try
FBusy := False;
finally
FLock.Leave;
end;
end;
function TTCIWavWriter.Pending: Integer;
begin
FLock.Enter;
try
Result := Length(FJobs);
if FBusy then Inc(Result);
finally
FLock.Leave;
end;
end;
procedure TTCIWavWriter.Close;
begin
FLock.Enter;
try
FClosed := True;
finally
FLock.Leave;
end;
end;
procedure TTCIWavWriter.WaitDrained;
begin
while Pending <> 0 do Sleep(5);
end; end;
procedure TTCIWavWriter.Execute; procedure TTCIWavWriter.Execute;
var J: TTCIRecJob;
begin
while True do
begin
if Pop(J) then
begin
try
// ★Взятое из очереди пишем ВСЕГДА, даже если уже идёт остановка: за
// каждое задание клиенту сказано «сохранено», и молча выбросить его
// нельзя. Остановка до этого места и не доходит — TCIStopWriter
// сперва дожидается пустой очереди.
WriteJob(J);
except
// Исключение из потока утащило бы за собой процесс. Сказать о беде
// всё равно некому: SAVE давно подтверждён клиенту.
end;
ReleaseJob(J);
Done;
Continue;
end;
if Terminated then Break;
// Таймаут, а не голое ожидание: Terminate между Pop и сюда не потеряется.
RTLEventWaitFor(FWake, 200);
end;
end;
procedure TTCIWavWriter.WriteJob(const J: TTCIRecJob);
// Заголовок собираем в буфере: WAV — это фиксированные 44 байта, и городить // Заголовок собираем в буфере: WAV — это фиксированные 44 байта, и городить
// два десятка отдельных Write ради них незачем (а строковые литералы в // два десятка отдельных записей ради них незачем (а строковые литералы в
// нетипизированный Write в FPC ещё и передаются не тем, чем кажется). // нетипизированный Write в FPC ещё и передаются не тем, чем кажутся).
var var
FS: THandleStream; HT: THandle;
H: THandle; Tmp: string;
Hdr: array[0..43] of Byte; Hdr: array[0..43] of Byte;
DataBytes: LongWord; DataBytes: LongWord;
i, n, Left: Integer; i, n, Left: Integer;
Ok: Boolean;
Pub: TTCIPubResult;
procedure PutTag(Ofs: Integer; const Tag: string); procedure PutTag(Ofs: Integer; const Tag: string);
var i: Integer; var i: Integer;
@@ -954,57 +1254,92 @@ var
end; end;
begin begin
try DataBytes := LongWord(Int64(J.Rec.Count) * 2 * SizeOf(SmallInt));
DataBytes := LongWord(Int64(FRec.Count) * 2 * SizeOf(SmallInt)); FillChar(Hdr, SizeOf(Hdr), 0);
FillChar(Hdr, SizeOf(Hdr), 0); PutTag(0, 'RIFF');
PutTag(0, 'RIFF'); PutU32(4, 36 + DataBytes);
PutU32(4, 36 + DataBytes); PutTag(8, 'WAVE');
PutTag(8, 'WAVE'); PutTag(12, 'fmt ');
PutTag(12, 'fmt '); PutU32(16, 16); // размер fmt-блока
PutU32(16, 16); // размер fmt-блока PutU16(20, 1); // PCM
PutU16(20, 1); // PCM PutU16(22, 2); // каналов
PutU16(22, 2); // каналов PutU32(24, LongWord(J.Rate));
PutU32(24, LongWord(FRate)); PutU32(28, LongWord(J.Rate) * 2 * 2); // байт в секунду
PutU32(28, LongWord(FRate) * 2 * 2); // байт в секунду PutU16(32, 4); // выравнивание блока
PutU16(32, 4); // выравнивание блока PutU16(34, 16); // бит на сэмпл
PutU16(34, 16); // бит на сэмпл PutTag(36, 'data');
PutTag(36, 'data'); PutU32(40, DataBytes);
PutU32(40, DataBytes);
// ★Не TFileStream/fmCreate: имя пришло из сети, и затирать им чужой файл // ★Целевого имени до конца записи не существует ВООБЩЕ. Пишем во временный
// нельзя. TCICreateNewFile создаёт только новый и не идёт по симлинку. // файл рядом (эксклюзивно и не по симлинку — имя пришло из сети), и только
H := TCICreateNewFile(FPath); // когда всё сошлось, публикуем его под целевым именем одним вызовом ядра.
// Не вышло (файл уже есть, нет прав, нет каталога) — просто уходим: выйти // Так клиент, увидевший файл, всегда прав: раньше по имени сначала
// отсюда через Exit нельзя, ниже ещё возврат памяти под данные. // появлялась пустышка на 0 байт, и на медленном диске её было видно всю
if H <> THandle(-1) then // запись, а до того — огрызок с заголовком на полную длину.
begin Ok := False;
FS := THandleStream.Create(H); Tmp := TCIRecTempName(J.Path);
try HT := TCICreateNewFile(Tmp);
FS.Write(Hdr[0], SizeOf(Hdr)); if HT <> THandle(-1) then
// Куски пишем подряд: в WAV сэмплы и так лежат встык, а хвост begin
// последнего куска за FRec.Count — не наши данные. try
Left := FRec.Count; Ok := TCIWriteAll(HT, Hdr, SizeOf(Hdr));
for i := 0 to High(FRec.Chunks) do Left := J.Rec.Count;
begin i := 0;
if Left <= 0 then Break; // Куски пишем подряд: в WAV сэмплы и так лежат встык, а хвост
n := FRec.Chunk; // последнего куска за Count — не наши данные.
if n > Left then n := Left; while Ok and (Left > 0) and (i <= High(J.Rec.Chunks)) do
FS.Write(FRec.Chunks[i][0], n * 2 * SizeOf(SmallInt)); begin
Dec(Left, n); n := J.Rec.Chunk;
end; if n > Left then n := Left;
finally Ok := TCIWriteAll(HT, J.Rec.Chunks[i][0], n * 2 * SizeOf(SmallInt));
FS.Free; Dec(Left, n);
FileClose(H); Inc(i);
end; end;
// Данных меньше, чем обещано заголовком, — это тот же огрызок.
if Left > 0 then Ok := False;
// ★Сброс на диск ДО переименования: на ext4 с отложенным размещением
// «нет места» приходит не в write, а вот здесь.
if Ok then Ok := FileFlush(HT);
finally
FileClose(HT);
end; end;
except // Публикуем только целое.
// Записать не вышло (нет прав, нет каталога, диск полон) — сказать об этом Pub := tpubTaken;
// клиенту уже некому: команда давно подтверждена. Молчим, но и не падаем: if Ok then Pub := TCIPublishFile(Tmp, J.Path);
// исключение из потока утащило бы за собой процесс. // Не дописали или имя занято — ни огрызка, ни временного файла.
// ★А вот tpubNoWay (на этой ФС публиковать нечем) временный файл ОСТАВЛЯЕТ:
// сказать клиенту уже нечем (SAVE подтверждён давно), и молча стереть его
// десятки мегабайт из-за нашей неспособности переименовать — хуже, чем
// оставить их под именем «<файл>.<pid>-<n>.part».
if (not Ok) or (Pub = tpubTaken) then DeleteFile(Tmp);
end; end;
ReleaseData;
end; end;
procedure TCIStopWriter(var W: TTCIWavWriter);
var Left: TTCIWavWriter;
begin
Left := W;
W := nil;
if Left = nil then Exit;
Left.Close; // очередь больше не растёт
Left.WaitDrained; // дожидаемся ВСЕЙ очереди
Left.Free; // Terminate + wake + WaitFor уже пустого потока
end;
{ ★Почему здесь нет никакого срока — история двух неверных попыток.
Внутрипроцессный поток, стоящий в write(2) или fsync, остановить нечем:
Terminate ему не указ, а Free обязан сделать WaitFor (бросить живой TThread
нельзя — он ходит в общий бюджет, RecBudgetLock, который финализация юнита
освобождает). Значит, срок НЕ ограничивает выход: на мёртвом каталоге
программа всё равно ждёт syscall. Ограничивал он ровно одно — сколько записей
мы выбросим по дороге: сперва весь остаток очереди по общему сроку, потом (с
отсчётом от последнего продвижения) остаток очереди на каталоге, который
тормозил дольше срока и оживал. То есть срок не покупал ничего и стоил
подтверждённых клиенту записей. Убран.
Кому действительно нужен ограниченный выход — писателя придётся выносить в
отдельный процесс, который гасится средствами ОС; одним TThread это не
делается. }
{ ═══════════════════════════════════════════════════════════════════════════ { ═══════════════════════════════════════════════════════════════════════════
Утилиты Утилиты
═══════════════════════════════════════════════════════════════════════════ } ═══════════════════════════════════════════════════════════════════════════ }
doc/TCI.md
+114 -16
View File
@@ -508,10 +508,97 @@ web-клиента: явная просьба сильнее умолчания
Кольцо «последние N секунд» вело себя иначе в обе стороны — начало записи Кольцо «последние N секунд» вело себя иначе в обе стороны — начало записи
затирало само себя, а `SAVE` через час после `START` отдавал файл, которого у затирало само себя, а `SAVE` через час после `START` отдавал файл, которого у
ExpertSDR3 давно бы не было. ExpertSDR3 давно бы не было.
`SAVE` завершает запись и отдаёт буфер отдельному потоку-писателю: `SAVE` завершает запись и ставит её в очередь писателю: файл бывает в десятки
файл бывает в десятки мегабайт, а команда пришла в потоке клиента, который в мегабайт, а команда пришла в потоке клиента, который в это время не читает свой
это время не читает свой сокет. MP3 не поддержан — кодера в проекте нет, сокет. MP3 не поддержан — кодера в проекте нет, и на `.mp3` уходит честный
и на `.mp3` уходит честный `tci_error`. `tci_error`.
**★Писатель — один поток с очередью, и он принадлежит адаптеру.** Раньше
`SAVE` был равен «создать поток с `FreeOnTerminate`»: его никто не держал и
никто не ждал. Отсюда две беды сразу.
- **Штатный выход из программы обрывал запись.** Замерено отдельным процессом:
при закрытии сразу после `SAVE` от ожидаемых 100000044 байт на диске
оставалось 40960 — то, что успел сбросить буфер файловой системы, — а
заголовок при этом заявлял полную длину.
- **Медленный (или зависший сетевой) каталог плодил потоки.** Идущие подряд
сохранения запускали их сотнями, и память их стеков в 128-МБ бюджет не
входила вовсе.
Теперь писатель один (`TTCIWavWriter`, заводится лениво — на первом `SAVE`),
очередь ограничена `TCI_RECORD_MAX_JOBS` = 16 заданиями (сверх — клиенту
`writer busy`, а место в бюджете возвращает `CmdRecorder`), а деструктор
адаптера гасит его через `TCIStopWriter`: закрыть приём заданий, дождаться,
пока очередь допишется, и освободить.
**★Срока у этого ожидания нет намеренно** — и это вывод из двух неверных
попыток его завести. Внутрипроцессный поток, стоящий в `write(2)` или `fsync`,
остановить нечем: `Terminate` ему не указ, а `Free` обязан сделать `WaitFor` —
бросить живой `TThread` нельзя, он ходит в общий бюджет (`RecBudgetLock`),
который освобождает финализация юнита. Значит, срок **не ограничивает выход**:
на мёртвом каталоге программа всё равно стоит в syscall. Ограничивал он ровно
одно — сколько подтверждённых клиенту записей мы выбросим по дороге (сперва
весь остаток очереди по общему сроку, потом, с отсчётом от последнего
продвижения, — остаток очереди на каталоге, который тормозил дольше срока и
оживал). То есть не покупал ничего и стоил данных. Убран вместе с
`FProgress`/`Advance`/`Terminate` в остановке.
Так что при закрытии программы очередь **дописывается вся**, а поток ждётся
по-настоящему (`Destroy` = `Terminate` + `WaitFor`): ни брошенных потоков, ни
выброшенных записей после нас не остаётся. Цена названа прямо: на мёртвой
сетевой ФС выход подвиснет вместе с syscall'ом. Кому нужен гарантированно
ограниченный выход — писателя придётся выносить в отдельный процесс, который
гасится средствами ОС; одним `TThread` это не делается.
**★Файл появляется целиком или не появляется вовсе.** Прежний код не смотрел
на результат записи: `FileWrite` (как и `THandleStream.Write` под ним)
возвращает **число записанных байт** и при ошибке отдаёт 0 или -1, не поднимая
исключения, — то есть на полном диске файл дописывался «до конца» и оставался
огрызком с заголовком на полную длину. Теперь:
1. данные пишутся во **временный файл** рядом (`<имя>.<pid>-<n>.part`,
создаётся эксклюзивно и не по симлинку — `TCICreateNewFile`), циклом, с
проверкой каждого вызова (короткая запись законна, её дописываем; 0 или
-1 — провал);
2. перед публикацией идёт `FileFlush`: на ext4 с отложенным размещением «нет
места» приходит не в `write`, а именно здесь;
3. и только после этого файл появляется под целевым именем — **одним вызовом
ядра и сразу целиком** (`TCIPublishFile`).
**★Целевого имени до этого момента не существует вовсе, и публикация ничего не
заменяет.** Промежуточный вариант — занять имя пустым файлом заранее — был
хуже обоих: на медленном диске клиент всю запись видел WAV на 0 байт, а по
имени файла он вправе считать запись готовой.
Портируемо получить сразу три свойства — атомарное появление, запрет замены и
работу на любой файловой системе — нельзя, поэтому `TCIPublishFile` идёт по
списку и на последнем шаге честно отказывается:
1. **`renameat2(AT_FDCWD, tmp, AT_FDCWD, dst, RENAME_NOREPLACE)`** — ровно то,
что нужно, одним вызовом ядра. Обёртки в RTL нет, зовём напрямую через
`Do_SysCall` (номера ABI: x86_64 316, i386 353, aarch64 276, arm 382);
`EEXIST` — имя занято, это отказ по существу.
2. **`link(2)` + `unlink`** — если `renameat2` нет (старое ядро, другая
архитектура, ФС не умеет флаг: `ENOSYS`/`EINVAL`). Семантика та же: новое
имя обязано не существовать, иначе `EEXIST`, и на симлинк по этому имени
`link` тоже не пойдёт. Каталог у временного и целевого файла один
(`tci.record_dir`), так что `EXDEV` здесь не бывает.
3. **Отказ**, если нет и жёстких ссылок (FAT/exFAT, часть CIFS/SMB и FUSE).
★`FileExists` + `rename` в этом месте недопустим: `rename` затирает то, что
лежит по имени СЕЙЧАС, а между проверкой и переносом туда может попасть что
угодно — это ровно то окно (TOCTOU), ради закрытия которого всё и
затевалось. Данные при этом не пропадают: писатель **оставляет их во
временном файле** `<имя>.wav.<pid>-<n>.part` — сказать клиенту уже нечем
(`SAVE` подтверждён давно), а стирать его десятки мегабайт из-за нашей
неспособности переименовать хуже, чем оставить их лежать. Лечится
настройкой `tci.record_dir` на обычную файловую систему.
На Windows нужной семантикой обладает `MoveFileW` без
`MOVEFILE_REPLACE_EXISTING`: существующее имя = отказ.
Провал записи (и занятое имя) убирает временный файл и не оставляет целевого;
единственное исключение — шаг 3 выше. Клиент, увидевший файл под запрошенным
именем, всегда вправе считать запись готовой.
**★Память рекордера — три замка, и все три нужны.** Авторизации в протоколе **★Память рекордера — три замка, и все три нужны.** Авторизации в протоколе
нет (§3.1), поэтому «сколько памяти займёт одна строка из сети» — это вопрос нет (§3.1), поэтому «сколько памяти займёт одна строка из сети» — это вопрос
@@ -523,8 +610,9 @@ ExpertSDR3 давно бы не было.
1. **Память набирается кусками по секунде, а не вся сразу.** `START` не стоит 1. **Память набирается кусками по секунде, а не вся сразу.** `START` не стоит
ни байта; молчащий, замьюченный или просто не звучащий приёмник не стоит ни байта; молчащий, замьюченный или просто не звучащий приёмник не стоит
ничего вовсе. Куски не перевыделяются (никакого `realloc` в DSP-потоке) и ничего вовсе. Куски не перевыделяются (никакого `realloc` в DSP-потоке) и
склеиваются один раз — в `Take`, а он идёт уже после того, как рекордер **не склеиваются вовсе**: `Take` отдаёт их писателю как есть, а тот пишет
вынут из таблицы, то есть без DSP-потока на плечах. их в файл подряд. Сам `Take` идёт уже после того, как рекордер вынут из
таблицы, то есть без DSP-потока на плечах.
2. **Общий бюджет `TCI_RECORD_MAX_BYTES` (128 МБ) на все рекордеры сразу.** 2. **Общий бюджет `TCI_RECORD_MAX_BYTES` (128 МБ) на все рекордеры сразу.**
Спрашивается при выделении каждого куска. Отказ не рушит запись: набранное Спрашивается при выделении каждого куска. Отказ не рушит запись: набранное
остаётся сохраняемым, просто дальше она не растёт — иначе клиент терял бы остаётся сохраняемым, просто дальше она не растёт — иначе клиент терял бы
@@ -535,10 +623,10 @@ ExpertSDR3 давно бы не было.
бюджете рядом жили бы 128 МБ кусков и 128 МБ копии, а счётчик показывал бы бюджете рядом жили бы 128 МБ кусков и 128 МБ копии, а счётчик показывал бы
128. Куски и так лежат встык, поэтому `TTCIWavWriter` пишет их в файл 128. Куски и так лежат встык, поэтому `TTCIWavWriter` пишет их в файл
подряд (хвост последнего за `Count` — не данные), а место отпускает в подряд (хвост последнего за `Count` — не данные), а место отпускает в
`ReleaseData` — ровно один раз, сколько бы путей выхода ни было у `ReleaseJob` — ровно один раз, сколько бы путей выхода ни было у записи.
`Execute`. Пока писатель ждёт медленный или зависший сетевой каталог, эти Пока писатель ждёт медленный или зависший сетевой каталог, эти байты
байты остаются занятыми: потолок держит и очередь сохранений, а не только остаются занятыми: потолок держит и очередь сохранений, а не только сами
сами записи. записи.
3. **Освобождение по трём событиям, а не по одному.** Раньше срок проверял 3. **Освобождение по трём событиям, а не по одному.** Раньше срок проверял
только `Feed`, то есть DSP-поток — а к мёртвому приёмнику он не приходит только `Feed`, то есть DSP-поток — а к мёртвому приёмнику он не приходит
никогда, и `START` на несуществующий номер оставлял память навсегда. никогда, и `START` на несуществующий номер оставлял память навсегда.
@@ -562,10 +650,12 @@ ExpertSDR3 давно бы не было.
(`bad file name`). Выйти за каталог после `ExtractFileName` нечем: (`bad file name`). Выйти за каталог после `ExtractFileName` нечем:
разделителей в имени уже не осталось, а `..` не проходит проверку. Файл разделителей в имени уже не осталось, а `..` не проходит проверку. Файл
создаётся **эксклюзивно** (`TCICreateNewFile`: `O_EXCL or O_NOFOLLOW` на Unix, создаётся **эксклюзивно** (`TCICreateNewFile`: `O_EXCL or O_NOFOLLOW` на Unix,
`CREATE_NEW` на Windows) — существующий не перезаписывается и симлинк не `CREATE_NEW` на Windows) — симлинк не уводит наружу, причём одним вызовом
уводит наружу, причём одним вызовом ядра, без окна между `FileExists` и ядра, без окна между `FileExists` и созданием; существующий файл не
созданием. Клиенту про уже занятое имя отвечаем `file exists` до постановки перезаписывается и при публикации (`renameat2(RENAME_NOREPLACE)`/`link`/
задачи писателю. `MoveFileW` — все без замены, см. выше). Клиенту про уже занятое имя отвечаем
`file exists` до постановки задачи писателю — но это вежливость, а не защита:
защита в самих вызовах.
**Потолок кадра.** Приёмный буфер соединения (`WsClient.WS_BUF_SIZE`) поднят **Потолок кадра.** Приёмный буфер соединения (`WsClient.WS_BUF_SIZE`) поднят
с 4 до 32 КБ: блок TX-аудио — это 64 байта заголовка плюс `data[16384]`, а с 4 до 32 КБ: блок TX-аудио — это 64 байта заголовка плюс `data[16384]`, а
@@ -843,7 +933,7 @@ ExpertSDR3 давно бы не было.
движков и сети валится с AV — клиент получает `tci_error`, соединение живо) и движков и сети валится с AV — клиент получает `tci_error`, соединение живо) и
неразрывность пачки инициализации под крутящейся ручкой. неразрывность пачки инициализации под крутящейся ручкой.
### Стенд этапа 2 (бинарные потоки) — 158 проверок, все зелёные ### Стенд этапа 2 (бинарные потоки) — 214 проверок, все зелёные
Отдельная программа (`test/tci/tcitest.pas`, прогон — `test/tci/run.sh`, Отдельная программа (`test/tci/tcitest.pas`, прогон — `test/tci/run.sh`,
внешних библиотек не требует) проверяет потоки на четырёх уровнях: внешних библиотек не требует) проверяет потоки на четырёх уровнях:
@@ -874,7 +964,15 @@ ExpertSDR3 давно бы не было.
само, `Take` отдаёт САМИ КУСКИ (2.5 с записи = три куска по секунде, а не само, `Take` отдаёт САМИ КУСКИ (2.5 с записи = три куска по секунде, а не
один свёрнутый — сплошной копии не появляется ни на миг), счёт бюджета при один свёрнутый — сплошной копии не появляется ни на миг), счёт бюджета при
этом не меняется, склеенные куски дают непрерывный звук, и писатель этом не меняется, склеенные куски дают непрерывный звук, и писатель
возвращает резерв по окончании записи. ★Имя файла: простое имя ложится в каталог записей, а каталог из возвращает резерв по окончании записи. ★Писатель: задание принимается и
дописывается очередью, после удачи временного файла рядом не остаётся,
неудачная запись (счёт обещает больше, чем есть в кусках) не оставляет НИ
файла, НИ `.part`, за всё время записи 32 МБ целевое имя ни разу не видно
незаконченным, сверх потолка заданий следует отказ, после `Close` заданий не
берут, а `TCIStopWriter` дожидается 32-МБ записи — файл цел сразу после его
возврата, и **хвост очереди за ней дописан тоже** (это и есть проверка на
обрыв при выходе; без ожидания краснеют обе).
★Имя файла: простое имя ложится в каталог записей, а каталог из
просьбы отбрасывается — абсолютный путь, `..` и буква диска наружу не просьбы отбрасывается — абсолютный путь, `..` и буква диска наружу не
выводят; пусто, `..`, не-`.wav`, управляющий символ и отсутствие каталога выводят; пусто, `..`, не-`.wav`, управляющий символ и отсутствие каталога
записей дают отказ; существующий файл писатель не перезаписывает. записей дают отказ; существующий файл писатель не перезаписывает.
test/tci/tcitest.pas
+176 -41
View File
@@ -570,6 +570,45 @@ begin
end; end;
end; end;
function MakeTake(const Data: TTCIPcm; Reserved: Int64): TTCIRecTake;
// Одно задание писателю из готового куска: в проде куски приносит Take.
begin
Result.Chunks := nil;
SetLength(Result.Chunks, 1);
Result.Chunks[0] := Data;
Result.Chunk := Length(Data) div 2;
Result.Count := Result.Chunk;
Result.Reserved := Reserved;
end;
function Drained(W: TTCIWavWriter; TimeoutMs: Integer): Boolean;
// Ограниченное ожидание очереди — ТОЛЬКО для стенда: у писателя WaitDrained
// срока не имеет (см. TCIStopWriter), а зависший стенд ничего не сообщает.
var Waited: Integer;
begin
Waited := 0;
while (W.Pending > 0) and (Waited < TimeoutMs) do
begin
Sleep(5);
Inc(Waited, 5);
end;
Result := W.Pending = 0;
end;
function PartFiles(const Path: string): Integer;
// Сколько временных файлов писателя осталось рядом с целью.
var SR: TSearchRec;
begin
Result := 0;
if FindFirst(Path + '.*', faAnyFile, SR) = 0 then
begin
repeat
Inc(Result);
until FindNext(SR) <> 0;
end;
FindClose(SR);
end;
procedure TestRecorder; procedure TestRecorder;
var var
Rec: TTCIRecorder; Rec: TTCIRecorder;
@@ -581,9 +620,11 @@ var
Hdr: array[0..43] of Byte; Hdr: array[0..43] of Byte;
Sz: LongWord; Sz: LongWord;
W: TTCIWavWriter; W: TTCIWavWriter;
Waited: Integer;
Was: Int64; Was: Int64;
Tk: TTCIRecTake; Tk: TTCIRecTake;
Big: TTCIPcm;
Refused, k, Bad: Integer;
Path2: string;
begin begin
WriteLn('C. Рекордер линейного выхода'); WriteLn('C. Рекордер линейного выхода');
@@ -753,25 +794,21 @@ begin
Rec.Free; Rec.Free;
end; end;
// WAV: заголовок и длина. // ═══ WAV и писатель ═══════════════════════════════════════════════════
Path := GetTempDir + 'tcitest_rec.wav'; // ★Писатель — ОДИН поток с очередью, которым владеет адаптер. Поток на
// каждый SAVE с FreeOnTerminate не держал никто: штатный выход из программы
// обрывал запись на полуслове (из 100000044 байт на диске оставалось
// 40960), а медленный каталог плодил сотни потоков мимо бюджета.
Path := GetTempDir + 'tcitest_rec.wav';
Path2 := GetTempDir + 'tcitest_rec2.wav';
DeleteFile(Path); DeleteFile(Path);
DeleteFile(Path2);
SetLength(Data, 2000); SetLength(Data, 2000);
for i := 0 to 1999 do Data[i] := i * 8; for i := 0 to 1999 do Data[i] := i * 8;
Tk.Chunks := nil; W := TTCIWavWriter.Create;
SetLength(Tk.Chunks, 1);
Tk.Chunks[0] := Data; Check('WAV: задание принято', W.Enqueue(Path, MakeTake(Data, 0), 48000));
Tk.Chunk := Length(Data) div 2; Check('WAV: очередь дописана', Drained(W, 5000));
Tk.Count := Tk.Chunk;
Tk.Reserved := 0;
W := TTCIWavWriter.Create(Path, Tk, 48000);
Waited := 0;
while (not FileExists(Path)) and (Waited < 2000) do
begin
Sleep(10);
Inc(Waited, 10);
end;
Sleep(50);
if not FileExists(Path) then if not FileExists(Path) then
Check('WAV: файл создан', False) Check('WAV: файл создан', False)
else else
@@ -794,50 +831,148 @@ begin
finally finally
FS.Free; FS.Free;
end; end;
// ★Данные пишутся во временный файл и переименовываются поверх занятого
// имени: по имени файла клиент считает запись готовой, и незаконченного
// содержимого он там видеть не должен. После удачи «.part» не остаётся.
Check('WAV: временный файл убран', PartFiles(Path) = 0,
IntToStr(PartFiles(Path)));
// ★Писатель отпускает резерв, когда данные ему больше не нужны — иначе // ★Писатель отпускает резерв, когда данные ему больше не нужны — иначе
// потолок памяти держал бы только сами записи, а очередь сохранений на // потолок памяти держал бы только сами записи, а очередь сохранений на
// медленном диске росла бы мимо него. // медленном диске росла бы мимо него.
Was := TCIRecBudgetUsed; Was := TCIRecBudgetUsed;
Check('WAV: резерв под писателя взят', TCIRecBudgetTake(4096)); Check('WAV: резерв под писателя взят', TCIRecBudgetTake(4096));
DeleteFile(Path); DeleteFile(Path);
Tk.Chunks := nil; W.Enqueue(Path, MakeTake(Data, 4096), 48000);
SetLength(Tk.Chunks, 1); Drained(W, 5000);
Tk.Chunks[0] := Data;
Tk.Chunk := Length(Data) div 2;
Tk.Count := Tk.Chunk;
Tk.Reserved := 4096;
TTCIWavWriter.Create(Path, Tk, 48000);
Waited := 0;
while (TCIRecBudgetUsed <> Was) and (Waited < 2000) do
begin
Sleep(10);
Inc(Waited, 10);
end;
Check('WAV: писатель вернул резерв по окончании', Check('WAV: писатель вернул резерв по окончании',
TCIRecBudgetUsed = Was, IntToStr(TCIRecBudgetUsed - Was)); TCIRecBudgetUsed = Was, IntToStr(TCIRecBudgetUsed - Was));
// ★Существующий файл не трогаем: имя приходит из сети, и fmCreate затирал // ★Существующий файл не трогаем: имя приходит из сети, и fmCreate затирал
// бы любой доступный процессу файл. Пишем поверх заведомо другой длиной — // бы любой доступный процессу файл. Пишем поверх заведомо другой длиной —
// файл обязан остаться прежним. // файл обязан остаться прежним.
SetLength(Data, 10); Tk := MakeTake(Copy(Data, 0, 10), 0);
for i := 0 to 9 do Data[i] := 1; W.Enqueue(Path, Tk, 48000);
Tk.Chunks := nil; Drained(W, 5000);
SetLength(Tk.Chunks, 1);
Tk.Chunks[0] := Data;
Tk.Chunk := 5;
Tk.Count := 5;
Tk.Reserved := 0;
TTCIWavWriter.Create(Path, Tk, 48000);
Sleep(200);
FS := TFileStream.Create(Path, fmOpenRead); FS := TFileStream.Create(Path, fmOpenRead);
try try
Check('WAV: существующий файл не перезаписан', // Публикация идёт link(2)/MoveFileW без замены: занятое имя = отказ.
Check('WAV: существующий файл не перезаписан',
FS.Size = 44 + 2000 * 2, IntToStr(FS.Size)); FS.Size = 44 + 2000 * 2, IntToStr(FS.Size));
finally finally
FS.Free; FS.Free;
end; end;
Check('WAV: отказ не оставил временного файла', PartFiles(Path) = 0);
DeleteFile(Path); DeleteFile(Path);
end; end;
// ★Запись не удалась на полпути — по имени не остаётся НИЧЕГО. Прежний код
// не смотрел на результат записи вовсе (FileWrite и THandleStream.Write при
// ошибке возвращают 0 и не поднимают исключения), и на полном диске
// оставался огрызок с заголовком на полную длину. Здесь тот же путь:
// счётчик обещает больше, чем есть в кусках.
Tk := MakeTake(Data, 0);
Tk.Count := Tk.Count * 3; // кусков под это нет
W.Enqueue(Path, Tk, 48000);
Drained(W, 5000);
Check('WAV: неудачная запись не оставила файла', not FileExists(Path));
Check('WAV: неудачная запись не оставила «.part»', PartFiles(Path) = 0);
// ★Целевого имени не существует, ПОКА запись не готова. Раньше по нему
// сразу появлялась пустышка на 0 байт (имя занималось эксклюзивно, а данные
// шли во временный файл), и на медленном диске клиент видел её всю запись —
// а по имени файла он вправе считать запись готовой. Пишем 32 МБ и всё это
// время следим за именем: увидели его непустым, но не полным, или пустым —
// проверка красная.
SetLength(Big, 16 * 1024 * 1024); // 32 МБ: заведомо дольше, чем цикл ниже
FillChar(Big[0], Length(Big) * SizeOf(SmallInt), 0);
DeleteFile(Path2);
W.Enqueue(Path2, MakeTake(Big, 0), 48000);
Bad := 0;
while W.Pending > 0 do
if FileExists(Path2) then
begin
FS := TFileStream.Create(Path2, fmOpenRead or fmShareDenyNone);
try
if FS.Size <> 44 + Int64(Length(Big)) * SizeOf(SmallInt) then Inc(Bad);
finally
FS.Free;
end;
end;
Check('WAV: незаконченной записи под целевым именем не видно', Bad = 0,
IntToStr(Bad));
Check('WAV: 32 МБ дописаны', Drained(W, 20000));
DeleteFile(Path2);
// ★Очередь ограничена. Раньше «сохранить» было равно «создать поток», и
// зависший сетевой каталог давал их сотни — память стеков в бюджет не
// входит. Занимаем писателя большой записью и стучимся сверх потолка.
W.Enqueue(Path2, MakeTake(Big, 0), 48000);
Refused := 0;
for k := 0 to TCI_RECORD_MAX_JOBS + 3 do
if not W.Enqueue(GetTempDir + Format('tcitest_q%d.wav', [k]),
MakeTake(Data, 0), 48000) then
Inc(Refused);
Check('очередь: сверх потолка заданий отказ', Refused > 0,
IntToStr(Refused));
Check('очередь: дописалась', Drained(W, 20000));
for k := 0 to TCI_RECORD_MAX_JOBS + 3 do
DeleteFile(GetTempDir + Format('tcitest_q%d.wav', [k]));
// Гасят — новых заданий не принимаем: место в бюджете остаётся на
// вызывающем, и он обязан его вернуть сам (так и делает CmdRecorder).
W.Close;
Check('очередь: после Close заданий не берём',
not W.Enqueue(Path, MakeTake(Data, 0), 48000));
TCIStopWriter(W);
Check('очередь: TCIStopWriter обнуляет ссылку', W = nil);
// ★★Главная проверка P1: закрытие программы ДОЖИДАЕТСЯ записи. Раньше
// FreeOnTerminate-поток никто не ждал, и файл обрывался там, где его застал
// выход. Здесь сразу после остановки писателя файл обязан быть целым.
// ★И не только первый: за хвостом очереди клиенту тоже сказано «сохранено»,
// поэтому ждём ВСЮ очередь (срок остановки считается от последнего
// продвижения, а не от её начала).
DeleteFile(Path2);
for k := 0 to 1 do DeleteFile(GetTempDir + Format('tcitest_tail%d.wav', [k]));
W := TTCIWavWriter.Create;
Check('выход: задание принято', W.Enqueue(Path2, MakeTake(Big, 0), 48000));
for k := 0 to 1 do
W.Enqueue(GetTempDir + Format('tcitest_tail%d.wav', [k]),
MakeTake(Data, 0), 48000);
TCIStopWriter(W);
Bad := 0;
for k := 0 to 1 do
begin
Path := GetTempDir + Format('tcitest_tail%d.wav', [k]);
if not FileExists(Path) then Inc(Bad)
else
begin
FS := TFileStream.Create(Path, fmOpenRead);
try
if FS.Size <> 44 + 2000 * 2 then Inc(Bad);
finally
FS.Free;
end;
DeleteFile(Path);
end;
end;
Check('выход: хвост очереди тоже дописан', Bad = 0, IntToStr(Bad));
if not FileExists(Path2) then
Check('выход: файл дописан до конца', False)
else
begin
FS := TFileStream.Create(Path2, fmOpenRead);
try
Check('выход: файл дописан до конца',
FS.Size = 44 + Int64(Length(Big)) * SizeOf(SmallInt),
IntToStr(FS.Size));
finally
FS.Free;
end;
DeleteFile(Path2);
end;
Big := nil;
end; end;
{ ═══════════════════════════════════════════════════════════════════════════ { ═══════════════════════════════════════════════════════════════════════════