/**
 * webserver_pipeline.c — 实验3 任务3: Web服务器业务分割模型
 *
 * 架构:
 *   main 接受连接 → read_msg 线程池 (解析HTTP请求)
 *     → filename_queue → read_file 线程池 (读取文件)
 *       → msg_queue → send_msg 线程池 (发送响应)
 *
 * 三个线程池 + 两个消息队列构成流水线处理模型。
 * 包含监控线程定时输出性能参数。
 */

#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/types.h>
#include <time.h>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
#include <sys/prctl.h>

#include "thpool.h"

/* ======================== 常量定义 ======================== */

#define VERSION     23
#define BUFSIZE     8096
#define ERROR       42
#define LOG         44
#define FORBIDDEN   403
#define NOTFOUND    404

#define READ_MSG_THREADS  4
#define READ_FILE_THREADS 4
#define SEND_MSG_THREADS  4

#define FQ_SIZE           64
#define MQ_SIZE           64

#define MONITOR_INTERVAL  5

#ifndef SIGCLD
#define SIGCLD SIGCHLD
#endif

/* ======================== 文件类型表 ======================== */

struct {
    char *ext;
    char *filetype;
} extensions[] = {
    {"gif",  "image/gif"  },
    {"jpg",  "image/jpg"  },
    {"jpeg", "image/jpeg" },
    {"png",  "image/png"  },
    {"ico",  "image/ico"  },
    {"zip",  "image/zip"  },
    {"gz",   "image/gz"   },
    {"tar",  "image/tar"  },
    {"htm",  "text/html"  },
    {"html", "text/html"  },
    {0, 0}
};

/* ======================== 消息队列数据结构 ======================== */

typedef struct {
    char  filename[512];
    char  filetype[64];
    int   socket_fd;
    int   hit;
} fq_item_t;

typedef struct {
    fq_item_t *items;
    int        cap;
    int        inpos;
    int        outpos;
    sem_t      mutex;
    sem_t      avail;
    sem_t      ready;
    int        len;
} filename_queue_t;

typedef struct {
    char  *filebuf;
    long   filesize;
    char   filetype[64];
    int    socket_fd;
    int    hit;
} mq_item_t;

typedef struct {
    mq_item_t *items;
    int        cap;
    int        inpos;
    int        outpos;
    sem_t      mutex;
    sem_t      avail;
    sem_t      ready;
    int        len;
} msg_queue_t;

/* ======================== 统计数据 ======================== */

typedef struct {
    int  read_msg_active;
    int  read_file_active;
    int  send_msg_active;

    long long read_msg_active_us;
    long long read_file_active_us;
    long long send_msg_active_us;

    long long read_msg_block_us;
    long long read_file_block_us;
    long long send_msg_block_us;

    long read_msg_tasks;
    long read_file_tasks;
    long send_msg_tasks;

    int  read_msg_active_max;
    int  read_msg_active_min;
    int  read_file_active_max;
    int  read_file_active_min;
    int  send_msg_active_max;
    int  send_msg_active_min;

    int  fq_len;
    int  mq_len;

    pthread_mutex_t lock;
} stats_t;

/* ======================== 全局变量 ======================== */

static volatile int server_running = 1;
static int          listen_fd     = -1;   /* 供信号处理器关闭 */
static filename_queue_t fq;
static msg_queue_t      mq;
static stats_t          stats;
static threadpool       read_msg_pool;

/* ======================== 函数声明 ======================== */

void  logger(int type, const char *s1, const char *s2, int socket_fd);
long  calculate_time_diff(struct timespec start, struct timespec end);
void  ctrlc_handler(int sig);

/* 带超时的信号量等待——1秒超时，返回0=成功，-1=超时/错误 */
int   sem_timedwait_1s(sem_t *sem);

/* 可中断的队列操作——server_running=0 时返回 -1 */
int   fq_insert_try(filename_queue_t *qp, fq_item_t item);
int   fq_remove_try(filename_queue_t *qp, fq_item_t *out);
int   mq_insert_try(msg_queue_t *qp, mq_item_t item);
int   mq_remove_try(msg_queue_t *qp, mq_item_t *out);

void  fq_init(filename_queue_t *qp, int cap);
void  mq_init(msg_queue_t *qp, int cap);

void  read_msg_worker(void *arg);
void *read_file_thread_func(void *arg);
void *send_msg_thread_func(void *arg);
void *monitor_thread_func(void *arg);

/* ======================== 工具函数实现 ======================== */

