1008 lines
35 KiB
C
1008 lines
35 KiB
C
/**
|
||
* 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;
|
||
}
|