Browse Source

android: standby keepalive sync + rx logging during sleep

proxy
evgeny 1 week ago
parent
commit
2f3001458f
  1. 29
      src/call/call_audio.c
  2. 23
      src/transport_layer/etcp_connections.c
  3. 13
      src/transport_layer/etcp_keepalive.c
  4. 4
      src/transport_layer/etcp_keepalive.h
  5. 13
      src/transport_layer/reality_relay.c
  6. 23
      src/transport_layer/stcp.c
  7. 2
      src/transport_layer/stcp.h
  8. 10
      src/transport_layer/stcp_client.c
  9. 18
      src/transport_layer/stcp_server.c
  10. 2
      tools/chatgui-android/libutun_lite/instance_lite.c
  11. 9
      tools/chatgui-android/libutun_lite/standby.c
  12. 4
      tools/chatgui-android/libutun_lite/standby.h

29
src/call/call_audio.c

@ -37,10 +37,6 @@ static opus_codec_decoder_t* g_decoder = NULL;
static struct vj* g_vj = NULL;
static struct call_tones* g_tones = NULL;
/* диагностика ставок (DEBUG_CATEGORY_DEBUG): под g_mtx */
static uint32_t g_tx_feed = 0, g_tx_encode_fail = 0, g_rx_push = 0, g_rx_pull = 0;
static uint64_t g_rate_last_tb = 0;
static const struct call_audio_ops g_ops = {
.on_media = call_audio_on_media,
.get_stats = call_audio_get_stats,
@ -78,7 +74,6 @@ void call_audio_on_media(struct UTUN_INSTANCE* inst, uint64_t call_id,
if (g_active && call_id == g_call_id && g_vj && opus && len > 0) {
vj_push(g_vj, opus, len);
g_last_media_tb = get_time_tb();
g_rx_push++;
}
pthread_mutex_unlock(&g_mtx);
}
@ -204,20 +199,6 @@ void call_audio_stop(void) {
/* ── аудио-поток: TX / RX ── */
/* Раз в ~1с логируем ставки TX/RX — диагностика «звонок не проходит» (DEBUG_CATEGORY_DEBUG).
* Вызывается из аудио-потока (pull_pcm); счётчики под g_mtx. */
static void call_audio_rate_tick(uint64_t now_tb) {
uint32_t feed = 0, encfail = 0, push = 0, pull = 0, resync = 0;
pthread_mutex_lock(&g_mtx);
feed = g_tx_feed; encfail = g_tx_encode_fail; push = g_rx_push; pull = g_rx_pull;
g_tx_feed = g_tx_encode_fail = g_rx_push = g_rx_pull = 0;
if (g_vj) resync = vj_resyncs(g_vj);
g_rate_last_tb = now_tb;
pthread_mutex_unlock(&g_mtx);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: rates feed=%u/s encfail=%u push=%u/s pull=%u/s resync=%u",
CALL_AUDIO_ID, feed, encfail, push, pull, resync);
}
int call_audio_feed_pcm(uint64_t call_id, const int16_t* pcm, int count) {
if (!pcm || count != CALL_AUDIO_FRAME_SAMPLES) return -1;
@ -232,8 +213,6 @@ int call_audio_feed_pcm(uint64_t call_id, const int16_t* pcm, int count) {
return -1;
}
len = opus_codec_encode(g_encoder, pcm, count, opus, (int)sizeof(opus));
g_tx_feed++;
if (len <= 0) g_tx_encode_fail++;
ua = g_inst ? g_inst->ua : NULL;
inst = g_inst;
pthread_mutex_unlock(&g_mtx);
@ -269,19 +248,11 @@ int call_audio_pull_pcm(uint64_t call_id, int16_t* out, int max_samples) {
j = g_vj;
tones = g_tones;
last_media = g_last_media_tb;
g_rx_pull++;
}
pthread_mutex_unlock(&g_mtx);
if (!active || !j || !tones) return 0;
/* диагностика ставок: раз в ~1с (аудио-поток) */
{
uint64_t now_tb = get_time_tb();
if (g_rate_last_tb == 0) g_rate_last_tb = now_tb;
else if (now_tb - g_rate_last_tb >= 10000) call_audio_rate_tick(now_tb);
}
/* завершение: доигрываем тон (450мс), затем тишина. Тон стартует лениво
* здесь (аудио-поток) — begin_end лишь выставляет флаг, чтобы не гонять
* генератор тонов с GUI-потоком. */

23
src/transport_layer/etcp_connections.c

@ -28,6 +28,9 @@
#include "../lib/debug_config.h"
#include "etcp_loadbalancer.h"
#include "etcp_keepalive.h"
#ifdef UTUN_HAVE_STANDBY
#include "standby.h"
#endif
#include <stdlib.h>
#include <time.h>
#include "../lib/mem.h"
@ -2107,6 +2110,9 @@ static void etcp_process_packet(struct ETCP_SOCKET* e_sock, uint8_t* data,
salt[0], salt[1], salt[2], salt[3], salt[4], salt[5], salt[6], salt[7],
encrypted_pubkey[0], encrypted_pubkey[1], encrypted_pubkey[2], encrypted_pubkey[3]);
if (sc_decrypt(&sc, data, recv_len - SC_PUBKEY_ENC_SIZE, (uint8_t*)&pkt->timestamp, &pkt_len)) {
#ifdef UTUN_HAVE_STANDBY
{ char who[64]; snprintf(who, sizeof(who), "addr=%s", sockaddr_storage_to_str(&addr).str); standby_log_rx("udp undecryptable", who); }
#endif
if (link) {
DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO,
"packet undecryptable (normal+init fail) from=%s — existing link log=%s state=%d sess=%d",
@ -2140,6 +2146,9 @@ static void etcp_process_packet(struct ETCP_SOCKET* e_sock, uint8_t* data,
uint8_t code = pkt->data[0];
uint64_t peer_id = be64toh(*(uint64_t*)(pkt->data + 2));
if (code == ETCP_PING) {
#ifdef UTUN_HAVE_STANDBY
{ char who[64]; snprintf(who, sizeof(who), "addr=%s", sockaddr_storage_to_str(&addr).str); standby_log_rx("udp ping", who); }
#endif
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "X25519 decrypted: PING from peer=0x%016llx src=%s",
(unsigned long long)peer_id, sockaddr_storage_to_str(&addr).str);
int ret = handle_ping(e_sock, pkt, &addr, decrypted_pubkey, pkt_len);
@ -2147,6 +2156,9 @@ static void etcp_process_packet(struct ETCP_SOCKET* e_sock, uint8_t* data,
return;
}
if (code == ETCP_PONG) {
#ifdef UTUN_HAVE_STANDBY
{ char who[64]; snprintf(who, sizeof(who), "addr=%s", sockaddr_storage_to_str(&addr).str); standby_log_rx("udp pong", who); }
#endif
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "X25519 decrypted: PONG from peer=0x%016llx src=%s",
(unsigned long long)peer_id, sockaddr_storage_to_str(&addr).str);
int ret = handle_pong(e_sock, pkt, &addr, pkt_len);
@ -2483,6 +2495,17 @@ int etcp_packet_decrypted(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt,
uint8_t pkt_code = pkt->data[0];
#ifdef UTUN_HAVE_STANDBY
{
char who[256];
snprintf(who, sizeof(who), "%s addr=%s", link->etcp->log_name,
sockaddr_storage_to_str(&link->remote_addr).str);
int is_ka = (pkt_code == ETCP_KEEPALIVE || pkt_code == ETCP_KEEPALIVE_REQ || pkt_code == ETCP_KEEPALIVE_RESP);
standby_log_rx(is_ka ? (link->is_tcp ? "tcp keepalive" : "udp keepalive")
: (link->is_tcp ? "tcp data" : "udp data"), who);
}
#endif
if (etcp_keepalive_on_recv(e_sock, pkt, link, pkt_len)) {
return 0;
}

13
src/transport_layer/etcp_keepalive.c

@ -176,13 +176,13 @@ static uint8_t mobile_desired_ka_mode(struct UTUN_INSTANCE* inst) {
return (inst->client_activity == CLIENT_ACTIVITY_ACTIVE) ? KA_MODE_NORMAL : KA_MODE_STANDBY;
}
static void restart_keepalive_timer(struct ETCP_LINK* link) {
static void restart_keepalive_timer_ms(struct ETCP_LINK* link, uint16_t ms) {
if (!link || !link->etcp || !link->etcp->instance) return;
if (link->keepalive_timer) {
uasync_cancel_timeout(link->etcp->instance->ua, link->keepalive_timer);
link->keepalive_timer = NULL;
}
link->keepalive_timer = uasync_set_timeout(link->etcp->instance->ua, link->ka_period_ms * 10, link, keepalive_timer_cb, "link_keepalive");
link->keepalive_timer = uasync_set_timeout(link->etcp->instance->ua, (uint32_t)ms * 10, link, keepalive_timer_cb, "link_keepalive");
}
static void apply_link_ka_mode(struct ETCP_LINK* link, uint8_t mode) {
@ -194,7 +194,7 @@ static void apply_link_ka_mode(struct ETCP_LINK* link, uint8_t mode) {
DEBUG_INFO(DEBUG_CATEGORY_KEEPALIVE, "[%s] keepalive mode applied: mode=%s interval=%dms timeout=%dms pinned=%d",
link->etcp->log_name, mode == KA_MODE_STANDBY ? "standby" : "normal",
ms, link->keepalive_timeout, link->ka_pinned);
restart_keepalive_timer(link);
restart_keepalive_timer_ms(link, link->ka_period_ms);
}
static void etcp_keepalive_send_mode_pkt(struct ETCP_LINK* link, uint8_t code, uint8_t mode) {
@ -263,6 +263,13 @@ int etcp_keepalive_on_recv(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt,
link->keepalive_timeout = (uint32_t)peer_period * KA_TIMEOUT_MULT;
}
link->keepalive_recv_count++;
/* Сервер в standby синхронизируется с keepalive клиента: отвечает echo и
* пере-армит свой таймер на период+запас, чтобы собственная проба ушла
* только если клиент опаздывает. */
if (link->is_server && link->ka_pinned) {
etcp_link_send_keepalive(link);
restart_keepalive_timer_ms(link, (uint16_t)(link->ka_period_ms + KA_SYNC_MARGIN_MS));
}
memory_pool_free(e_sock->instance->pkt_pool, pkt);
return 1;
}

4
src/transport_layer/etcp_keepalive.h

@ -21,6 +21,10 @@ extern "C" {
#define KA_MOBILE_ACTIVE_MS 3000
#define KA_MOBILE_STANDBY_MS 30000
/* Запас (мс) поверх периода keepalive: сервер в standby пере-армит свой таймер
* на период+запас после эхо-ответа, чтобы пробу успеть отправить при опоздании клиента. */
#define KA_SYNC_MARGIN_MS 2000
// --- перенесены из etcp_connections.c (имена сохранены) ---
uint16_t negotiate_keepalive(uint8_t my_type, uint16_t my_ka,
uint8_t peer_type, uint16_t peer_ka);

13
src/transport_layer/reality_relay.c

@ -9,6 +9,9 @@
#include "../lib/mem.h"
#include "../lib/debug_config.h"
#include "../lib/platform_compat.h"
#ifdef UTUN_HAVE_STANDBY
#include "standby.h"
#endif
#include "../lib/async_dns.h"
#include <string.h>
#include <stdlib.h>
@ -132,6 +135,16 @@ static void relay_client_read_cb(socket_t sock, void *arg) {
if (r->dest_read_eof) relay_free(r);
return;
}
#ifdef UTUN_HAVE_STANDBY
{
struct sockaddr_storage sa; socklen_t slen = sizeof(sa);
char who[64];
if (getpeername(r->client_sock, (struct sockaddr*)&sa, &slen) == 0)
snprintf(who, sizeof(who), "addr=%s", sockaddr_storage_to_str(&sa).str);
else snprintf(who, sizeof(who), "?");
standby_log_rx("reality relay", who);
}
#endif
r->c2d = buf; r->c2d_len = (size_t)n; r->c2d_off = 0;
uasync_set_socket_read(r->ua, r->client_sid, 0); // backpressure
relay_flush_c2d(r);

23
src/transport_layer/stcp.c

@ -4,6 +4,9 @@
#include "../lib/mem.h"
#include "../lib/debug_config.h"
#include "../lib/platform_compat.h"
#ifdef UTUN_HAVE_STANDBY
#include "standby.h"
#endif
#include <stdlib.h>
#include <string.h>
#include <errno.h>
@ -25,6 +28,19 @@ void stcp_conn_set_on_close(struct stcp_conn *c, void (*cb)(struct stcp_conn *co
c->close_arg = arg;
}
static struct sockaddr_storage g_peer_storage;
static ip_str_t g_peer_ip;
const char* stcp_conn_peer_str(struct stcp_conn *c) {
if (!c || c->sock == SOCKET_INVALID) return "?";
memset(&g_peer_storage, 0, sizeof(g_peer_storage));
socklen_t slen = sizeof(g_peer_storage);
if (getpeername(c->sock, (struct sockaddr*)&g_peer_storage, &slen) == 0) {
g_peer_ip = sockaddr_storage_to_str(&g_peer_storage);
return g_peer_ip.str;
}
return "?";
}
void stcp_conn_free(struct stcp_conn *c) {
if (!c) return;
stcp_server_remove_conn(c);
@ -257,7 +273,12 @@ void stcp_recv_try(struct stcp_conn *c) {
size_t data_len = (size_t)c->recv_msg_size;
uint8_t *plain = c->recv_buf + 2;
if (stcp_check_crc(plain, data_len) != 0) { stcp_conn_do_close(c, 5); return; }
if (stcp_check_crc(plain, data_len) != 0) {
#ifdef UTUN_HAVE_STANDBY
standby_log_rx("tcp undecryptable", stcp_conn_peer_str(c));
#endif
stcp_conn_do_close(c, 5); return;
}
if (c->recv_on_chunk) c->recv_on_chunk(c, plain, data_len);
if (c->state == STCP_STATE_CLOSED || c->state == STCP_STATE_ERROR) return;

2
src/transport_layer/stcp.h

@ -148,6 +148,8 @@ struct stcp_conn {
};
void stcp_conn_free(struct stcp_conn *c);
/* Адрес пира соединения для логов (статический буфер). "?" если недоступен. */
const char* stcp_conn_peer_str(struct stcp_conn *c);
void stcp_conn_set_tx_queue(struct stcp_conn *c, struct ll_queue *q);
void stcp_conn_set_rx_queue(struct stcp_conn *c, struct ll_queue *q);
void stcp_conn_set_on_close(struct stcp_conn *c, void (*cb)(struct stcp_conn *conn, int err, void *arg), void *arg);

10
src/transport_layer/stcp_client.c

@ -10,6 +10,9 @@
#include "../lib/mem.h"
#include "../lib/debug_config.h"
#include "../lib/platform_compat.h"
#ifdef UTUN_HAVE_STANDBY
#include "standby.h"
#endif
#include <stdlib.h>
#include <string.h>
#include <errno.h>
@ -107,7 +110,12 @@ static void client_hs_cb(struct stcp_conn *c, uint8_t *data, size_t len) {
memcpy(enc_hs, data + SC_PUBKEY_ENC_SIZE, STCP_HS_ENC_SERVER);
log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_CRYPTO, "stcp_client enc_hs BEFORE xor", enc_hs, STCP_HS_ENC_SERVER);
size_t hs_data_len;
if (stcp_frame_decrypt(enc_hs, STCP_HS_ENC_SERVER, &c->stream_recv, &hs_data_len)) { stcp_conn_do_close(c, 3); return; }
if (stcp_frame_decrypt(enc_hs, STCP_HS_ENC_SERVER, &c->stream_recv, &hs_data_len)) {
#ifdef UTUN_HAVE_STANDBY
standby_log_rx("tcp undecryptable (hs)", stcp_conn_peer_str(c));
#endif
stcp_conn_do_close(c, 3); return;
}
log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_CRYPTO, "stcp_client enc_hs AFTER xor", enc_hs, STCP_HS_ENC_SERVER);
memcpy(c->peer_ed25519_pubkey, enc_hs, SC_PUBKEY_SIZE); c->peer_ed25519_set = 1;
c->peer_got_initial_pkt = enc_hs[32];

18
src/transport_layer/stcp_server.c

@ -13,6 +13,9 @@
#include "../utun_instance.h"
#include "../ntp_time.h"
#include "etcp.h"
#ifdef UTUN_HAVE_STANDBY
#include "standby.h"
#endif
#include <stdlib.h>
#include <string.h>
#include <errno.h>
@ -90,6 +93,9 @@ static void server_hs_phase1_cb(struct stcp_conn *c, uint8_t *data, size_t len)
memcpy(enc_hs, data + SC_PUBKEY_ENC_SIZE, STCP_HS_ENC_CLIENT);
size_t hs_data_len;
if (stcp_frame_decrypt(enc_hs, STCP_HS_ENC_CLIENT, &c->stream_recv, &hs_data_len)) {
#ifdef UTUN_HAVE_STANDBY
standby_log_rx("tcp undecryptable (hs)", stcp_conn_peer_str(c));
#endif
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "decrypt/CRC failed");
stcp_conn_do_close(c, 2); return;
}
@ -160,6 +166,9 @@ static void server_hs_phase2_cb(struct stcp_conn *c, uint8_t *data, size_t len)
int r = stcp_try_send(c, resp, total_resp);
if (r < 0) { u_free(resp); stcp_conn_do_close(c, 3); return; }
if (c->peer_flags & STCP_HANDSHAKE_FLAG_PING) {
#ifdef UTUN_HAVE_STANDBY
standby_log_rx("tcp ping", stcp_conn_peer_str(c));
#endif
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_server: ping answered, graceful close sock=%d", (int)c->sock);
if (r == 0) { stcp_conn_do_close(c, 0); return; }
c->close_after_send = 1;
@ -216,12 +225,18 @@ static void reality_ch_hdr_cb(struct stcp_conn *c, uint8_t *data, size_t len) {
(void)len;
memcpy(c->reality_hdr, data, 5);
if (data[0] != 0x16) {
#ifdef UTUN_HAVE_STANDBY
standby_log_rx("reality undecryptable", stcp_conn_peer_str(c));
#endif
DEBUG_WARN(DEBUG_CATEGORY_REALITY, "stcp_server: first byte 0x%02x — not TLS, relay", data[0]);
reality_server_start_relay(c, data, 5, c->recv_buf + 5, c->recv_buf_len - 5);
return;
}
uint16_t rec_len = (uint16_t)((data[3] << 8) | data[4]);
if (rec_len < 4 || rec_len > REALITY_MAX_CH_SIZE) {
#ifdef UTUN_HAVE_STANDBY
standby_log_rx("reality undecryptable", stcp_conn_peer_str(c));
#endif
DEBUG_WARN(DEBUG_CATEGORY_REALITY, "stcp_server: bad TLS record len %u, relay", rec_len);
reality_server_start_relay(c, data, 5, c->recv_buf + 5, c->recv_buf_len - 5);
return;
@ -249,6 +264,9 @@ static void reality_ch_body_cb(struct stcp_conn *c, uint8_t *data, size_t len) {
return;
}
DEBUG_INFO(DEBUG_CATEGORY_REALITY, "stcp_server: reality auth failed rc=%d, relay", rc);
#ifdef UTUN_HAVE_STANDBY
standby_log_rx("reality undecryptable (auth)", stcp_conn_peer_str(c));
#endif
reality_server_start_relay(c, full, 5 + len, c->recv_buf + len, c->recv_buf_len - len);
}

2
tools/chatgui-android/libutun_lite/instance_lite.c

@ -267,6 +267,7 @@ static void* instance_thread(void* arg) {
struct utun_config* config = parse_config_from_buf(config_text, strlen(config_text), "android");
u_free(config_text);
if (!config) { IL_LOGE("parse_config_from_buf failed"); __atomic_store_n(&g_thread_running, 0, __ATOMIC_RELEASE); pthread_detach(pthread_self()); return NULL; }
config->global.client_type = CLIENT_TYPE_MOBILE; /* Android — всегда мобильный узел (keepalive standby/normal) */
install_crash_handlers();
@ -378,6 +379,7 @@ static void* instance_thread(void* arg) {
struct utun_config* config = parse_config_from_buf(cfg, strlen(cfg), "android");
u_free(cfg);
if (!config) { IL_LOGE("poll exit: parse_config failed"); break; }
config->global.client_type = CLIENT_TYPE_MOBILE; /* Android — всегда мобильный узел */
g_inst = utun_instance_create_from_config(g_ua, config);
my_inst = g_inst;

9
tools/chatgui-android/libutun_lite/standby.c

@ -378,6 +378,15 @@ void standby_notify_network_activity(void) {
standby_wake_all_waiters();
}
void standby_log_rx(const char* kind, const char* who) {
if (!g_enabled || g_phase != PHASE_SLEEP) return;
uint64_t slept = get_time_tb() - g_phase_start_tb;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG,
"standby: rx in SLEEP: %s from %s, slept %d.%ds",
kind ? kind : "?", who ? who : "?",
(int)(slept / STANDBY_TB_PER_SEC), (int)((slept % STANDBY_TB_PER_SEC) / 1000));
}
void standby_set_intervals_ms(int active_ms, int sleep_ms, int min_sleep_ms) {
if (active_ms <= 0 || sleep_ms <= 0) {
g_override = 0;

4
tools/chatgui-android/libutun_lite/standby.h

@ -50,6 +50,10 @@ void standby_wait_cancel(void* handle);
* Вызывается на uasync-потоке. */
void standby_notify_network_activity(void);
/* Диагностика: входящий пакет во время SLEEP-фазы (фазу НЕ меняет).
* kind — транспорт/тип пакета (строка), who — идентификатор отправителя. */
void standby_log_rx(const char* kind, const char* who);
/* Переопределить интервалы (мс), минуя chat_setting. Для тестов.
* active_ms/sleep_ms <= 0 → вернуться к chat_setting.
* min_sleep_ms может быть 0 (реагировать на сетевую активность сразу). */

Loading…
Cancel
Save