void logger(int type, const char *s1, const char *s2, int socket_fd)
{
    char logbuffer[BUFSIZE * 2];
    char timebuffer[64];
    struct timeval tv;
    gettimeofday(&tv, NULL);

    struct tm *tm_info = localtime(&tv.tv_sec);
    strftime(timebuffer, sizeof(timebuffer), "%Y-%m-%d %H:%M:%S", tm_info);

    char full_timebuffer[256];
    snprintf(full_timebuffer, sizeof(full_timebuffer),
             "[%s.%03ld]", timebuffer, tv.tv_usec / 1000);

    switch (type) {
    case ERROR:
        snprintf(logbuffer, sizeof(logbuffer),
                 "%s [ERROR] %s:%s Errno=%d pid=%d",
                 full_timebuffer, s1, s2, errno, getpid());
        break;
    case FORBIDDEN:
        if (write(socket_fd,
                  "HTTP/1.1 403 Forbidden\nContent-Length: 185\n"
                  "Connection: close\nContent-Type: text/html\n\n"
                  "<html><head>\n<title>403 Forbidden</title>\n</head><body>\n"
                  "<h1>Forbidden</h1>\n"
                  "The requested URL, file type or operation is not allowed "
                  "on this simple static file webserver.\n"
                  "</body></html>\n", 271) < 0) {}
        snprintf(logbuffer, sizeof(logbuffer),
                 "%s [FORBIDDEN] %s:%s", full_timebuffer, s1, s2);
        break;
    case NOTFOUND:
        if (write(socket_fd,
                  "HTTP/1.1 404 Not Found\nContent-Length: 136\n"
                  "Connection: close\nContent-Type: text/html\n\n"
                  "<html><head>\n<title>404 Not Found</title>\n</head><body>\n"
                  "<h1>Not Found</h1>\n"
                  "The requested URL was not found on this server.\n"
                  "</body></html>\n", 224) < 0) {}
        snprintf(logbuffer, sizeof(logbuffer),
                 "%s [NOT FOUND] %s:%s", full_timebuffer, s1, s2);
        break;
    case LOG:
        snprintf(logbuffer, sizeof(logbuffer),
                 "%s [INFO] %s:%s:%d", full_timebuffer, s1, s2, socket_fd);
        break;
    }

#ifdef DEBUG
    {
        int log_fd;
        if ((log_fd = open("server_pipeline.log",
                           O_CREAT | O_WRONLY | O_APPEND, 0644)) >= 0) {
            if (write(log_fd, logbuffer, strlen(logbuffer)) < 0) {}
            if (write(log_fd, "\n", 1) < 0) {}
            close(log_fd);
        }
    }
#endif
}

long calculate_time_diff(struct timespec start, struct timespec end)
{
    long diff_sec  = end.tv_sec  - start.tv_sec;
    long diff_nsec = end.tv_nsec - start.tv_nsec;
    return diff_sec * 1000 + diff_nsec / 1000000;
}

void ctrlc_handler(int sig)
{
    (void)sig;
    printf("\n[Ctrl+C] 正在关闭服务器...\n");
    server_running = 0;
    /* 关闭监听 socket 以唤醒阻塞在 accept() 的主线程 */
    if (listen_fd >= 0) {
        shutdown(listen_fd, SHUT_RDWR);
        close(listen_fd);
        listen_fd = -1;
    }
}

/* ======================== 可中断的信号量等待 ======================== */

/**
 * sem_timedwait_1s — 等待信号量最多1秒
 * 返回 0=成功获取, -1=超时或出错
 */
int sem_timedwait_1s(sem_t *sem)
{
    struct timespec ts;
    if (clock_gettime(CLOCK_REALTIME, &ts) == -1)
        return -1;
    ts.tv_sec += 1;
    return sem_timedwait(sem, &ts);
}

/* ======================== 可中断队列操作 ======================== */

void fq_init(filename_queue_t *qp, int cap)
{
    qp->items  = calloc((size_t)cap, sizeof(fq_item_t));
    qp->cap    = cap;
    qp->inpos  = 0;
    qp->outpos = 0;
    qp->len    = 0;
    sem_init(&qp->mutex, 0, 1);
    sem_init(&qp->avail, 0, (unsigned int)cap);
    sem_init(&qp->ready, 0, 0);
}

void mq_init(msg_queue_t *qp, int cap)
{
    qp->items  = calloc((size_t)cap, sizeof(mq_item_t));
    qp->cap    = cap;
    qp->inpos  = 0;
    qp->outpos = 0;
    qp->len    = 0;
    sem_init(&qp->mutex, 0, 1);
    sem_init(&qp->avail, 0, (unsigned int)cap);
    sem_init(&qp->ready, 0, 0);
}

