利用线程池给客户端传文件

主函数

#include <func.h>
#include "threadPool.h"
#include "server.h"

#define EVENTSNUM 100

int pipefd[2];

void sigHandle(int signo){
    printf("%d signal\n",signo);
    int exitNum = 1;
    write(pipefd[1],&exitNum,sizeof(int));
}

int main(int argc,char* argv[]){
    
    if(argc != 4){
        error(1,errno,"need three arguments");
    }
    char* ip = argv[1];
    int port = atoi(argv[2]);
    int threadNum = atoi(argv[3]);
    pipe(pipefd);

    pid_t pid = fork();
    if(pid > 0){
        signal(SIGUSR1,sigHandle);
        wait(NULL);
        exit(0);
    }

    //创建线程池
    threadPool *thread = calloc(1,sizeof(threadPool));
    ThreadPoolInit(thread,threadNum);
    ThreadPoolStart(thread);


    //设置套接字，绑定ip和端口号，并监听
    int listenFd = serverCreate(ip,port);
    if(listenFd == -1){
        perror("serverCreate");
    }
    printf("listening\n");
    //创建epoll
    int epfd = epoll_create1(0);
    if(epfd == -1){
        perror("epoll_create");
    }
    
    int ret = listenAdd(epfd,listenFd);
    if(ret == -1){
        perror("listenAdd");
    }
    listenAdd(epfd,pipefd[0]);
    struct epoll_event events[EVENTSNUM];
    
    while(1){
        int nReady = epoll_wait(epfd,events,EVENTSNUM,-1);
        for(int i = 0;i < nReady;i++){
            int fd = events[i].data.fd;
            if(fd == listenFd){
                printf("connection\n");
                int peerFd = accept(listenFd,NULL,NULL);
                if(peerFd == -1){
                    perror("accept");
                }
                TaskEnqueue(&thread->taskQueue,peerFd);
            }else if(fd == pipefd[0]){
                int result;
                printf("will exit\n");
                read(fd,&result,sizeof(result));
                ThreadPoolStop(thread);
                ThreadPoolDestroy(thread);
                close(epfd);
                close(listenFd);
                free(thread);
                exit(0);
            }
        }
    }
    return 0;
}

server

#include "server.h"
#include <func.h>
int serverCreate(char*ip,int port){
    int sfd = socket(AF_INET,SOCK_STREAM,0);
    if(sfd == -1){
        perror("socket");
    }

    struct sockaddr_in serverAddr;
    serverAddr.sin_family = AF_INET;
    serverAddr.sin_addr.s_addr = inet_addr(ip);
    serverAddr.sin_port = htons(port);
    int on = 1;
    setsockopt(sfd,SOL_SOCKET,SO_REUSEADDR,&on,sizeof(on));

    int ret = bind(sfd,(struct sockaddr*)&serverAddr,sizeof(serverAddr));
    if(ret == -1){
        perror("bind");
    }
    ret = listen(sfd,10);
    if(ret == -1){
        perror("listen");
    }
    return sfd;
}

int listenAdd(int epfd,int fd){
    struct epoll_event ev;
    ev.data.fd = fd;
    ev.events = EPOLLIN;
    int ret = epoll_ctl(epfd,EPOLL_CTL_ADD,fd,&ev);
    if(ret == -1){
        perror("epoll_ctl");
    }
    return 0;
}

int listenDel(int epfd,int fd){
    int ret = epoll_ctl(epfd,EPOLL_CTL_DEL,fd,NULL);
    if(ret == -1){
        perror("epoll_ctl");
    }
    return 0;
}

int sendFile(int clientFd,char* fileName){
    fileInfo file;

    memset(&file,0,sizeof(file));

    file.len = strlen(fileName);

    strcpy(file.fileContent,fileName);    
    send(clientFd,&file,4 + file.len,0);
    printf("fileName: %s\n",file.fileContent);

    int fd = open(fileName,O_RDWR);
    struct stat buf;
    int ret = fstat(fd,&buf);
    if(ret == -1){
        error(1,errno,"fstat");
    }


    file.len = buf.st_size;
    char buffer[100];
    read(fd,buffer,sizeof(buffer));
    strncpy(file.fileContent,buffer,file.len);
    send(clientFd,&file,4 + file.len,0);
    
    close(fd);
    printf("fileContent: %s\n",file.fileContent);
    return 0;
}

