248 lines
27 KiB
Markdown
248 lines
27 KiB
Markdown
# u_async — Центральный цикл событий (event loop)
|
||
|
||
## 1. Назначение
|
||
|
||
u_async — центральный планировщик проекта uTun. Каждый поток, которому нужна асинхронная обработка сокетов, таймеров и отложенных задач, создаёт свой экземпляр `struct UASYNC` и запускает `uasync_mainloop()` / `uasync_poll()`. Модуль объединяет:
|
||
|
||
- **Сокеты** — мониторинг fd/socket на чтение/запись/ошибки через Linux epoll (предпочтительно) с fallback на poll/select.
|
||
- **Таймеры** — однократные таймеры на базе `timeout_heap` (min-heap) и `memory_pool`. Аргумент задаётся в единицах 0.1 мс; положительный абсолютный срок округляется вверх до миллисекунды.
|
||
- **Отложенное выполнение** — FIFO-очередь `uasync_call_soon`. Callback может выполниться в текущем `uasync_poll`, при следующем проходе обработки этой очереди.
|
||
- **Межпоточную связь** — `uasync_post` / `uasync_post_reserved` публикуют callback для потока, обслуживающего экземпляр (wakeup через eventfd/pipe/UDP-сокет).
|
||
|
||
На Linux используется epoll, на FreeBSD/Windows — poll/select. Платформенная абстракция прозрачна для вызывающего кода.
|
||
|
||
## 2. Как пользоваться
|
||
|
||
### Правила
|
||
|
||
- **Один обслуживающий поток на экземпляр.** Независимые экземпляры могут работать в разных потоках. Сокеты, таймеры и обработка очередей конкретного экземпляра принадлежат его потоку. Из другого потока разрешены `uasync_post`, `uasync_post_reserved`, `uasync_wakeup` и `uasync_stop` при согласованном времени жизни экземпляра.
|
||
- **Никаких sleep/usleep.** Если поток заблокирован на sleep, цикл событий стоит. Все ожидания — только через таймеры `uasync_set_timeout`.
|
||
- **Закрытие из callback.** Можно удалять сокет и освобождать его пользовательский контекст из callback. Перед следующим callback модуль повторно проверяет регистрацию, поколение и разрешённые события. Для завершения цикла вызывайте `uasync_stop`; `uasync_destroy` выполняйте после возврата из `uasync_poll` / `uasync_mainloop`, вне callbacks.
|
||
- **Указатель `user_arg`** передаётся во все callback и позволяет передать контекстную структуру (например, `UTUN_INSTANCE`).
|
||
- **Владение контекстом.** `user_arg` не копируется и автоматически не освобождается. При отмене или уничтожении ожидающей задачи её callback не вызывается; контекст освобождает владелец. Перед `destroy` завершите producers, которые могут обращаться к экземпляру или публиковать задачи.
|
||
- **Статусы `running` и `stop` независимы.** После `create` оба равны 0, API уже доступен. `mark_running` / `mark_stopped` только выставляют статус владельца; цикл и регистрацию задач не меняют. `stop` устанавливает отдельный флаг выхода из `mainloop`, который автоматически не сбрасывается.
|
||
|
||
### Типовой сценарий
|
||
|
||
```c
|
||
// 1. Создать uasync
|
||
struct UASYNC* ua = uasync_create();
|
||
if (!ua) return;
|
||
|
||
// 2. Зарегистрировать сокет
|
||
void* sock_id = uasync_add_socket(ua, fd, my_read_cb, NULL, NULL, "my_fd", my_ctx);
|
||
// Вместо предыдущей строки для socket_t:
|
||
// void* sock_id = uasync_add_socket_t(ua, sock, my_read_cb_sock, my_write_cb_sock, NULL, "my_socket", my_ctx);
|
||
|
||
// 3. Зарегистрировать таймер (timebase = 0.1 мс, т.е. timeout_tb=100 = 10 мс)
|
||
void* t_id = uasync_set_timeout(ua, 100, my_ctx, my_timer_cb, "my_timer");
|
||
|
||
// 4. Отменить таймер (обязательно обнулить дескриптор!)
|
||
uasync_cancel_timeout(ua, t_id);
|
||
t_id = NULL;
|
||
|
||
// 5. Отложенное выполнение при обработке FIFO call_soon
|
||
void* soon_id = uasync_call_soon(ua, my_ctx, deferred_cb);
|
||
|
||
// 6. Динамически включить/отключить мониторинг записи на сокете
|
||
uasync_set_socket_write(ua, sock_id, 1); // включить
|
||
uasync_set_socket_write(ua, sock_id, 0); // отключить
|
||
|
||
// 7. Запустить главный цикл (блокирующий, выход через uasync_stop)
|
||
uasync_mainloop(ua);
|
||
|
||
// 8. Вместо mainloop можно вызывать отдельные шаги:
|
||
// uasync_poll(ua, 500); // системное ожидание до 50 мс, время callbacks этим не ограничено
|
||
|
||
// 9. Завершение
|
||
uasync_destroy(ua, 1); // закрыть оставшиеся зарегистрированные fd и внутренние ресурсы
|
||
```
|
||
|
||
### Таймеры: правильный паттерн
|
||
|
||
```c
|
||
void* reconnect_timer = NULL; // дескриптор всегда обнуляем
|
||
|
||
// При запуске:
|
||
if (!reconnect_timer) {
|
||
reconnect_timer = uasync_set_timeout(ua, 5000, ctx, reconnect_cb, "reconnect");
|
||
}
|
||
|
||
// В callback или при отмене:
|
||
void reconnect_cb(void* arg) {
|
||
reconnect_timer = NULL; // сработал — обнулили
|
||
// ...
|
||
}
|
||
|
||
// При ручной отмене:
|
||
if (reconnect_timer) {
|
||
uasync_cancel_timeout(ua, reconnect_timer);
|
||
reconnect_timer = NULL;
|
||
}
|
||
```
|
||
|
||
### Межпоточное взаимодействие (uasync_post)
|
||
|
||
```c
|
||
// Из другого потока:
|
||
void notify_main(void* arg) {
|
||
UTUN_INSTANCE* u = arg;
|
||
// Работаем в потоке, обслуживающем u->ua: здесь можно регистрировать таймеры и сокеты.
|
||
uasync_set_timeout(u->ua, 10, u, handle_work, "work");
|
||
}
|
||
|
||
void thread_func(UTUN_INSTANCE* u) {
|
||
// …
|
||
uasync_post(u->ua, notify_main, u);
|
||
}
|
||
|
||
// Отдельный uasync_memsync перед post не требуется: публикация и получение задачи используют posted_lock.
|
||
```
|
||
|
||
Публикация не продлевает время жизни `ua` или `arg` и не проверяет `running` / `stop`.
|
||
Обычный `uasync_post` выделяет `posted_task`; при OOM пишет ошибку в лог и не ставит задачу, статуса возврата нет.
|
||
Для завершения worker без выделения памяти заранее выделите `posted_task` через `u_malloc`, заполните `callback` и `arg`
|
||
и один раз передайте в `uasync_post_reserved`. После передачи производитель к задаче не обращается.
|
||
Задачу освобождает uasync после исполнения, успешной отмены или при уничтожении без вызова callback; `arg` не освобождается.
|
||
Ошибка wakeup оставляет уже опубликованную задачу в очереди; обе функции публикации не возвращают эту ошибку.
|
||
|
||
Отмена через `uasync_cancel_post` выполняется в потоке владельца вне callbacks после завершения производителя.
|
||
`ERR_OK` означает, что задача найдена в ожидающей очереди и освобождена; `ERR_FAIL` — не найдена.
|
||
Порция, уже взятая на исполнение, в этой очереди не находится. `uasync_memsync` лишь захватывает и отпускает
|
||
`posted_lock`; он не ждёт исполнения задач и не заменяет согласование времени жизни контекста.
|
||
|
||
### Получение времени
|
||
|
||
```c
|
||
uint64_t now_tb = get_time_tb(); // timebase 0.1 мс (монотонные часы)
|
||
uint64_t now_us = get_time_us(); // микросекунды (для burst-измерений)
|
||
```
|
||
|
||
## 3. API
|
||
|
||
### Жизненный цикл
|
||
|
||
| Функция | Описание |
|
||
|---------|----------|
|
||
| `uasync_create()` | Создать готовый к регистрации и опросу экземпляр с `running=0`, `stop=0`. Возвращает указатель или `NULL` при ошибке инициализации. |
|
||
| `uasync_destroy(ua, close_fds)` | Уничтожить вне callbacks после возврата из poll/mainloop и завершения producers. При `close_fds != 0` закрывает оставшиеся зарегистрированные fd; внутренние ресурсы закрывает всегда. Ожидающие задачи и таймеры удаляет без callbacks и без освобождения `arg`. Проверяет баланс внутренних аллокаций (abort при несовпадении). `NULL` допустим. |
|
||
| `uasync_mark_running(ua)` / `uasync_mark_stopped(ua)` | Атомарно выставить `running=1/0`. Не меняют `stop`, не управляют циклом и не ограничивают API. |
|
||
| `uasync_is_running(ua)` | Атомарно прочитать статус владельца; для `NULL` возвращает 0. |
|
||
| `uasync_stop(ua)` | Атомарно выставить `stop=1` и попытаться разбудить ожидание. Допустим из другого потока при живом экземпляре. Текущий poll, оставшуюся обработку событий и callbacks не прерывает; `running` не меняет. |
|
||
| `uasync_mainloop(ua)` | Цикл `while(!stop) uasync_poll(ua, -1)`. При уже установленном `stop` сразу возвращается, флаг не сбрасывает. |
|
||
| `uasync_poll(ua, timeout_tb)` | Один проход обработки событий. Отрицательное значение снимает пользовательский предел системного ожидания; 0 запрещает блокирующее ожидание; положительное задаёт предел в 0.1 мс с округлением вверх до миллисекунд, сокращаемый таймерами. Время callbacks этим пределом не ограничено. После stop ожидание неблокирующее. |
|
||
|
||
### Таймеры (timebase = 0.1 мс)
|
||
|
||
| Функция | Описание |
|
||
|---------|----------|
|
||
| `uasync_set_timeout(ua, timeout_tb, arg, cb, name)` | Однократный таймер: `timeout_tb >= 0`, callback обязателен. Возвращает handle или `NULL` при неверных аргументах/ошибке ресурсов. `arg` и строка `name` сохраняются по указателю; ограничения имени в 15 символов нет. |
|
||
| `uasync_cancel_timeout(ua, t_id)` | Отменить ещё живой таймер этого экземпляра: `ERR_OK` / `ERR_FAIL`. При успехе узел помечается отменённым; освобождение отложено до очистки heap или destroy. После успеха обнулить handle. |
|
||
|
||
Для положительного `timeout_tb` абсолютный срок в миллисекундах равен
|
||
`ceil((now_us + timeout_tb * 100) / 1000)`, где `now_us` взят при регистрации.
|
||
При нулевом интервале срок равен `floor(now_us / 1000)` и таймер готов к ближайшей обработке heap.
|
||
Callback синхронно из `set_timeout` не вызывается; его исполнение может задержаться из-за ожидания или других callbacks.
|
||
Строка `name` должна оставаться доступной до освобождения узла, включая отложенную очистку после отмены.
|
||
В начале callback сохранённый handle нужно обнулить. После исполнения, освобождения или destroy старый указатель
|
||
использовать нельзя: безопасная ошибка при повторной отмене не гарантируется.
|
||
|
||
### Немедленное выполнение (FIFO)
|
||
|
||
| Функция | Описание |
|
||
|---------|----------|
|
||
| `uasync_call_soon(ua, arg, cb)` | Добавить callback в FIFO для отложенного исполнения. Возвращает handle или `NULL` при неверных аргументах/ошибке ресурсов; узел выделяется из `timeout_pool`. |
|
||
| `uasync_call_soon_cancel(ua, t_id)` | Занулить callback ещё живого узла этого экземпляра за O(1); узел освобождается при обработке очереди/destroy. `ERR_OK` при отмене, `ERR_FAIL` при `NULL`/чужом экземпляре. После успеха обнулить handle. |
|
||
| `uasync_drain_immediate(ua)` | Синхронно обработать все порции call_soon до опустошения, включая новые callbacks. Не обрабатывает posted tasks, сокеты или heap таймеров. Постоянный самоперезапуск callback может не дать функции вернуться. |
|
||
|
||
Один проход обработки FIFO берёт исходную порцию. Обычный успешный `poll` обрабатывает FIFO в начале,
|
||
затем после ожидания и обработки I/O/posted tasks, перед проверкой heap таймеров.
|
||
Поэтому callback из обработчика сокета или из первой порции call_soon может выполниться в том же `poll`.
|
||
Ошибка ожидания может прервать проход до второй обработки FIFO. Гарантии «следующая итерация poll/mainloop» нет.
|
||
При повторной отмене call_soon до обработки узла возможен `ERR_OK`; после освобождения узла старый handle недействителен.
|
||
|
||
### Сокеты
|
||
|
||
| Функция | Описание |
|
||
|---------|----------|
|
||
| `uasync_add_socket(ua, fd, r_cb, w_cb, e_cb, name, arg)` | Зарегистрировать fd. Callbacks могут быть `NULL`; `name` и `arg` заимствуются. Возвращает handle либо `NULL` при неверном fd, дублирующей регистрации или ошибке ресурсов/backend. |
|
||
| `uasync_add_socket_t(ua, sock, r_cb, w_cb, e_cb, name, arg)` | То же для `socket_t`; callbacks получают значение сокета. |
|
||
| `uasync_remove_socket(ua, s_id)` | Удалить регистрацию по актуальному handle, не закрывая fd и не освобождая `arg`. Слот становится доступным для повторного использования. `ERR_OK` / `ERR_FAIL`. |
|
||
| `uasync_remove_socket_t(ua, sock)` | Удалить регистрацию типа SOCK по значению `socket_t`, не закрывая сокет. `ERR_OK` / `ERR_FAIL`. |
|
||
| `uasync_set_socket_read(ua, s_id, enable)` | По актуальному handle включить (`enable != 0`) / отключить (`0`) мониторинг чтения для заданного callback. `ERR_OK` при успехе, `ERR_FAIL` при неверном handle/ошибке backend. |
|
||
| `uasync_set_socket_write(ua, s_id, enable)` | То же для мониторинга записи. |
|
||
| `uasync_lookup_socket(ua, fd, &s_id)` | 0 и актуальный handle, либо -1. При допустимых аргументах и отсутствии регистрации записывает `NULL`; при неверных аргументах выход не меняет. |
|
||
|
||
Handle кодирует индекс без поколения и сохраняется при расширении массива. После удаления его нужно обнулить:
|
||
новая регистрация может получить то же числовое значение. Поколения защищают внутреннюю обработку событий,
|
||
но не делают безопасными вызовы API со старым handle. Строка `name` должна жить до удаления регистрации.
|
||
|
||
### Межпоточное взаимодействие
|
||
|
||
| Функция | Описание |
|
||
|---------|----------|
|
||
| `uasync_post(ua, cb, arg)` | **Потокобезопасно.** Выделить и опубликовать задачу для потока экземпляра, попытаться разбудить его. При OOM задача не публикуется, ошибка логируется; статуса возврата нет. |
|
||
| `uasync_post_reserved(ua, task)` | **Потокобезопасно.** Передать заранее выделенный через `u_malloc` task с заполненными callback/arg без новых аллокаций. `ua` и `task` обязательны. Владение task переходит uasync. |
|
||
| `uasync_cancel_post(ua, task)` | В потоке владельца вне callbacks после завершения производителя: `ERR_OK` при удалении и освобождении ожидающего task, `ERR_FAIL`, если он не найден. `ua` и `task` обязательны. |
|
||
| `uasync_memsync(ua)` | Lock/unlock `posted_lock` живого экземпляра. Не ждёт исполнения задач; публикация через post уже использует этот мьютекс. |
|
||
| `uasync_wakeup(ua)` | 0 при успешной записи, -1 при ошибке/неинициализированном wakeup. POSIX: запись `uint64_t(1)` в eventfd/pipe; Windows: send одного байта в UDP-сокет. Экземпляр должен оставаться живым. |
|
||
| `uasync_get_wakeup_fd(ua)` | Заимствованный endpoint записи, либо -1 при `NULL`/неинициализированном wakeup. Linux: eventfd, тот же fd для чтения и записи (писать `uint64_t`); прочие POSIX: pipe; Windows: UDP-сокет. Самостоятельно не закрывать. |
|
||
|
||
### Время
|
||
|
||
| Функция | Описание |
|
||
|---------|----------|
|
||
| `get_time_tb()` | Время в единицах 0.1 мс: Linux `CLOCK_BOOTTIME` (включает suspend), остальные POSIX `CLOCK_MONOTONIC`, Windows `QueryPerformanceCounter`. Изменение календарных часов на шкалу не влияет. |
|
||
| `get_time_us()` | Монотонное время в микросекундах (для burst-измерений производительности). |
|
||
|
||
### Диагностика
|
||
|
||
| Функция | Описание |
|
||
|---------|----------|
|
||
| `uasync_get_stats(ua, ...)` | Получить счётчики аллокаций/освобождений таймеров и сокетов. |
|
||
| `uasync_get_load(ua, &load)` | Актуализировать по сроку 100 мс и скопировать снимок CPU потока: текущий процент, среднее за секунду, время замера, длительность интервала и окна, valid. Только в потоке uasync. |
|
||
| `uasync_print_resources(ua, prefix)` | Вывести в лог все активные таймеры (имя, оставшееся время) и сокеты. Вызывается также в `uasync_destroy` перед очисткой. |
|
||
|
||
CPU измеряется с первого `uasync_poll` или `uasync_get_load`: по сроку 100 мс читается CPU-время текущего потока
|
||
(`CLOCK_THREAD_CPUTIME_ID` на POSIX, user+kernel `GetThreadTimes` на Windows). 100% означает одно полностью занятое ядро.
|
||
Учитывается вся работа этого потока между замерами, включая код между вызовами poll; CPU других потоков не учитывается.
|
||
Проценты вычисляются по фактической длительности интервала: `100 * cpu_ns / (elapsed_us * 1000)`.
|
||
Реальное время берётся из `get_time_us` (Linux: `CLOCK_BOOTTIME`, включая suspend; Windows: QPC).
|
||
|
||
Сроки измерения идут по сетке 100 мс без накопления дрейфа. Проверка выполняется при работе цикла и при запросе нагрузки.
|
||
Измеритель не создаёт таймеров, не ограничивает ожидание событий и не будит спящий поток.
|
||
После непрерывного ожидания не менее секунды весь старый интервал нагрузки вытеснен: текущий процент и среднее равны 0%.
|
||
Это фиксируется сразу после выхода из ожидания, до callbacks; CPU последующих обработчиков учитывается с новой точки отсчёта.
|
||
История хранится внутри `UASYNC`, без аллокаций. Частые запросы не создают лишних отсчётов.
|
||
Среднее ограничено последней секундой; вклад пересекающего границу интервала оценивается пропорционально перекрытию.
|
||
До первого замера `valid=0`, затем окно растёт до секунды. Долгий callback задерживает публикацию;
|
||
пропущенные отсчёты не восстанавливаются, текущий процент относится ко всему фактическому интервалу.
|
||
При большом перерыве вне ожидания среднее внутри этого интервала является оценкой по его общей загрузке.
|
||
Снимок содержит `sample_time_us`, чтобы потребитель мог показать возраст данных.
|
||
Ошибка CPU-часов логируется и отключает измеритель (`valid=0`), сохраняя работу цикла событий.
|
||
В `sys=debug` раз в секунду выводятся проценты, длительность интервала/окна и опоздание;
|
||
пропуск полного периода из-за долгой обработки выводится как WARN. Статус чата и headless-команда `status` показывают CPU и среднее.
|
||
|
||
GUI показывает округлённое секундное среднее 0–100% поверх полосы с градиентом зелёный → жёлтый → красный:
|
||
desktop — рядом с заголовком каналов, Android — рядом с версией в списке групп.
|
||
Запрос `chat_core_collect_cpu_load_trampoline` выполняется в потоке ядра и возвращает событие `CHAT_EVT_CPU_LOAD`:
|
||
два байта `[valid][average_percent]`, где среднее округлено до целого процента. При `valid=0` GUI показывает 0%.
|
||
Desktop-таймер проверяет готовность ядра каждые 333 мс; Android отправляет запросы с паузой 333 мс,
|
||
только пока экран списка групп RESUMED. В обоих случаях допускается не более одного незавершённого запроса.
|
||
При паузе или уходе с Android-экрана coroutine опроса отменяется и новые запросы не отправляются.
|
||
Уже опубликованный запрос не отменяется: ответ снимает pending, но при скрытом экране валидный ответ не меняет индикатор.
|
||
Для индикатора не создаются таймеры ядра.
|
||
|
||
### Внутренние структуры
|
||
|
||
- **`struct timeout_node`** — узел таймера: name, arg, callback, expiration_ms, heap_index. Используется и для таймеров в heap, и для FIFO-очереди immediate (через поле `next`). Выделяется из `timeout_pool`.
|
||
- **`struct socket_node`** — узел сокета: fd/sock, тип (FD/SOCK), колбэки, user_data, флаги active/enable_read/enable_write, gen (защита от stale epoll-событий после переиспользования fd).
|
||
- **`struct socket_array`** — массив сокетов: O(1) доступ по fd через `fd_to_index[]`, обход активных через `active_indices[]`, динамическое расширение.
|
||
- **`struct posted_task`** — задача из другого потока: callback + arg + next, защищена `posted_lock`.
|
||
|
||
Таймеры используют шкалу `get_time_us`; правила округления абсолютного срока и порядок обработки
|
||
call_soon описаны в разделах API выше. Сокеты и posted tasks обрабатываются после ожидания перед heap таймеров;
|
||
ошибка ожидания может завершить проход раньше. Установка `stop` не прерывает эти обработчики.
|
||
|
||
Для fallback `poll` массив дескрипторов резервируется при создании экземпляра и до регистрации нового сокета. Отказ памяти возвращает `NULL` из `uasync_create` / `uasync_add_socket*`; существующие регистрации сохраняются. Перестроение массива в `uasync_poll` не выделяет память. Расширение из callback сохраняет события текущего прохода, новый сокет опрашивается со следующего прохода.
|