/**
 * fq_insert_try — 尝试插入 Filename Queue，可被 Ctrl+C 中断
 * 返回 0=成功, -1=服务器关闭中
 */
int fq_insert_try(filename_queue_t *qp, fq_item_t item)
{
    while (server_running) {
        if (sem_timedwait_1s(&qp->avail) != 0)
            continue;   /* 超时，重新检查 server_running */
        /* 获取到空位 */
        sem_wait(&qp->mutex);
        qp->items[qp->inpos] = item;
        qp->inpos = (qp->inpos + 1) % qp->cap;
        qp->len++;
        sem_post(&qp->mutex);
        sem_post(&qp->ready);
        return 0;
    }
    return -1;  /* 服务器关闭 */
}

/**
 * fq_remove_try — 尝试从 Filename Queue 取出，可被 Ctrl+C 中断
 * 返回 0=成功, -1=服务器关闭中
 */
int fq_remove_try(filename_queue_t *qp, fq_item_t *out)
{
    while (server_running) {
        if (sem_timedwait_1s(&qp->ready) != 0)
            continue;
        sem_wait(&qp->mutex);
        *out = qp->items[qp->outpos];
        qp->outpos = (qp->outpos + 1) % qp->cap;
        qp->len--;
        sem_post(&qp->mutex);
        sem_post(&qp->avail);
        return 0;
    }
    return -1;
}

int mq_insert_try(msg_queue_t *qp, mq_item_t item)
{
    while (server_running) {
        if (sem_timedwait_1s(&qp->avail) != 0)
            continue;
        sem_wait(&qp->mutex);
        qp->items[qp->inpos] = item;
        qp->inpos = (qp->inpos + 1) % qp->cap;
        qp->len++;
        sem_post(&qp->mutex);
        sem_post(&qp->ready);
        return 0;
    }
    return -1;
}

int mq_remove_try(msg_queue_t *qp, mq_item_t *out)
{
    while (server_running) {
        if (sem_timedwait_1s(&qp->ready) != 0)
            continue;
        sem_wait(&qp->mutex);
        *out = qp->items[qp->outpos];
        qp->outpos = (qp->outpos + 1) % qp->cap;
        qp->len--;
        sem_post(&qp->mutex);
        sem_post(&qp->avail);
        return 0;
    }
    return -1;
}

/* ======================== 流水线阶段实现 ======================== */

typedef struct {
    int socket_fd;
    int hit;
} read_msg_arg_t;

