| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## 1. Где и когда это может происходить
Гонка проявляется при **одновременном** завершении задачи (Unregister/Join) и выполнении finish-действия `CommitTimer()` в воркере. Типичные сценарии:
- **Остановка агента** — при shutdown закрываются сессии, по задачам вызывается Unregister(); в это же время воркер может доходить до finish-действий уже «уходящей» задачи.
- **Закрытие сессии/канала** — например, отключение клиента, ошибка — владелец вызывает Unregister() по задаче, в которой использовался таймер.
- **Плагины и задачи, подверженные багу:** любые, что используют `TLocalTimersQueue` (отложенные таймеры через `DelayedExecutor`). В коде UA это, в частности:
- **http_output** (`plugins/lib/http/http_sender.cpp`) — таймеры для отложенных повторов запросов и flush;
- **file_input** (`plugins/file_input/file_input.cpp`) — таймеры при работе с файлами;
- **logbroker_output_new** (`plugins/logbroker_output_new/logbroker_output_impl.h`) — таймер обновления метрик.
Во всех этих случаях задача владеет `TLocalTimersQueue`, при установке таймера в finish-действие попадает `CommitTimer()`, и при быстром Unregister() возможна гонка с Join().
---
{% cut "Зачем потоку воркера вызывать Ref()" %}
Таймер планируется **асинхронно**: `DelayedExecutor.SetTimer(Timer, triggerTime)` регистрирует callback в отдельном потоке (sleeper). Когда время наступит, callback вызовется **уже после** того, как `Run()` задачи завершился. Задача и её контекст (сессия, `TLocalTimersQueue`, `ExecutionJoiner`) должны оставаться валидными до срабатывания или отмены таймера — иначе callback приведёт к use-after-free.
**Ref()** — это «удержание» задачи: пока есть лишний Ref, `Join()` не завершится (Refs не станет 0). То есть: «задача не считается полностью завершённой, пока таймер не сработал или не был сброшен». Когда таймер сбрасывают или очередь финализируют, вызывается **UnRef()** — тогда задача может перейти в joined.
**Где именно вызывается Ref/UnRef:**
Файл `logbroker/unified_agent/common/delayed_executor.cpp`:
```cpp
void TLocalTimersQueue::CommitTimer() {
if (Queue.GetCount() > 0) {
const auto triggerTime = Top().Value();
if (!TimerTriggerTime.Defined() || ...) {
if (!TimerTriggerTime.Defined()) {
// ← ЗДЕСЬ: перед первой установкой таймера в DelayedExecutor
// держим задачу «живой» до срабатывания/сброса таймера
if (!TTaskExecutor::CurrentTaskOrDie().ExecutionJoiner().TryRef()) {
CommitTimerScheduled = false;
return;
}
}
DelayedExecutor.SetTimer(Timer, triggerTime); // асинхронный таймер
TimerTriggerTime = triggerTime;
}
} else if (TimerTriggerTime.Defined()) {
DelayedExecutor.ResetTimer(Timer);
TTaskExecutor::CurrentTaskOrDie().ExecutionJoiner().UnRef(); // таймер снят — отдаём Ref
TimerTriggerTime.Clear();
}
...
}
```
`CommitTimer()` вызывается из **finish-действия** задачи (добавляется в `EnsureCommitTimerScheduled()` → `AddFinishAction([this]() { CommitTimer(); })`), т.е. выполняется в потоке воркера после выхода из `Run()`. Пример использования таймера из кода плагина — `plugins/lib/http/http_sender.cpp`: там `LocalTimerQueue.SetTimer(request->Timer, triggerTime)` планирует отложенный повтор запроса; callback при срабатывании постит событие в сессию.
{% endcut %}
---
{% cut "Контекст: участники и суть гонки" %}
**Участники:**
- **ExecutionJoiner** (`TAsyncJoiner`) — объект с атомарным счётчиком `Refs` (начальное значение 1). Пока `Refs >= 1`, задачу считают «активной». `Join()` вызывает `UnRef()`; когда `Refs` становится 0, вызывается `Promise.SetValue()` («joined»).
- **Поток задачи (Task/Unregister)** — владелец задачи; вызывает `Unregister()` → `ExecutionJoiner_.Join()` → внутри один раз `UnRef()`.
- **Поток таймера (Worker/Timer)** — воркер пула задач; выполняет finish-действия задачи. Одно из них — `CommitTimer()`, которое при первой установке таймера вызывало `ExecutionJoiner().Ref()`.
**Гонка:** между моментом, когда поток задачи делает `Join()` (и доводит `Refs` до 0), и моментом, когда поток воркера выполняет `CommitTimer()` и вызывает `Ref()`. Если `Ref()` вызывается уже после перехода в «joined», `fetch_add(1)` возвращает 0 и срабатывает `Y_ABORT_UNLESS(result >= 1, "already joined")`.
{% endcut %}
---
{% cut "До фикса: креш при гонке" %}
Поток задачи вызывает `Unregister()` и ждёт `Join()`. Поток воркера после завершения `Run()` выполняет finish-действие `CommitTimer()`. Если к этому моменту `Join()` уже выполнил `UnRef()` и `Refs == 0`, вызов `Ref()` в `CommitTimer()` приводит к падению.
```mermaid
sequenceDiagram
participant TaskThread as Поток задачи
participant Joiner as ExecutionJoiner
participant WorkerThread as Поток воркера
Note over TaskThread,WorkerThread: Задача с таймером: Run() вызвал SetTimer(), в finish-действия добавлен CommitTimer()
TaskThread->>TaskThread: Unregister()
TaskThread->>Joiner: Join()
Joiner->>Joiner: UnRef() → Refs = 0
Joiner->>Joiner: Promise.SetValue() — joined
WorkerThread->>WorkerThread: Выполняет finish-действия
WorkerThread->>WorkerThread: CommitTimer()
WorkerThread->>Joiner: Ref()
Joiner->>Joiner: fetch_add(1) → result = 0
Joiner->>WorkerThread: Y_ABORT_UNLESS(result >= 1) — CRASH
```
**Итог:** в момент вызова `Ref()` в `CommitTimer()` объект уже в состоянии «joined» (`Refs == 0`), проверка в `Ref()` не выполняется → **SIGABRT**.
{% endcut %}
---
{% cut "После фикса: корректный выход без креша" %}
В `CommitTimer()` вместо `Ref()` вызывается `TryRef()`: атомарно проверяется `Refs >= 1` (через CAS); если уже 0, `TryRef()` возвращает `false` и `CommitTimer()` сразу выходит, не вызывая `Ref()` и не трогая таймер.
```mermaid
sequenceDiagram
participant TaskThread as Поток задачи
participant Joiner as ExecutionJoiner
participant WorkerThread as Поток воркера
Note over TaskThread,WorkerThread: Та же гонка: Join() и CommitTimer() выполняются почти одновременно
TaskThread->>TaskThread: Unregister()
TaskThread->>Joiner: Join()
Joiner->>Joiner: UnRef() → Refs = 0
Joiner->>Joiner: Promise.SetValue() — joined
WorkerThread->>WorkerThread: Выполняет finish-действия
WorkerThread->>WorkerThread: CommitTimer()
WorkerThread->>Joiner: TryRef()
Joiner->>Joiner: load(Refs) = 0 → current < 1
Joiner->>WorkerThread: return false
WorkerThread->>WorkerThread: CommitTimerScheduled = false return
Note over WorkerThread: Таймер не ставится, креша нет
```
**Итог:** при уже «joined» состоянии `TryRef()` возвращает `false`, `CommitTimer()` завершается без вызова `Ref()` и без падения.
{% endcut %}
---
{% cut "Сводка изменений" %}
| Место | До фикса | После фикса |
|-------|----------|-------------|
| `CommitTimer()` при первой установке таймера | `Ref()` → при Refs=0 креш | `TryRef()` → при `false` ранний выход |
| `TAsyncJoiner` | Только `Ref()` / `UnRef()` | Добавлен `TryRef()` (CAS, при refs<1 возврат false) |
| `Finalize()` | Не сбрасывал активный таймер | При `TimerTriggerTime.Defined()` — `ResetTimer`, `UnRef()`, `Clear()` до `Finalized = true` |
Тест `TestTimerQueueUnregisterNoCrash` (500 итераций: задача с таймером → Pulse → Unregister) без фикса периодически воспроизводит креш; с фиксом — стабильно зелёный.
{% endcut %}
commit_hash:5f57d88fc53f44db31e87deaeca57a7e9ef262ca
|
