feat(ncm §21): 边下边播阶段1+2——stream_src流式源+decoder双路open_cb+下载线程预缓冲/stall联动+P2 HTTP空闲超时+cache fd写者+stream_sim全矩阵PCM金标准(0真实请求,未部署)
This commit is contained in:
MeowOS committed
1 parent
dbee37e1d6
commit
9e25db77cc
18 files changed
+1335
-136
No files matched your search
@@ -0,0 +1,299 @@
|
||||
/* play/stream_src.c — 边下边播共享流源实现（接口与契约见 stream_src.h，REPORT §21）。
|
||||
*
|
||||
* 关键实测事实（NCM_STREAM_PLAY.md，红线）：
|
||||
* F7 read_cb 必须严格 fread 语义（填满才返回，真 EOF 才短读），否则 minimp3 假播完；
|
||||
* F5 写端中断（final=截断点）→ 读端短读 → 解码干净结束于截断点；
|
||||
* F4 seek 到未下载区只是改 read_pos，read 自然阻塞等待，数据到后继续出声。
|
||||
*
|
||||
* 等待/唤醒：pthread condvar + 100ms timedwait 兜底（防丢唤醒）。
|
||||
* 再缓冲迟滞（§4.2）：单次调用累计等待 >500ms 后，要求 written-read_pos >= 64KB 才放行
|
||||
* （防「每追上 1 帧就出声」的爆音感）；final/failed/abort 时不受限。
|
||||
*/
|
||||
#include "stream_src.h"
|
||||
|
||||
#include <fcntl.h>
|
||||
#include <pthread.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <time.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#define WAIT_SLICE_MS 100 /* cond timedwait 兜底片长 */
|
||||
#define REBUFFER_AFTER_MS 500 /* 单次调用累计等待超此值后启用再缓冲迟滞 */
|
||||
#define REBUFFER_MIN (64 * 1024) /* 迟滞放行门槛（≈1.6s@320kbps） */
|
||||
#define STALL_MS 1000 /* 单次调用等待超此值 → 置 stalled（播放线程 pause 联动） */
|
||||
|
||||
struct stream_src {
|
||||
int fd; /* 只读 fd（pread；写端另有自己的写 fd） */
|
||||
pthread_mutex_t mu;
|
||||
pthread_cond_t cv; /* written/final/failed/abort 变化时 broadcast */
|
||||
uint64_t written; /* 已落盘字节（单调增，写端推进） */
|
||||
uint64_t final; /* 0=下载中；>0=最终大小（完成=全长，中断=截断点） */
|
||||
int failed; /* <0=下载/IO 错误码 */
|
||||
int abort; /* 切歌/停止：读端按 EOF 处理，写端回调返回 NZ */
|
||||
uint64_t read_pos; /* 解码逻辑读指针（seek 回调改写；允许 > written） */
|
||||
/* 统计 */
|
||||
uint64_t wait_ms_total;
|
||||
int underruns;
|
||||
int stalled;
|
||||
};
|
||||
|
||||
static int64_t mono_ms(void)
|
||||
{
|
||||
struct timespec ts;
|
||||
clock_gettime(CLOCK_MONOTONIC, &ts);
|
||||
return (int64_t)ts.tv_sec * 1000 + ts.tv_nsec / 1000000;
|
||||
}
|
||||
|
||||
static int src_alloc(stream_src_t **out, int fd)
|
||||
{
|
||||
stream_src_t *s = (stream_src_t *)calloc(1, sizeof *s);
|
||||
if (!s) { close(fd); return -1; }
|
||||
s->fd = fd;
|
||||
pthread_mutex_init(&s->mu, NULL);
|
||||
pthread_cond_init(&s->cv, NULL);
|
||||
*out = s;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int stream_src_open_stream(stream_src_t **out, const char *tmp_path)
|
||||
{
|
||||
if (!out || !tmp_path) return -1;
|
||||
int fd = open(tmp_path, O_RDONLY);
|
||||
if (fd < 0) return -1;
|
||||
return src_alloc(out, fd); /* written=0/final=0：下载中 */
|
||||
}
|
||||
|
||||
int stream_src_open_static(stream_src_t **out, const char *path, uint64_t size)
|
||||
{
|
||||
if (!out || !path) return -1;
|
||||
int fd = open(path, O_RDONLY);
|
||||
if (fd < 0) return -1;
|
||||
int rc = src_alloc(out, fd);
|
||||
if (rc == 0) {
|
||||
(*out)->written = size;
|
||||
(*out)->final = size ? size : 1; /* 空文件也置 final（read 立即 EOF） */
|
||||
}
|
||||
return rc;
|
||||
}
|
||||
|
||||
void stream_src_close(stream_src_t *s)
|
||||
{
|
||||
if (!s) return;
|
||||
if (s->fd >= 0) close(s->fd);
|
||||
pthread_mutex_destroy(&s->mu);
|
||||
pthread_cond_destroy(&s->cv);
|
||||
free(s);
|
||||
}
|
||||
|
||||
void stream_src_note_written(stream_src_t *s, uint64_t n)
|
||||
{
|
||||
if (!s || !n) return;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
s->written += n;
|
||||
pthread_cond_broadcast(&s->cv);
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
}
|
||||
|
||||
void stream_src_finish(stream_src_t *s, int err)
|
||||
{
|
||||
if (!s) return;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
s->final = s->written ? s->written : 1; /* 0 字节下载也视为结束（final=1 > written=0） */
|
||||
if (err) s->failed = err;
|
||||
pthread_cond_broadcast(&s->cv);
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
}
|
||||
|
||||
void stream_src_abort(stream_src_t *s)
|
||||
{
|
||||
if (!s) return;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
s->abort = 1;
|
||||
pthread_cond_broadcast(&s->cv);
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
}
|
||||
|
||||
int stream_src_aborted(stream_src_t *s)
|
||||
{
|
||||
if (!s) return 1;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
int a = s->abort;
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
return a;
|
||||
}
|
||||
|
||||
int stream_src_wait(stream_src_t *s, uint64_t need, int timeout_ms)
|
||||
{
|
||||
if (!s) return -1;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
if (s->written >= need) { pthread_mutex_unlock(&s->mu); return 1; }
|
||||
if (s->final || s->failed || s->abort) { pthread_mutex_unlock(&s->mu); return -1; }
|
||||
struct timespec ts;
|
||||
clock_gettime(CLOCK_REALTIME, &ts);
|
||||
ts.tv_nsec += (long)timeout_ms * 1000000L;
|
||||
if (ts.tv_nsec >= 1000000000L) { ts.tv_sec++; ts.tv_nsec -= 1000000000L; }
|
||||
pthread_cond_timedwait(&s->cv, &s->mu, &ts);
|
||||
int r;
|
||||
if (s->written >= need) r = 1;
|
||||
else if (s->final || s->failed || s->abort) r = -1;
|
||||
else r = 0;
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
return r;
|
||||
}
|
||||
|
||||
size_t stream_src_peek(stream_src_t *s, void *buf, size_t n, uint64_t off)
|
||||
{
|
||||
if (!s || !buf || !n) return 0;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
uint64_t avail = (s->written > off) ? s->written - off : 0;
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
if (!avail) return 0;
|
||||
if (n > avail) n = (size_t)avail;
|
||||
ssize_t r = pread(s->fd, buf, n, (off_t)off);
|
||||
return r > 0 ? (size_t)r : 0;
|
||||
}
|
||||
|
||||
/* 核心读：严格 fread 语义。minimp3（MP3D_READ_CB）与 dr_flac（drflac_read_proc）
|
||||
* 共用——dr_flac 接受「实际读到的字节数」，尽力填满是其超集，两契约同满足。 */
|
||||
static size_t src_read(void *buf, size_t size, void *ud)
|
||||
{
|
||||
stream_src_t *s = (stream_src_t *)ud;
|
||||
size_t done = 0;
|
||||
int64_t waited = 0; /* 本次调用累计等待 ms */
|
||||
int waited_this_call = 0; /* 本次调用是否已进入过等待（underrun 按调用计一次） */
|
||||
pthread_mutex_lock(&s->mu);
|
||||
while (done < size) {
|
||||
uint64_t avail = (s->written > s->read_pos) ? s->written - s->read_pos : 0;
|
||||
/* 再缓冲迟滞：本轮累计等过 >500ms 后，前方不足 64KB 且未结束 → 继续等
|
||||
*（仅本轮尚未产出字节时下闸；已有产出则尽力填满）。 */
|
||||
int gated = (done == 0 && avail > 0 && waited > REBUFFER_AFTER_MS &&
|
||||
avail < REBUFFER_MIN && !s->final && !s->failed && !s->abort);
|
||||
if (avail > 0 && !gated) {
|
||||
size_t n = (size - done) < (size_t)avail ? (size - done) : (size_t)avail;
|
||||
uint64_t pos = s->read_pos;
|
||||
pthread_mutex_unlock(&s->mu); /* pread 期间放锁，不阻塞写端 broadcast */
|
||||
ssize_t r = pread(s->fd, (char *)buf + done, n, (off_t)pos);
|
||||
pthread_mutex_lock(&s->mu);
|
||||
if (r <= 0) { /* 盘 IO 错：按中断收尾（真 EOF 语义） */
|
||||
if (!s->failed) s->failed = -3; /* NCM_ERR_IO（本层不依赖 core 头） */
|
||||
s->final = s->written ? s->written : 1;
|
||||
pthread_cond_broadcast(&s->cv);
|
||||
break;
|
||||
}
|
||||
s->read_pos += (uint64_t)r;
|
||||
done += (size_t)r;
|
||||
continue;
|
||||
}
|
||||
if (s->final || s->failed || s->abort) break; /* 真 EOF：短读/0 返回 */
|
||||
/* 追上下载：记 underrun（每次 read 调用至多计一次——minimp3 一次 refill 常先给
|
||||
* 部分数据再等待，done==0 口径会漏记；按「本调用发生了等待」计数） */
|
||||
if (!waited_this_call) { waited_this_call = 1; s->underruns++; }
|
||||
int64_t t0 = mono_ms();
|
||||
struct timespec ts;
|
||||
clock_gettime(CLOCK_REALTIME, &ts);
|
||||
ts.tv_nsec += WAIT_SLICE_MS * 1000000L;
|
||||
if (ts.tv_nsec >= 1000000000L) { ts.tv_sec++; ts.tv_nsec -= 1000000000L; }
|
||||
pthread_cond_timedwait(&s->cv, &s->mu, &ts);
|
||||
int64_t dt = mono_ms() - t0;
|
||||
if (dt > 0) { waited += dt; s->wait_ms_total += (uint64_t)dt; }
|
||||
if (waited > STALL_MS) s->stalled = 1;
|
||||
}
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
return done;
|
||||
}
|
||||
|
||||
size_t stream_src_read_cb(void *buf, size_t size, void *ud) { return src_read(buf, size, ud); }
|
||||
|
||||
int stream_src_seek_cb(uint64_t pos, void *ud)
|
||||
{
|
||||
stream_src_t *s = (stream_src_t *)ud;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
s->read_pos = pos; /* 允许 > written：读端自然阻塞等待（F4） */
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
return 0;
|
||||
}
|
||||
|
||||
size_t stream_src_drflac_read(void *ud, void *buf, size_t bytes_to_read)
|
||||
{
|
||||
return src_read(buf, bytes_to_read, ud);
|
||||
}
|
||||
|
||||
int stream_src_drflac_seek(void *ud, int offset, int origin)
|
||||
{
|
||||
stream_src_t *s = (stream_src_t *)ud;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
if (origin == STREAM_SRC_SEEK_SET) {
|
||||
s->read_pos = (offset < 0) ? 0 : (uint64_t)offset;
|
||||
} else if (origin == STREAM_SRC_SEEK_END) {
|
||||
/* dr_flac open 期用 SEEK_END+tell 取文件总长做元数据校验；下载中不知总长，
|
||||
* 按 dr_flac 约定返回 FALSE（hasKnownFileSize=false → 跳过该校验）。
|
||||
* 静态源/已完成的下载 final 已知 → 正常定位。 */
|
||||
if (!s->final) { pthread_mutex_unlock(&s->mu); return 0; }
|
||||
int64_t np = (int64_t)s->final + offset;
|
||||
if (np < 0) { pthread_mutex_unlock(&s->mu); return 0; }
|
||||
s->read_pos = (uint64_t)np;
|
||||
} else {
|
||||
int64_t np = (int64_t)s->read_pos + offset;
|
||||
if (np < 0) { pthread_mutex_unlock(&s->mu); return 0; } /* drflac_bool32 false */
|
||||
s->read_pos = (uint64_t)np;
|
||||
}
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
return 1; /* drflac_bool32 true */
|
||||
}
|
||||
|
||||
int stream_src_drflac_tell(void *ud, int64_t *cursor)
|
||||
{
|
||||
stream_src_t *s = (stream_src_t *)ud;
|
||||
if (!cursor) return 0;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
*cursor = (int64_t)s->read_pos;
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
return 1;
|
||||
}
|
||||
|
||||
void stream_src_stats(stream_src_t *s, int *underruns, uint64_t *wait_ms)
|
||||
{
|
||||
if (!s) return;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
if (underruns) *underruns = s->underruns;
|
||||
if (wait_ms) *wait_ms = s->wait_ms_total;
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
}
|
||||
|
||||
int stream_src_take_stalled(stream_src_t *s)
|
||||
{
|
||||
if (!s) return 0;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
int v = s->stalled;
|
||||
s->stalled = 0;
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
return v;
|
||||
}
|
||||
|
||||
uint64_t stream_src_written(stream_src_t *s)
|
||||
{
|
||||
if (!s) return 0;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
uint64_t w = s->written;
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
return w;
|
||||
}
|
||||
|
||||
uint64_t stream_src_final(stream_src_t *s)
|
||||
{
|
||||
if (!s) return 1;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
uint64_t f = s->final;
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
return f;
|
||||
}
|
||||
|
||||
int stream_src_failed(stream_src_t *s)
|
||||
{
|
||||
if (!s) return -1;
|
||||
pthread_mutex_lock(&s->mu);
|
||||
int f = s->failed;
|
||||
pthread_mutex_unlock(&s->mu);
|
||||
return f;
|
||||
}
|
||||
Reference in new issue
Block a user