void read_msg_worker(void *arg)
{
    read_msg_arg_t *p = (read_msg_arg_t *)arg;
    int fd  = p->socket_fd;
    int hit = p->hit;
    free(arg);

    if (!server_running) { close(fd); return; }

    struct timespec t_start, t_end;
    long i, ret, buflen, len;
    int  j;
    char *fstr;
    char buffer[BUFSIZE + 1];

    clock_gettime(CLOCK_MONOTONIC, &t_start);
    logger(LOG, "[Stage1] Read Msg start", "", fd);

    ret = read(fd, buffer, BUFSIZE);
    if (ret <= 0) {
        logger(FORBIDDEN, "[Stage1] Failed to read browser request", "", fd);
        clock_gettime(CLOCK_MONOTONIC, &t_end);
        goto done_active;
    }
    buffer[ret] = '\0';

#ifdef DEBUG
    printf("[Stage1] 第%d个客户请求: %s\n", hit, buffer);
#endif

    for (i = 0; i < ret; i++)
        if (buffer[i] == '\r' || buffer[i] == '\n')
            buffer[i] = '*';

    if (strncmp(buffer, "GET ", 4)) {
        logger(FORBIDDEN, "[Stage1] Only simple GET operation supported", buffer, fd);
        clock_gettime(CLOCK_MONOTONIC, &t_end);
        goto done_active;
    }

    for (i = 4; i < BUFSIZE; i++) {
        if (buffer[i] == ' ') { buffer[i] = 0; break; }
    }

    for (j = 0; j < (int)i - 1; j++) {
        if (buffer[j] == '.' && buffer[j + 1] == '.') {
            logger(FORBIDDEN, "[Stage1] Parent directory not supported", buffer, fd);
            clock_gettime(CLOCK_MONOTONIC, &t_end);
            goto done_active;
        }
    }

    if (!strncmp(&buffer[0], "GET /\0", 6))
        strcpy(buffer, "GET /index.html");

    buflen = strlen(buffer);
    fstr   = NULL;
    for (i = 0; extensions[i].ext != 0; i++) {
        len = strlen(extensions[i].ext);
        if (!strncmp(&buffer[buflen - len], extensions[i].ext, len)) {
            fstr = extensions[i].filetype;
            break;
        }
    }

    if (fstr == NULL) {
        logger(FORBIDDEN, "[Stage1] File extension type not supported", buffer, fd);
        clock_gettime(CLOCK_MONOTONIC, &t_end);
        goto done_active;
    }

    /* 构造 filename queue 项目并推入 */
    {
        fq_item_t item;
        strncpy(item.filename, &buffer[5], sizeof(item.filename) - 1);
        item.filename[sizeof(item.filename) - 1] = '\0';
        strncpy(item.filetype, fstr, sizeof(item.filetype) - 1);
        item.filetype[sizeof(item.filetype) - 1] = '\0';
        item.socket_fd = fd;
        item.hit       = hit;

        logger(LOG, "[Stage1] Pushing to filename_queue", item.filename, hit);

        clock_gettime(CLOCK_MONOTONIC, &t_end);
        {
            long long elapsed_us =
                (t_end.tv_sec - t_start.tv_sec) * 1000000LL +
                (t_end.tv_nsec - t_start.tv_nsec) / 1000LL;
            pthread_mutex_lock(&stats.lock);
            stats.read_msg_active_us += elapsed_us;
            stats.read_msg_tasks++;
            pthread_mutex_unlock(&stats.lock);
        }

        /* 可中断的队列插入 */
        {
            struct timespec b_start, b_end;
            clock_gettime(CLOCK_MONOTONIC, &b_start);
            int rc = fq_insert_try(&fq, item);
            clock_gettime(CLOCK_MONOTONIC, &b_end);
            long long block_us =
                (b_end.tv_sec - b_start.tv_sec) * 1000000LL +
                (b_end.tv_nsec - b_start.tv_nsec) / 1000LL;
            pthread_mutex_lock(&stats.lock);
            stats.read_msg_block_us += block_us;
            pthread_mutex_unlock(&stats.lock);

            if (rc != 0) {
                /* 服务器关闭中 */
                close(fd);
                return;
            }
        }
    }
    return;

done_active:
    clock_gettime(CLOCK_MONOTONIC, &t_end);
    {
        long long elapsed_us =
            (t_end.tv_sec - t_start.tv_sec) * 1000000LL +
            (t_end.tv_nsec - t_start.tv_nsec) / 1000LL;
        pthread_mutex_lock(&stats.lock);
        stats.read_msg_active_us += elapsed_us;
        stats.read_msg_tasks++;
        pthread_mutex_unlock(&stats.lock);
    }
    close(fd);
}

/**
 * 阶段2: Read File — 持久线程，从 filename_queue 取任务，读文件，推入 msg_queue
 */
