limingjie/OmniSocketGo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "feat: 把 A 端的 Session/KCP/视频/控制 都收口到一个本地 daemon 进程里，Django 和输入发送端都改成通过本机 UDS HTTP 去访问它，同时补齐了观测、性能和可用性上的几个关键问题。"

Browse Source 
This reverts commit 2f2c2008e7.
This commit is contained in:
Mock
2026-04-01 22:11:48 +08:00
parent 61b9d43413
commit b780d2e1cf
17 changed files with 1 additions and 1690 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
python/omnisocket/_omnisocket.c
View File

@@ -22,13 +22,6 @@ PyDoc_STRVAR(
"current frame has already been consumed and is lost."
);
PyDoc_STRVAR(
PyOmniSession_kcp_metrics_doc,
"kcp_metrics() -> dict\n"
"\n"
"Return a snapshot of low-level KCP metrics for the current session."
);
static PyObject *PyOmniSession_new(PyTypeObject *type, PyObject *args, PyObject *kwargs) {
PyOmniSession *self;
(void) args;
@@ -288,67 +281,6 @@ static PyObject *PyOmniSession_stats(PyOmniSession *self, PyObject *Py_UNUSED(ig
);
}
static PyObject *PyOmniSession_kcp_metrics(PyOmniSession *self, PyObject *Py_UNUSED(ignored)) {
omnisocket_session_kcp_metrics_t metrics;
memset(&metrics, 0, sizeof(metrics));
if (omnisocket_session_kcp_metrics_snapshot(&self->session, &metrics) != 0) {
return PyErr_SetFromErrno(PyExc_OSError);
}
return Py_BuildValue(
"{s:i,s:i,s:I,s:s,s:s,s:I,s:i,s:i,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K}",
"connected",
metrics.connected,
"has_conv",
metrics.has_conv,
"conv",
metrics.conv,
"local_addr",
metrics.local_addr,
"remote_addr",
metrics.remote_addr,
"rto_ms",
metrics.rto_ms,
"srtt_ms",
metrics.srtt_ms,
"srttvar_ms",
metrics.srttvar_ms,
"bytes_sent",
(unsigned long long) metrics.bytes_sent,
"bytes_received",
(unsigned long long) metrics.bytes_received,
"in_pkts",
(unsigned long long) metrics.in_pkts,
"out_pkts",
(unsigned long long) metrics.out_pkts,
"in_segs",
(unsigned long long) metrics.in_segs,
"out_segs",
(unsigned long long) metrics.out_segs,
"retrans_segs",
(unsigned long long) metrics.retrans_segs,
"fast_retrans_segs",
(unsigned long long) metrics.fast_retrans_segs,
"early_retrans_segs",
(unsigned long long) metrics.early_retrans_segs,
"lost_segs",
(unsigned long long) metrics.lost_segs,
"repeat_segs",
(unsigned long long) metrics.repeat_segs,
"in_errs",
(unsigned long long) metrics.in_errs,
"kcp_in_errs",
(unsigned long long) metrics.kcp_in_errs,
"ring_buffer_snd_queue",
(unsigned long long) metrics.ring_buffer_snd_queue,
"ring_buffer_rcv_queue",
(unsigned long long) metrics.ring_buffer_rcv_queue,
"ring_buffer_snd_buffer",
(unsigned long long) metrics.ring_buffer_snd_buffer
);
}
static PyMethodDef PyOmniSession_methods[] = {
{"connect", (PyCFunction) PyOmniSession_connect, METH_VARARGS | METH_KEYWORDS, NULL},
{"close", (PyCFunction) PyOmniSession_close, METH_NOARGS, NULL},
@@ -356,7 +288,6 @@ static PyMethodDef PyOmniSession_methods[] = {
{"recv", (PyCFunction) PyOmniSession_recv, METH_VARARGS | METH_KEYWORDS, PyOmniSession_recv_doc},
{"recv_into", (PyCFunction) PyOmniSession_recv_into, METH_VARARGS | METH_KEYWORDS, PyOmniSession_recv_into_doc},
{"stats", (PyCFunction) PyOmniSession_stats, METH_NOARGS, NULL},
{"kcp_metrics", (PyCFunction) PyOmniSession_kcp_metrics, METH_NOARGS, PyOmniSession_kcp_metrics_doc},
{NULL, NULL, 0, NULL}
};

69
python/omnisocket/omnisocket_client.c
View File

@@ -246,72 +246,3 @@ void omnisocket_session_stats_snapshot(omnisocket_session_t *session, omnisocket
*out_stats = session->stats;
pthread_mutex_unlock(&session->mutex);
}
int omnisocket_session_kcp_metrics_snapshot(
omnisocket_session_t *session,
omnisocket_session_kcp_metrics_t *out_metrics
) {
kcp_client_t *client = NULL;
kcp_conn_metrics_t metrics;
int rc = 0;
if (session == NULL || out_metrics == NULL) {
errno = EINVAL;
return -1;
}
memset(out_metrics, 0, sizeof(*out_metrics));
pthread_mutex_lock(&session->mutex);
if (session->client != NULL && !session->closing) {
client = session->client;
session->active_ops += 1;
}
pthread_mutex_unlock(&session->mutex);
if (client == NULL) {
return 0;
}
memset(&metrics, 0, sizeof(metrics));
rc = kcp_client_metrics_snapshot(client, &metrics);
pthread_mutex_lock(&session->mutex);
if (session->active_ops > 0) {
session->active_ops -= 1;
}
if (session->closing && session->active_ops == 0) {
pthread_cond_broadcast(&session->idle_cond);
}
pthread_mutex_unlock(&session->mutex);
if (rc != 0) {
return rc;
}
out_metrics->connected = metrics.connected;
out_metrics->has_conv = metrics.has_conv;
out_metrics->conv = metrics.conv;
snprintf(out_metrics->local_addr, sizeof(out_metrics->local_addr), "%s", metrics.local_addr);
snprintf(out_metrics->remote_addr, sizeof(out_metrics->remote_addr), "%s", metrics.remote_addr);
out_metrics->rto_ms = metrics.rto_ms;
out_metrics->srtt_ms = metrics.srtt_ms;
out_metrics->srttvar_ms = metrics.srttvar_ms;
out_metrics->bytes_sent = metrics.bytes_sent;
out_metrics->bytes_received = metrics.bytes_received;
out_metrics->in_pkts = metrics.in_pkts;
out_metrics->out_pkts = metrics.out_pkts;
out_metrics->in_segs = metrics.in_segs;
out_metrics->out_segs = metrics.out_segs;
out_metrics->retrans_segs = metrics.retrans_segs;
out_metrics->fast_retrans_segs = metrics.fast_retrans_segs;
out_metrics->early_retrans_segs = metrics.early_retrans_segs;
out_metrics->lost_segs = metrics.lost_segs;
out_metrics->repeat_segs = metrics.repeat_segs;
out_metrics->in_errs = metrics.in_errs;
out_metrics->kcp_in_errs = metrics.kcp_in_errs;
out_metrics->ring_buffer_snd_queue = metrics.ring_buffer_snd_queue;
out_metrics->ring_buffer_rcv_queue = metrics.ring_buffer_rcv_queue;
out_metrics->ring_buffer_snd_buffer = metrics.ring_buffer_snd_buffer;
return 0;
}

31
python/omnisocket/omnisocket_client.h
View File

@@ -14,33 +14,6 @@ typedef struct omnisocket_session_stats {
int connected;
} omnisocket_session_stats_t;
typedef struct omnisocket_session_kcp_metrics {
int connected;
int has_conv;
uint32_t conv;
char local_addr[OMNI_MAX_ADDR_TEXT];
char remote_addr[OMNI_MAX_ADDR_TEXT];
uint32_t rto_ms;
int32_t srtt_ms;
int32_t srttvar_ms;
uint64_t bytes_sent;
uint64_t bytes_received;
uint64_t in_pkts;
uint64_t out_pkts;
uint64_t in_segs;
uint64_t out_segs;
uint64_t retrans_segs;
uint64_t fast_retrans_segs;
uint64_t early_retrans_segs;
uint64_t lost_segs;
uint64_t repeat_segs;
uint64_t in_errs;
uint64_t kcp_in_errs;
uint64_t ring_buffer_snd_queue;
uint64_t ring_buffer_rcv_queue;
uint64_t ring_buffer_snd_buffer;
} omnisocket_session_kcp_metrics_t;
typedef struct omnisocket_session {
pthread_mutex_t mutex;
pthread_cond_t idle_cond;
@@ -74,9 +47,5 @@ int omnisocket_session_recv_into(
int timeout_ms
);
void omnisocket_session_stats_snapshot(omnisocket_session_t *session, omnisocket_session_stats_t *out_stats);
int omnisocket_session_kcp_metrics_snapshot(
omnisocket_session_t *session,
omnisocket_session_kcp_metrics_t *out_metrics
);
#endif

9
python/omnisocket_a_side/__init__.py
View File

@@ -1,9 +0,0 @@
from pathlib import Path
PACKAGE_ROOT = Path(__file__).resolve().parent
PYTHON_ROOT = PACKAGE_ROOT.parent
REPO_ROOT = PYTHON_ROOT.parent
DEFAULT_SOCKET_PATH = "/tmp/omnisocket-a-side.sock"
DEFAULT_CONFIG_PATH = REPO_ROOT / "config" / "a_side_omnidaemon.yaml"
VERSION = "0.1.0"

5
python/omnisocket_a_side/__main__.py
View File

@@ -1,5 +0,0 @@
from .daemon import main
if __name__ == "__main__":
main()

149
python/omnisocket_a_side/client.py
View File

@@ -1,149 +0,0 @@
"""Local Unix-domain HTTP client for the A-side OmniDaemon."""
from __future__ import annotations
import http.client
import json
import os
import socket
import threading
from typing import Any
from . import DEFAULT_SOCKET_PATH
class OmniDaemonError(RuntimeError):
def __init__(self, message: str, status_code: int | None = None) -> None:
super().__init__(message)
self.status_code = status_code
class UnixHTTPConnection(http.client.HTTPConnection):
def __init__(self, socket_path: str, timeout: float = 2.0) -> None:
super().__init__("localhost", timeout=timeout)
self.socket_path = socket_path
def connect(self) -> None: # pragma: no cover - runtime depends on Linux socket support
if not hasattr(socket, "AF_UNIX"):
raise OSError("AF_UNIX sockets are not available on this platform")
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.sock.settimeout(self.timeout)
self.sock.connect(self.socket_path)
class OmniDaemonClient:
def __init__(self, socket_path: str | None = None, timeout: float = 2.0) -> None:
self.socket_path = socket_path or os.getenv("OMNIDAEMON_SOCKET", DEFAULT_SOCKET_PATH)
self.timeout = timeout
self._local = threading.local()
def get_health(self) -> dict[str, Any]:
return self._request_json("GET", "/v1/health")
def get_state(self) -> dict[str, Any]:
return self._request_json("GET", "/v1/state")
def get_video_frame(self) -> bytes:
return self._request_bytes("GET", "/v1/video/frame")
def get_control_status(self) -> dict[str, Any]:
return self._request_json("GET", "/v1/control/status")
def send_control_event(
self,
*,
source: str,
event_code: str,
drive_value: float = 1.0,
client_time_ms: int | None = None,
) -> dict[str, Any]:
payload = {
"source": source,
"event_code": event_code,
"drive_value": float(drive_value),
"client_time_ms": client_time_ms,
}
return self._request_json("POST", "/v1/control/event", payload)
def close(self) -> None:
self._reset_connection()
def _request_json(
self,
method: str,
path: str,
payload: dict[str, Any] | None = None,
) -> dict[str, Any]:
raw = self._request_bytes(method, path, payload)
if not raw:
return {}
try:
return json.loads(raw.decode("utf-8"))
except json.JSONDecodeError as error:
raise OmniDaemonError(f"invalid daemon JSON response: {error}") from error
def _request_bytes(
self,
method: str,
path: str,
payload: dict[str, Any] | None = None,
) -> bytes:
body = b""
headers: dict[str, str] = {}
if payload is not None:
body = json.dumps(payload).encode("utf-8")
headers["Content-Type"] = "application/json"
headers["Content-Length"] = str(len(body))
headers.setdefault("Connection", "keep-alive")
for attempt in range(2):
connection = self._get_connection()
try:
connection.request(method, path, body=body, headers=headers)
response = connection.getresponse()
raw = response.read()
except FileNotFoundError as error:
self._reset_connection()
raise OmniDaemonError(
f"daemon socket not found: {self.socket_path}"
) from error
except (OSError, http.client.HTTPException) as error:
self._reset_connection()
if attempt == 0:
continue
raise OmniDaemonError(
f"daemon request failed via {self.socket_path}: {error}"
) from error
if getattr(response, "will_close", False):
self._reset_connection()
if response.status >= 400:
message = raw.decode("utf-8", errors="replace").strip() or response.reason
try:
parsed = json.loads(message)
if isinstance(parsed, dict) and "error" in parsed:
message = str(parsed["error"])
except json.JSONDecodeError:
pass
raise OmniDaemonError(message, status_code=response.status)
return raw
raise OmniDaemonError(f"daemon request failed via {self.socket_path}: retry exhausted")
def _get_connection(self) -> UnixHTTPConnection:
connection = getattr(self._local, "connection", None)
if connection is None:
connection = UnixHTTPConnection(self.socket_path, timeout=self.timeout)
self._local.connection = connection
return connection
def _reset_connection(self) -> None:
connection = getattr(self._local, "connection", None)
if connection is None:
return
try:
connection.close()
except OSError:
pass
self._local.connection = None

90
python/omnisocket_a_side/control_codec.py
View File

@@ -1,90 +0,0 @@
"""Binary control packet codec shared by the daemon and local clients."""
from __future__ import annotations
from dataclasses import dataclass
import struct
import time
CONTROL_PACKET_VERSION = 1
CONTROL_PACKET_STRUCT = struct.Struct("!BBHIfQ")
EVENT_NAME_TO_ID = {
"pose_home": 1,
"pose_hold": 2,
"mode_stride": 3,
"surge_up": 6,
"surge_down": 7,
"sway_left": 8,
"sway_right": 9,
"spin_left": 10,
"spin_right": 11,
"set_surge": 12,
"set_sway": 13,
"set_spin": 14,
"set_lift": 15,
"lift_up": 16,
"lift_down": 17,
"trim_reset": 18,
"session_quit": 19,
}
EVENT_ID_TO_NAME = {value: key for key, value in EVENT_NAME_TO_ID.items()}
ANALOG_EVENT_CODES = {"set_surge", "set_sway", "set_spin"}
@dataclass(slots=True)
class ControlPacket:
seq_id: int
event_id: int
drive_value: float = 1.0
sent_at_ns: int = 0
@property
def event_name(self) -> str:
return EVENT_ID_TO_NAME.get(self.event_id, f"unknown_{self.event_id}")
def encode(self) -> bytes:
sent_at_ns = self.sent_at_ns or time.time_ns()
return CONTROL_PACKET_STRUCT.pack(
CONTROL_PACKET_VERSION,
self.event_id,
0,
int(self.seq_id),
float(self.drive_value),
int(sent_at_ns),
)
@classmethod
def decode(cls, payload: bytes) -> "ControlPacket":
if len(payload) != CONTROL_PACKET_STRUCT.size:
raise ValueError(
f"invalid control packet length {len(payload)}; "
f"want {CONTROL_PACKET_STRUCT.size}"
)
version, event_id, _reserved, seq_id, drive_value, sent_at_ns = (
CONTROL_PACKET_STRUCT.unpack(payload)
)
if version != CONTROL_PACKET_VERSION:
raise ValueError(f"unsupported control packet version {version}")
return cls(
seq_id=int(seq_id),
event_id=int(event_id),
drive_value=float(drive_value),
sent_at_ns=int(sent_at_ns),
)
def make_control_packet(
seq_id: int,
event_name: str,
drive_value: float = 1.0,
sent_at_ns: int | None = None,
) -> ControlPacket:
event_id = EVENT_NAME_TO_ID[event_name]
return ControlPacket(
seq_id=seq_id,
event_id=event_id,
drive_value=drive_value,
sent_at_ns=time.time_ns() if sent_at_ns is None else sent_at_ns,
)

1063
python/omnisocket_a_side/daemon.py
View File

File diff suppressed because it is too large Load Diff

7
python/setup.py
View File

@@ -35,12 +35,7 @@ COMMON_SOURCES = [
setup(
name="omnisocket",
version="0.1.0",
packages=["omnisocket", "omnisocket_a_side"],
entry_points={
"console_scripts": [
"omnisocket-a-side-daemon=omnisocket_a_side.daemon:main",
]
},
packages=["omnisocket"],
ext_modules=[
Extension(
"omnisocket._omnisocket",