feat(vqt6/requests/vipersimd/vthreading): 气泡聊天与流式 https 阶段改动、vipersimd FMA 批量内核, 清理 vqt6 copy 与 requests.vp

This commit is contained in:
TermiNexus committed 2026-09-09 21:10:14 +08:00
1 parent 44e0e898d3
commit 7d092f25b1
26 files changed
+830 -6560

No files matched your search

+417 -9
View File
@@ -6,6 +6,8 @@ import stdio
import stdlib
import requests
import openssl
import json
import memhub
# ============================================================
@@ -38,6 +40,65 @@ _scratch_store: t.CArray[t.CChar, 512] # 临时拼接缓冲
_last_verify_val: INT = -1 # 最近一次握手后 SSL_get_verify_result(0=通过;-1=未启用)
_last_verify_flag: INT = -1 # 最近一次请求 verify 标志(1=开启;0=关闭;-1=未知)
# ============================================================
# SSE 流式内容接收区(post_stream 专用;单路请求,UI 线程轮询拉取)
# _st_len: 工作线程已写入的内容字节数
# _st_consumed: UI 已消费游标(仅拉取方推进)
# _st_state: 0=运行中 1=完成(2xx) 2=失败
# _st_status: HTTP 状态码(失败网络路径为负错误码)
# ============================================================
STREAM_CAP: t.CDefine = 262144
_st_buf: t.CArray[t.CChar, 262144]
_st_len: t.CSizeT = 0
_st_consumed: t.CSizeT = 0
_st_state: INT = 0
_st_status: INT = 0
_st_err: t.CArray[t.CChar, 512]
def stream_begin():
"""开始新一轮流式接收:重置缓冲/游标/状态。"""
global _st_len, _st_consumed, _st_state, _st_status
_st_len = 0
_st_consumed = 0
_st_state = 0
_st_status = 0
_st_err[0] = 0
def stream_pull(out: str, cap: INT) -> INT:
"""拉取未消费的增量内容到 out(自动补 0),返回拷贝字节数。"""
global _st_consumed
if out is None or cap <= 1:
return 0
avail: INT = 0
if _st_len > _st_consumed:
avail = INT(_st_len - _st_consumed)
if avail > cap - 1:
avail = cap - 1
i: t.CSizeT = 0
while i < t.CSizeT(avail):
out[i] = _st_buf[_st_consumed + i]
i += 1
out[t.CSizeT(avail)] = 0
_st_consumed += t.CSizeT(avail)
return avail
def stream_state() -> INT:
"""0=运行中 1=完成(2xx) 2=失败。"""
return _st_state
def stream_status() -> INT:
"""HTTP 状态码;负数为网络/解析错误。"""
return _st_status
def stream_error_text() -> str:
"""失败原因文本(_st_state==2 时)。"""
return _st_err
def last_verify_result() -> int:
"""返回最近一次握手的证书校验结果:0 通过;>0 失败码;-1 未开启校验。"""
@@ -163,11 +224,13 @@ def _TlsErrText():
# 核心请求
# ============================================================
def _HttpsRequest(method: str, url: str, verify: bool, ca_file: str, data: str, content_type: str) -> requests.Response | t.CPtr:
def _HttpsRequest(method: str, url: str, verify: bool, ca_file: str, data: str, content_type: str, auth: str = None) -> requests.Response | t.CPtr:
"""HTTPS GET/POST 核心实现。
verify=False 时跳过证书链/主机名校验(不推荐);verify=True 时
ca_file 提供 PEM CA bundle,否则使用系统默认 CA 路径。
auth(可选):形如 "Bearer <key>" 的 Authorization 头值,
非空时仅在 POST 请求头中插入 Authorization 头(默认 None = 行为不变)。
"""
global _last_verify_val, _last_verify_flag
if requests._mbuddy is None: return None
@@ -333,9 +396,17 @@ def _HttpsRequest(method: str, url: str, verify: bool, ca_file: str, data: str,
if is_post:
body_len = INT(string.strlen(data)) if data is not None else 0
ct: str = content_type if content_type is not None else "application/x-www-form-urlencoded"
stdio.snprintf(req_buf, requests.HTTP_BUFSIZ,
"POST %s HTTP/1.1\r\nHost: %s\r\nContent-Type: %s\r\nContent-Length: %d\r\nAccept-Encoding: gzip\r\nConnection: close\r\n\r\n",
path_buf, host_buf, ct, body_len)
# 分段组头,便于按需插入 Authorization(auth 非空才加,默认路径输出与原来完全一致)
hdr_off: INT = 0
hdr_cap: INT = requests.HTTP_BUFSIZ
stdio.snprintf(req_buf, hdr_cap, "POST %s HTTP/1.1\r\nHost: %s\r\n", path_buf, host_buf)
hdr_off = INT(string.strlen(req_buf))
if auth is not None and auth[0] != 0:
stdio.snprintf(req_buf + hdr_off, hdr_cap - hdr_off, "Authorization: %s\r\n", auth)
hdr_off = INT(string.strlen(req_buf))
stdio.snprintf(req_buf + hdr_off, hdr_cap - hdr_off,
"Content-Type: %s\r\nContent-Length: %d\r\nAccept-Encoding: gzip\r\nConnection: close\r\n\r\n",
ct, body_len)
else:
stdio.snprintf(req_buf, requests.HTTP_BUFSIZ,
"GET %s HTTP/1.1\r\nHost: %s\r\nAccept-Encoding: gzip\r\nConnection: close\r\n\r\n",
@@ -396,8 +467,12 @@ def _HttpsRequest(method: str, url: str, verify: bool, ca_file: str, data: str,
# 公共 API
# ============================================================
def _HttpsRedirectLoop(method: str, url: str, verify: bool, ca_file: str, data: str, content_type: str) -> requests.Response | t.CPtr:
"""HTTPS GET/POST + 重定向跟随(跨 http/https,最多 5 跳)。"""
def _HttpsRedirectLoop(method: str, url: str, verify: bool, ca_file: str, data: str, content_type: str, auth: str = None) -> requests.Response | t.CPtr:
"""HTTPS GET/POST + 重定向跟随(跨 http/https,最多 5 跳)。
auth(可选)原样透传给 _HttpsRequest;None 时行为与原来完全一致。
注意:跨主机重定向时 Authorization 同样会透传(仅提示,逻辑未改动)。
"""
if requests._mbuddy is None: return None
cur: str = requests._mbuddy.alloc(2048)
@@ -410,7 +485,7 @@ def _HttpsRedirectLoop(method: str, url: str, verify: bool, ca_file: str, data:
hop: INT = 0
while hop < 5:
resp: requests.Response | t.CPtr = _HttpsRequest(method, cur, verify, ca_file, data, content_type)
resp: requests.Response | t.CPtr = _HttpsRequest(method, cur, verify, ca_file, data, content_type, auth)
if resp is None:
requests._mbuddy.free(cur)
return None
@@ -475,7 +550,7 @@ def get(url: str, verify: bool = True, ca_file: str = None) -> requests.Response
def post(url: str, data: str = None, content_type: str = "application/x-www-form-urlencoded",
verify: bool = True, ca_file: str = None) -> requests.Response | t.CPtr:
verify: bool = True, ca_file: str = None, auth: str = None) -> requests.Response | t.CPtr:
"""HTTPS POST request(自动跟随 3xx 重定向,最多 5 跳)。
Args:
@@ -484,10 +559,343 @@ def post(url: str, data: str = None, content_type: str = "application/x-www-form
content_type: Content-Type header
verify: True 时校验证书链 + 主机名(默认)
ca_file: PEM CA bundle 路径;None 时使用系统默认 CA 路径
auth: 可选 Authorization 头值(形如 "Bearer <key>"),
非空时随请求头发送;None/空串 = 与旧版完全一致。
Returns:
Pointer to a Response object, None on failure (see last_error()).
"""
return _HttpsRedirectLoop("POST", url, verify, ca_file, data, content_type)
return _HttpsRedirectLoop("POST", url, verify, ca_file, data, content_type, auth)
# ============================================================
# SSE 流式 Chat Completions(post_stream)
#
# 与 post() 不同:请求头带 Accept: text/event-stream;响应体按行解析
# SSE(data: {json}),把每段 choices[0].delta.content 抽进 _st_buf 供
# UI 线程 stream_pull() 增量拉取显示。全程阻塞读取,应在后台线程调用。
#
# 返回: >=0 HTTP 状态码(2xx 正常完成;3xx/4xx/5xx 状态仍返回给调用方,
# 正文中的非 SSE 内容不会进 _st_buf);-1/-2 网络/解析失败
# (_st_state=2,失败原因见 stream_error_text()/last_error())。
# ============================================================
def _SseDeltaContent(root: json.JsonValue | t.CPtr) -> str:
"""从流式 data 事件取 choices[0].delta.content;缺失返回 None。"""
if root is None or not root.is_object():
return None
cv: json.JsonValue | t.CPtr = root["choices"]
if cv is None or not cv.is_array():
return None
m0: json.JsonValue | t.CPtr = cv.get_item(0)
if m0 is None or not m0.is_object():
return None
dv: json.JsonValue | t.CPtr = m0["delta"]
if dv is None or not dv.is_object():
return None
content_v: json.JsonValue | t.CPtr = dv["content"]
if content_v is None or not content_v.is_string():
return None
return content_v.as_string()
def _StreamAppendContent(txt: str):
"""把一段增量内容追加到 _st_buf(超容量则截断丢弃并记录)。"""
global _st_len
if txt is None:
return
pos: t.CSizeT = _st_len
i: t.CSizeT = 0
while txt[i] != 0:
if pos >= t.CSizeT(STREAM_CAP) - 1:
break
_st_buf[pos] = txt[i]
pos += 1
i += 1
_st_len = pos
def _SetStreamFail(rc: INT):
"""流式失败收尾:从 _err_store 拷原因文本并置状态。"""
global _st_state, _st_status
_st_state = 2
_st_status = rc
stdio.snprintf(_st_err, 512, "%s", _err_store)
def post_stream(url: str, data: str, content_type: str = "application/json",
verify: bool = True, ca_file: str = None, auth: str = None) -> INT:
"""SSE 流式 POST(阻塞,应在后台线程调用)。
data 应为已含 "stream":true 的 JSON 请求体。
返回 HTTP 状态码;负数为网络/解析失败(_st_state=2)。
"""
global _st_state, _st_status
if requests._mbuddy is None:
return -1
_st_state = 0
_st_status = 0
# Winsock/openssl 初始化(幂等;流式首次可能在后台线程执行)
if _EnsureReady() != 0:
_SetStreamFail(-1)
return -1
# ---- 连接/TLS(与 _HttpsRequest 一致)----
errmsg: str = requests._mbuddy.alloc(512)
if errmsg is None:
return -1
host_buf: t.CArray[t.CChar, 256]
path_buf: t.CArray[t.CChar, 1024]
port: INT = 0
if _ParseHttpsUrl(url, host_buf, 256, path_buf, 1024, c.Addr(port)) != 0:
stdio.snprintf(errmsg, 512, "bad https url: %s", url)
_RecordErrorText(errmsg)
_SetStreamFail(-1)
requests._mbuddy.free(errmsg)
return -1
sock = socket.Socket(socket.AF_INET, socket.SOCK_STREAM)
if sock.fd == socket.INVALID_SOCKET:
_RecordErrorText("socket() failed")
_SetStreamFail(-1)
requests._mbuddy.free(errmsg)
return -1
if sock.connect(host_buf, port) != 0:
stdio.snprintf(errmsg, 512, "connect %s:%d failed", host_buf, port)
_RecordErrorText(errmsg)
_SetStreamFail(-1)
sock.close()
requests._mbuddy.free(errmsg)
return -1
ctx: openssl.SslCtxPtr = openssl.SSL_CTX_new(openssl.TLS_client_method())
if ctx is None:
_RecordErrorText("SSL_CTX_new failed")
_SetStreamFail(-1)
sock.close()
requests._mbuddy.free(errmsg)
return -1
if verify:
openssl.SSL_CTX_set_verify(ctx, openssl.SslVerifyPeer, None)
if ca_file is not None and ca_file[0] != 0:
if openssl.SSL_CTX_load_verify_locations(ctx, ca_file, None) != 1:
_TlsErrText()
stdio.snprintf(errmsg, 512, "load CA file failed: %s (%s)", ca_file, _ScratchBuf())
_RecordErrorText(errmsg)
_SetStreamFail(-1)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
return -1
else:
openssl.SSL_CTX_set_default_verify_paths(ctx)
else:
openssl.SSL_CTX_set_verify(ctx, openssl.SslVerifyNone, None)
ssl: openssl.SslPtr = openssl.SSL_new(ctx)
if ssl is None:
_RecordErrorText("SSL_new failed")
_SetStreamFail(-1)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
return -1
openssl.SSL_set_fd(ssl, sock.fd)
openssl.SSL_ctrl(ssl, openssl.SslCtrlSetTlsextHostname, openssl.TlsextNametypeHostName, host_buf)
if verify:
openssl.SSL_set1_host(ssl, host_buf)
hret: INT = openssl.SSL_connect(ssl)
if hret <= 0:
want: INT = openssl.SSL_get_error(ssl, hret)
_TlsErrText()
stdio.snprintf(errmsg, 512, "SSL_connect failed (want=%d): %s", want, _ScratchBuf())
_RecordErrorText(errmsg)
_SetStreamFail(-1)
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
return -1
if verify:
vres: t.CInt32T = openssl.SSL_get_verify_result(ssl)
if vres != 0:
reason: str = openssl.X509_verify_cert_error_string(vres)
if reason is None:
stdio.snprintf(errmsg, 512, "SSL certificate verify failed: error %d", INT(vres))
else:
stdio.snprintf(errmsg, 512, "SSL certificate verify failed (%d): %s", INT(vres), reason)
_RecordErrorText(errmsg)
_SetStreamFail(-1)
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
return -1
# ---- 组请求头(POST + Accept: text/event-stream)并发送 ----
req_buf: str = requests._mbuddy.alloc(requests.HTTP_BUFSIZ)
if req_buf is None:
_RecordErrorText("alloc request buffer failed")
_SetStreamFail(-1)
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
return -1
body_len: INT = INT(string.strlen(data)) if data is not None else 0
ct: str = content_type if content_type is not None else "application/json"
hdr_off: INT = 0
stdio.snprintf(req_buf, requests.HTTP_BUFSIZ, "POST %s HTTP/1.1\r\nHost: %s\r\n", path_buf, host_buf)
hdr_off = INT(string.strlen(req_buf))
if auth is not None and auth[0] != 0:
stdio.snprintf(req_buf + hdr_off, requests.HTTP_BUFSIZ - hdr_off, "Authorization: %s\r\n", auth)
hdr_off = INT(string.strlen(req_buf))
stdio.snprintf(req_buf + hdr_off, requests.HTTP_BUFSIZ - hdr_off,
"Content-Type: %s\r\nContent-Length: %d\r\nAccept: text/event-stream\r\nConnection: close\r\n\r\n",
ct, body_len)
hdr_len: INT = INT(string.strlen(req_buf))
sent: INT = openssl.SSL_write(ssl, req_buf, hdr_len)
if sent <= 0:
_TlsErrText()
stdio.snprintf(errmsg, 512, "SSL_write header failed: %s", _ScratchBuf())
_RecordErrorText(errmsg)
_SetStreamFail(-1)
requests._mbuddy.free(req_buf)
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
return -1
if body_len > 0 and data is not None:
openssl.SSL_write(ssl, data, body_len)
requests._mbuddy.free(req_buf)
# ---- 读取响应:跳过响应头 → SSE 行解析 → 抽 delta.content ----
rd: str = requests._mbuddy.alloc(16384)
if rd is None:
_RecordErrorText("alloc stream buffer failed")
_SetStreamFail(-1)
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
return -1
# JSON 解析池(事件粒度 reset,足够容纳单个 data 事件)
pmem: str = stdlib.malloc(524288)
if pmem is None:
_RecordErrorText("alloc json pool failed")
_SetStreamFail(-1)
requests._mbuddy.free(rd)
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
return -1
jpool: memhub.MemPool | t.CPtr = memhub.MemPool(pmem, 524288)
leftover: t.CArray[t.CChar, 16384]
llen: INT = 0
header_done: INT = 0
status_code: INT = 0
had_read: INT = 0
while True:
n: INT = openssl.SSL_read(ssl, rd, 16384)
if n <= 0:
break
had_read = 1
# 1) 追加到 leftover
ai: INT = 0
while ai < n and llen < 16383:
leftover[llen] = rd[ai]
llen += 1
ai += 1
# 2) 逐行处理
while True:
# 找行尾 \n
nl: INT = -1
k: INT = 0
while k < llen:
if leftover[k] == 10:
nl = k
break
k += 1
if nl < 0:
break
e: INT = nl
if e > 0 and leftover[e - 1] == 13:
e -= 1
leftover[e] = 0
line: str = leftover
if header_done == 0:
if string.strncmp(line, "HTTP/", 5) == 0:
# 状态码:HTTP/1.1 200 OK
sp: INT = 0
while line[sp] != 0 and line[sp] != ' ':
sp += 1
if line[sp] == ' ':
sp += 1
code_val: INT = 0
while line[sp] >= '0' and line[sp] <= '9':
code_val = code_val * 10 + INT(line[sp] - '0')
sp += 1
status_code = code_val
if e == 0:
header_done = 1
else:
if e > 0 and line[0] != ':':
if string.strncmp(line, "data:", 5) == 0:
pl: str = line + 5
while pl[0] == ' ':
pl += 1
if pl[0] != 0:
if string.strncmp(pl, "[DONE]", 6) != 0 or pl[6] != 0:
jpool.reset()
ev: json.JsonValue | t.CPtr = json.parse(jpool, pl)
dtx: str = _SseDeltaContent(ev)
if dtx is not None:
_StreamAppendContent(dtx)
# 3) 移除已处理行
rem: INT = llen - (nl + 1)
j2: INT = 0
while j2 < rem:
leftover[j2] = leftover[nl + 1 + j2]
j2 += 1
llen = rem
# ---- 收尾 ----
openssl.SSL_shutdown(ssl)
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(rd)
stdlib.free(pmem)
requests._mbuddy.free(errmsg)
if status_code == 0:
if had_read == 0:
_RecordErrorText("no response data")
else:
_RecordErrorText("missing HTTP status line")
_SetStreamFail(-2)
return -2
_st_status = status_code
if status_code >= 200 and status_code < 300:
_st_state = 1
else:
_st_state = 2
stdio.snprintf(_st_err, 512, "HTTP %d(请检查 Key/模型/地址或账号额度)", status_code)
return status_code
# ============================================================