void *read_file_thread_func(void *arg)
{
    int thread_id = *(int *)arg;
    free(arg);

    printf("[ReadFile Thread %d] 启动\n", thread_id);
    prctl(PR_SET_NAME, "pipeline-rf", 0, 0, 0);

    while (server_running) {
        struct timespec b_start, b_end;
        fq_item_t item;

        /* 可中断的队列取出 */
        clock_gettime(CLOCK_MONOTONIC, &b_start);
        int rc = fq_remove_try(&fq, &item);
        clock_gettime(CLOCK_MONOTONIC, &b_end);
        {
            long long block_us =
                (b_end.tv_sec - b_start.tv_sec) * 1000000LL +
                (b_end.tv_nsec - b_start.tv_nsec) / 1000LL;
            pthread_mutex_lock(&stats.lock);
            stats.read_file_block_us += block_us;
            stats.read_file_active++;
            pthread_mutex_unlock(&stats.lock);
        }

        if (rc != 0 || !server_running) break;

        struct timespec t_start, t_end;
        clock_gettime(CLOCK_MONOTONIC, &t_start);

        logger(LOG, "[Stage2] Read File start", item.filename, item.hit);

        int file_fd = open(item.filename, O_RDONLY);
        if (file_fd == -1) {
            logger(NOTFOUND, "[Stage2] Failed to open file", item.filename, item.socket_fd);
            close(item.socket_fd);
            clock_gettime(CLOCK_MONOTONIC, &t_end);
            goto done_file_active;
        }

        long filesize = (long)lseek(file_fd, 0, SEEK_END);
        lseek(file_fd, 0, SEEK_SET);

        char *filebuf = (char *)malloc((size_t)filesize);
        if (filebuf == NULL) {
            logger(ERROR, "[Stage2] malloc failed", item.filename, item.socket_fd);
            close(file_fd);
            close(item.socket_fd);
            clock_gettime(CLOCK_MONOTONIC, &t_end);
            goto done_file_active;
        }

        ssize_t total_read = 0;
        while (total_read < filesize) {
            ssize_t n = read(file_fd, filebuf + total_read,
                             (size_t)(filesize - total_read));
            if (n <= 0) break;
            total_read += n;
        }
        close(file_fd);

#ifdef DEBUG
        printf("[Stage2] 读取文件 %s, 大小 %ld 字节, hit=%d\n",
               item.filename, filesize, item.hit);
#endif

        {
            mq_item_t mitem;
            mitem.filebuf  = filebuf;
            mitem.filesize = filesize;
            strncpy(mitem.filetype, item.filetype, sizeof(mitem.filetype) - 1);
            mitem.filetype[sizeof(mitem.filetype) - 1] = '\0';
            mitem.socket_fd = item.socket_fd;
            mitem.hit       = item.hit;

            logger(LOG, "[Stage2] Pushing to msg_queue", item.filename, item.hit);

            clock_gettime(CLOCK_MONOTONIC, &t_end);
            {
                long long elapsed_us =
                    (t_end.tv_sec - t_start.tv_sec) * 1000000LL +
                    (t_end.tv_nsec - t_start.tv_nsec) / 1000LL;
                pthread_mutex_lock(&stats.lock);
                stats.read_file_active_us += elapsed_us;
                stats.read_file_tasks++;
                stats.read_file_active--;
                pthread_mutex_unlock(&stats.lock);
            }

            /* 可中断的队列插入 */
            {
                struct timespec b2_start, b2_end;
                clock_gettime(CLOCK_MONOTONIC, &b2_start);
                int rc2 = mq_insert_try(&mq, mitem);
                clock_gettime(CLOCK_MONOTONIC, &b2_end);
                long long b2_us =
                    (b2_end.tv_sec - b2_start.tv_sec) * 1000000LL +
                    (b2_end.tv_nsec - b2_start.tv_nsec) / 1000LL;
                pthread_mutex_lock(&stats.lock);
                stats.read_file_block_us += b2_us;
                pthread_mutex_unlock(&stats.lock);

                if (rc2 != 0) {
                    free(filebuf);
                    close(item.socket_fd);
                }
            }
        }
        continue;

    done_file_active:
        clock_gettime(CLOCK_MONOTONIC, &t_end);
        {
            long long elapsed_us =
                (t_end.tv_sec - t_start.tv_sec) * 1000000LL +
                (t_end.tv_nsec - t_start.tv_nsec) / 1000LL;
            pthread_mutex_lock(&stats.lock);
            stats.read_file_active_us += elapsed_us;
            stats.read_file_tasks++;
            stats.read_file_active--;
            pthread_mutex_unlock(&stats.lock);
        }
    }

    printf("[ReadFile Thread %d] 退出\n", thread_id);
    return NULL;
}

/**
 * 阶段3: Send Msg — 持久线程，从 msg_queue 取任务，发送响应，释放资源
 */
void *send_msg_thread_func(void *arg)
{
    int thread_id = *(int *)arg;
    free(arg);

    printf("[SendMsg Thread %d] 启动\n", thread_id);
    prctl(PR_SET_NAME, "pipeline-sm", 0, 0, 0);

    while (server_running) {
        struct timespec b_start, b_end;
        mq_item_t item;

        /* 可中断的队列取出 */
        clock_gettime(CLOCK_MONOTONIC, &b_start);
        int rc = mq_remove_try(&mq, &item);
        clock_gettime(CLOCK_MONOTONIC, &b_end);
        {
            long long block_us =
                (b_end.tv_sec - b_start.tv_sec) * 1000000LL +
                (b_end.tv_nsec - b_start.tv_nsec) / 1000LL;
            pthread_mutex_lock(&stats.lock);
            stats.send_msg_block_us += block_us;
            stats.send_msg_active++;
            pthread_mutex_unlock(&stats.lock);
        }

        if (rc != 0 || !server_running) break;

        struct timespec t_start, t_end;
        clock_gettime(CLOCK_MONOTONIC, &t_start);

        logger(LOG, "[Stage3] Send Msg start", "", item.hit);

        char header[BUFSIZE];
        int header_len = snprintf(header, sizeof(header),
            "HTTP/1.1 200 OK\n"
            "Server: nweb-pipeline/%d.0\n"
            "Content-Length: %ld\n"
            "Connection: close\n"
            "Content-Type: %s\n\n",
            VERSION, item.filesize, item.filetype);

        if (write(item.socket_fd, header, (size_t)header_len) < 0) {
            logger(ERROR, "[Stage3] Failed to write HTTP header", "", item.socket_fd);
            free(item.filebuf);
            close(item.socket_fd);
            clock_gettime(CLOCK_MONOTONIC, &t_end);
            goto done_send_active;
        }

        ssize_t total_sent = 0;
        while (total_sent < item.filesize) {
            ssize_t n = write(item.socket_fd, item.filebuf + total_sent,
                              (size_t)(item.filesize - total_sent));
            if (n < 0) {
                logger(ERROR, "[Stage3] Failed to write file content", "", item.socket_fd);
                break;
            }
            total_sent += n;
        }

#ifdef DEBUG
        printf("[Stage3] 发送响应完成, hit=%d, 大小=%ld 字节\n",
               item.hit, item.filesize);
#endif

        free(item.filebuf);
        close(item.socket_fd);

        clock_gettime(CLOCK_MONOTONIC, &t_end);
        {
            long long elapsed_us =
                (t_end.tv_sec - t_start.tv_sec) * 1000000LL +
                (t_end.tv_nsec - t_start.tv_nsec) / 1000LL;
            pthread_mutex_lock(&stats.lock);
            stats.send_msg_active_us += elapsed_us;
            stats.send_msg_tasks++;
            stats.send_msg_active--;
            pthread_mutex_unlock(&stats.lock);
        }
        continue;

    done_send_active:
        clock_gettime(CLOCK_MONOTONIC, &t_end);
        {
            long long elapsed_us =
                (t_end.tv_sec - t_start.tv_sec) * 1000000LL +
                (t_end.tv_nsec - t_start.tv_nsec) / 1000LL;
            pthread_mutex_lock(&stats.lock);
            stats.send_msg_active_us += elapsed_us;
            stats.send_msg_tasks++;
            stats.send_msg_active--;
            pthread_mutex_unlock(&stats.lock);
        }
    }

    printf("[SendMsg Thread %d] 退出\n", thread_id);
    return NULL;
}