int sendBigFile(int clientFd,char* fileName){
    fileInfo file;
    file.len = strlen(fileName);

    strcpy(file.fileContent,fileName);    
    send(clientFd,&file,4 + file.len,0);
    printf("fileName: %s\n",file.fileContent);

    int fd = open(fileName,O_RDONLY);
    struct stat buf;
    int ret = fstat(fd,&buf);
    if(ret == -1){
        error(1,errno,"fstat");
    }


    file.len = buf.st_size;
    while(1){

        ret = read(fd,file.fileContent,1024);
        if(ret == 0){
            break;
        }
        file.len = ret;
        send(clientFd,&file,4 + file.len,0);
    } 

    printf("send file success\n");
    return 0;
}

threadPool

#include "threadPool.h"
#include "server.h"

void* ThreadFunc(void* args){
    taskQueue *q = args;
    char* fileName = "dog.jpg";
    while(1){
        int clienFd = TaskDequeue(q);
        if(clienFd == -1){
            break;
        }
        sendBigFile(clienFd,fileName);
        close(clienFd);
    }
    return NULL;
}

void ThreadPoolInit(threadPool* thread,int num){
    if(thread != NULL){
        thread->threads = calloc(num,sizeof(pthread_t));
        thread->threadNum = num;
        TaskQueueInit(&thread->taskQueue);

    }
}

void ThreadPoolStart(threadPool* thread){
    for(int i = 0;i < thread->threadNum;i++){
        printf("thread %d create\n",i);
        pthread_create(&thread->threads[i],NULL,ThreadFunc,&thread->taskQueue);
    }
}

void broadcastAll(taskQueue* q){
   q->exitFlag = 1;
   pthread_cond_broadcast(&q->taskCond);
}

void ThreadPoolStop(threadPool* thread){
    while(!TaskIsEmpty(&thread->taskQueue)){
        sleep(1);
        printf("there have tasks");
    }
    broadcastAll(&thread->taskQueue);
    for(int i = 0;i < thread->threadNum;i++){
        pthread_join(thread->threads[i],NULL);
    }

}

void ThreadPoolDestroy(threadPool* thread){
    if(thread != NULL){
        free(thread->threads);
        TaskQueueDestroy(&thread->taskQueue);        
    }
}

void TaskQueueInit(taskQueue* q){
    q = calloc(1,sizeof(taskQueue));
    q->pFront = q->pRear = NULL;
    pthread_cond_init(&q->taskCond,NULL);
    pthread_mutex_init(&q->taskMutex,NULL);
    q->taskSize = 0;
    q->exitFlag = 0;
}

void TaskEnqueue(taskQueue* task,int peerFd){
    pthread_mutex_lock(&task->taskMutex);
    if(task != NULL){
        taskNode* node = calloc(1,sizeof(taskNode));
        node->peerFd = peerFd;
        node->next = NULL;

        if(task->taskSize == 0){
            task->pRear = task->pFront = node;
        }else{
            task->pRear->next = node;
            task->pRear = node;
        }
        task->taskSize++;
        pthread_cond_signal(&task->taskCond);
        printf("task in the queue\n");
    }
    pthread_mutex_unlock(&task->taskMutex);
}

int TaskDequeue(taskQueue* task){

    if(task != NULL){
        pthread_mutex_lock(&task->taskMutex);
        while(task->taskSize == 0 && task->exitFlag == 0){
            pthread_cond_wait(&task->taskCond,&task->taskMutex);
        }
        
        int ret = -1;
        if(task->exitFlag == 0){
            taskNode* retNode = task->pFront;
            ret = retNode->peerFd;
            if(task->taskSize > 1){
                task->pFront = task->pFront->next;
            }else{
                task->pFront = task->pRear = NULL;
            }
            task->taskSize--;
            free(retNode);
        }
        pthread_mutex_unlock(&task->taskMutex);
        return ret;
    }

    return -1;
}

void TaskQueueDestroy(taskQueue* task){
    if(task != NULL){
        pthread_mutex_destroy(&task->taskMutex);
        pthread_cond_destroy(&task->taskCond);
    }
}

int TaskIsEmpty(taskQueue* task){
    return task->taskSize == 0;
}