MeowOS-For-R3P2/patch_v2/ncm_app/app/play/stream_src.c

/* 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 联动） */
#define STREAM_SRC_REFAULT_MS 50  /* pread 慢于此阈且数据已在盘上 → 计一次 refault（§2.3） */

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;
    int             refaults;      /* 数据在盘上但 pread 慢（>50ms）= DONTNEED 丢页重读 TF（§2.3） */
};

/* SDK meow_cachegov 提供的 refault 上报（strong 时覆盖弱符号；未链 SDK 时为空指针，调用需判空）。 */
#if defined(__GNUC__)
extern void meow_cachegov_note_refault(void) __attribute__((weak));
#else
extern void meow_cachegov_note_refault(void);
#endif

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_stream_at(stream_src_t **out, const char *path, uint64_t written)
{
    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 = written;   /* 续传（§5.2）：前缀已落盘，读端 read_pos=0 可即读 */
    return rc;
}

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;
            int had_data = (pos < s->written);  /* 数据已在盘上（放锁前采样；勿在放锁后读 written） */
            pthread_mutex_unlock(&s->mu);       /* pread 期间放锁，不阻塞写端 broadcast */
            int64_t t0 = mono_ms();
            ssize_t r = pread(s->fd, (char *)buf + done, n, (off_t)pos);
            int64_t pread_ms = mono_ms() - t0;
            pthread_mutex_lock(&s->mu);
            if (r > 0 && had_data && pread_ms > STREAM_SRC_REFAULT_MS) {
                /* 数据在盘上但 pread 慢 = DONTNEED 丢页后重读 TF（underrun 确认是 refault，
                 * 非网络等待）。上报 SDK meow_cachegov（弱符号，未链时为空指针需判空）。 */
                s->refaults++;
                if (meow_cachegov_note_refault) meow_cachegov_note_refault();
            }
            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;
}

int stream_src_refaults(stream_src_t *s)
{
    if (!s) return 0;
    pthread_mutex_lock(&s->mu);
    int r = s->refaults;
    pthread_mutex_unlock(&s->mu);
    return r;
}

uint64_t stream_src_read_pos(stream_src_t *s)
{
    if (!s) return 0;
    pthread_mutex_lock(&s->mu);
    uint64_t p = s->read_pos;
    pthread_mutex_unlock(&s->mu);
    return p;
}

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;
}