/* ======================== 监控线程 ======================== */

void *monitor_thread_func(void *arg)
{
    (void)arg;
    printf("[Monitor] 监控线程启动, 每 %d 秒输出一次性能参数\n", MONITOR_INTERVAL);

    while (server_running) {
        /* 每秒检查一次，共 MONITOR_INTERVAL 秒 */
        for (int s = 0; s < MONITOR_INTERVAL && server_running; s++)
            sleep(1);
        if (!server_running) break;

        pthread_mutex_lock(&stats.lock);

        stats.fq_len = fq.len;
        stats.mq_len = mq.len;
        stats.read_msg_active = thpool_num_threads_working(read_msg_pool);

        if (stats.read_msg_active > stats.read_msg_active_max)
            stats.read_msg_active_max = stats.read_msg_active;
        if (stats.read_msg_active < stats.read_msg_active_min)
            stats.read_msg_active_min = stats.read_msg_active;

        if (stats.read_file_active > stats.read_file_active_max)
            stats.read_file_active_max = stats.read_file_active;
        if (stats.read_file_active < stats.read_file_active_min)
            stats.read_file_active_min = stats.read_file_active;

        if (stats.send_msg_active > stats.send_msg_active_max)
            stats.send_msg_active_max = stats.send_msg_active;
        if (stats.send_msg_active < stats.send_msg_active_min)
            stats.send_msg_active_min = stats.send_msg_active;

        long long total_active_us =
            stats.read_msg_active_us +
            stats.read_file_active_us +
            stats.send_msg_active_us;
        long long total_block_us =
            stats.read_msg_block_us +
            stats.read_file_block_us +
            stats.send_msg_block_us;

        long total_tasks =
            stats.read_msg_tasks +
            stats.read_file_tasks +
            stats.send_msg_tasks;

        long avg_rm_active = stats.read_msg_tasks > 0
            ? (long)(stats.read_msg_active_us / stats.read_msg_tasks) : 0;
        long avg_rf_active = stats.read_file_tasks > 0
            ? (long)(stats.read_file_active_us / stats.read_file_tasks) : 0;
        long avg_sm_active = stats.send_msg_tasks > 0
            ? (long)(stats.send_msg_active_us / stats.send_msg_tasks) : 0;
        long avg_rm_block  = stats.read_msg_tasks > 0
            ? (long)(stats.read_msg_block_us / stats.read_msg_tasks) : 0;
        long avg_rf_block  = stats.read_file_tasks > 0
            ? (long)(stats.read_file_block_us / stats.read_file_tasks) : 0;
        long avg_sm_block  = stats.send_msg_tasks > 0
            ? (long)(stats.send_msg_block_us / stats.send_msg_tasks) : 0;

        printf("\n");
        printf("╔══════════════════════════════════════════════════════════════╗\n");
        printf("║       业务分割模型 (Pipeline) 性能监控报告                   ║\n");
        printf("╠══════════════════════════════════════════════════════════════╣\n");
        printf("║  处理请求总数: %6ld                                        ║\n",
               stats.read_msg_tasks);
        printf("╠══════════════════════════════════════════════════════════════╣\n");
        printf("║  线程池        │ 活跃 │ 最高 │ 最低 │ 平均活跃 │ 平均阻塞   ║\n");
        printf("║                │(当前)│      │      │ 时间(us) │ 时间(us)    ║\n");
        printf("╠──────────────────────────────────────────────────────────────╣\n");
        printf("║ Read Msg       │ %4d │ %4d │ %4d │ %8ld │ %9ld    ║\n",
               stats.read_msg_active,
               stats.read_msg_active_max, stats.read_msg_active_min,
               avg_rm_active, avg_rm_block);
        printf("║ Read File      │ %4d │ %4d │ %4d │ %8ld │ %9ld    ║\n",
               stats.read_file_active,
               stats.read_file_active_max, stats.read_file_active_min,
               avg_rf_active, avg_rf_block);
        printf("║ Send Msg       │ %4d │ %4d │ %4d │ %8ld │ %9ld    ║\n",
               stats.send_msg_active,
               stats.send_msg_active_max, stats.send_msg_active_min,
               avg_sm_active, avg_sm_block);
        printf("╠══════════════════════════════════════════════════════════════╣\n");
        printf("║  消息队列      │ 当前长度 │ 容量                             ║\n");
        printf("╠──────────────────────────────────────────────────────────────╣\n");
        printf("║ Filename Queue │ %6d   │ %4d                             ║\n",
               stats.fq_len, FQ_SIZE);
        printf("║ Msg Queue      │ %6d   │ %4d                             ║\n",
               stats.mq_len, MQ_SIZE);
        printf("╠══════════════════════════════════════════════════════════════╣\n");
        printf("║  累计活跃时间: %lld us  累计阻塞时间: %lld us            ║\n",
               total_active_us, total_block_us);
        printf("║  累计总任务数: %ld                                          ║\n",
               total_tasks);
        printf("╚══════════════════════════════════════════════════════════════╝\n");
        printf("\n");

        pthread_mutex_unlock(&stats.lock);
    }

    printf("[Monitor] 监控线程退出\n");
    return NULL;
}

/* ======================== 主函数 ======================== */

int main(int argc, char **argv)
{
    int i, port, socketfd, hit;
    socklen_t length;
    static struct sockaddr_in cli_addr;
    static struct sockaddr_in serv_addr;

    if (argc < 3 || argc > 3 || !strcmp(argv[1], "-?")) {
        (void)printf(
            "hint: webserver_pipeline Port-Number Top-Directory\t\tv%d\n\n"
            "\tPipeline (业务分割) 版本 — 3线程池 + 2消息队列\n\n"
            "\tExample: ./webserver_pipeline 8088 .\n\n"
            "\tOnly Supports:", VERSION);
        for (i = 0; extensions[i].ext != 0; i++)
            (void)printf(" %s", extensions[i].ext);
        (void)printf(
            "\n\tNot Supported: .., Java, Javascript, CGI\n"
            "\tNot Supported: / /etc /bin /lib /tmp /usr /dev /sbin\n");
        exit(0);
    }

    if (signal(SIGINT, ctrlc_handler) == SIG_ERR)
        (void)printf("警告: 无法设置 SIGINT 信号处理\n");

    if (!strncmp(argv[2], "/", 2) || !strncmp(argv[2], "/etc", 5) ||
        !strncmp(argv[2], "/bin", 5) || !strncmp(argv[2], "/lib", 5) ||
        !strncmp(argv[2], "/tmp", 5) || !strncmp(argv[2], "/usr", 5) ||
        !strncmp(argv[2], "/dev", 5) || !strncmp(argv[2], "/sbin", 6)) {
        (void)printf("ERROR: Bad top directory %s\n", argv[2]);
        exit(3);
    }

    if (chdir(argv[2]) == -1) {
        perror("Failed to change directory");
        exit(4);
    }

    port = atoi(argv[1]);
    if (port < 1 || port > 60000) {
        fprintf(stderr, "Invalid port number: %d\n", port);
        exit(1);
    }

    if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        perror("Socket creation failed");
        exit(1);
    }

    int optval = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR,
               (const void *)&optval, sizeof(int));

    serv_addr.sin_family      = AF_INET;
    serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    serv_addr.sin_port        = htons((unsigned short)port);

    if (bind(listen_fd, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0) {
        perror("Bind failed");
        exit(1);
    }
    if (listen(listen_fd, 64) < 0) {
        perror("Listen failed");
        exit(1);
    }

    printf("╔══════════════════════════════════════════════════════════╗\n");
    printf("║  业务分割模型 (Pipeline) Web 服务器 v%d                  ║\n", VERSION);
    printf("║  端口: %-5d   目录: %-30s ║\n", port, argv[2]);
    printf("╠══════════════════════════════════════════════════════════╣\n");
    printf("║  Accept → [Read Msg:%d] → FQ → [Read File:%d]        ║\n",
           READ_MSG_THREADS, READ_FILE_THREADS);
    printf("║         → MQ → [Send Msg:%d] → 客户端                ║\n",
           SEND_MSG_THREADS);
    printf("╚══════════════════════════════════════════════════════════╝\n");

    memset(&stats, 0, sizeof(stats));
    stats.read_msg_active_min   = 9999;
    stats.read_file_active_min  = 9999;
    stats.send_msg_active_min   = 9999;
    pthread_mutex_init(&stats.lock, NULL);

    fq_init(&fq, FQ_SIZE);
    mq_init(&mq, MQ_SIZE);
    printf("[Init] 消息队列: FQ(%d) MQ(%d)\n", FQ_SIZE, MQ_SIZE);

    read_msg_pool = thpool_init(READ_MSG_THREADS);
    printf("[Init] Read Msg 线程池: %d 线程\n", READ_MSG_THREADS);

    pthread_t *read_file_threads = malloc(sizeof(pthread_t) * READ_FILE_THREADS);
    for (i = 0; i < READ_FILE_THREADS; i++) {
        int *tid = malloc(sizeof(int)); *tid = i;
        pthread_create(&read_file_threads[i], NULL, read_file_thread_func, tid);
    }
    printf("[Init] Read File 线程池: %d 线程\n", READ_FILE_THREADS);

    pthread_t *send_msg_threads = malloc(sizeof(pthread_t) * SEND_MSG_THREADS);
    for (i = 0; i < SEND_MSG_THREADS; i++) {
        int *tid = malloc(sizeof(int)); *tid = i;
        pthread_create(&send_msg_threads[i], NULL, send_msg_thread_func, tid);
    }
    printf("[Init] Send Msg 线程池: %d 线程\n", SEND_MSG_THREADS);

    pthread_t monitor_thread;
    pthread_create(&monitor_thread, NULL, monitor_thread_func, NULL);
    printf("[Init] 监控线程: 间隔 %d 秒\n\n", MONITOR_INTERVAL);

    /* ---- 主循环 ---- */
    for (hit = 1; server_running; hit++) {
        length = sizeof(cli_addr);
        socketfd = accept(listen_fd, (struct sockaddr *)&cli_addr, &length);
        if (socketfd < 0) {
            if (!server_running) break;
            continue;
        }

        read_msg_arg_t *new_arg = malloc(sizeof(read_msg_arg_t));
        new_arg->socket_fd = socketfd;
        new_arg->hit       = hit;
        thpool_add_work(read_msg_pool, read_msg_worker, new_arg);
    }

    /* ---- 关闭 ---- */
    printf("\n[Main] 正在关闭...\n");

    printf("[Main] 等待 Read Msg 线程池...\n");
    thpool_wait(read_msg_pool);
    thpool_destroy(read_msg_pool);

    printf("[Main] 等待 Read File / Send Msg / Monitor 线程...\n");
    for (i = 0; i < READ_FILE_THREADS; i++)
        pthread_join(read_file_threads[i], NULL);
    for (i = 0; i < SEND_MSG_THREADS; i++)
        pthread_join(send_msg_threads[i], NULL);
    pthread_join(monitor_thread, NULL);

    free(read_file_threads);
    free(send_msg_threads);
    free(fq.items);
    free(mq.items);
    pthread_mutex_destroy(&stats.lock);
    sem_destroy(&fq.mutex); sem_destroy(&fq.avail); sem_destroy(&fq.ready);
    sem_destroy(&mq.mutex); sem_destroy(&mq.avail); sem_destroy(&mq.ready);

    printf("\n╔══════════════════════════════════════════════════════════╗\n");
    printf("║              服务器已关闭 - 最终统计                     ║\n");
    printf("╠══════════════════════════════════════════════════════════╣\n");
    printf("║  Read Msg  : %6ld    Read File: %6ld                  ║\n",
           stats.read_msg_tasks, stats.read_file_tasks);
    printf("║  Send Msg  : %6ld                                       ║\n",
           stats.send_msg_tasks);
    printf("╚══════════════════════════════════════════════════════════╝\n");

    logger(LOG, "Pipeline webserver exit normally", "", 0);
    return 0;
}