add threadpool server

.vscode/settings.json

@@ -3,6 +3,7 @@
         "_GNU_SOURCE"
     ],
     "files.associations": {
-        "socket_wrapper.h": "c"
+        "socket_wrapper.h": "c",
+        "stdalign.h": "c"
     }
 }

Makefile

@@ -13,7 +13,7 @@ socket_wrapper.o: socket_wrapper.c socket_wrapper.h
 client: socket_wrapper.o client.c
 	$(CC) -o client client.c socket_wrapper.o $(CFLAGS)
 server:  socket_wrapper.o server.c
-	$(CC) -o server server.c socket_wrapper.o $(CFLAGS)
+	$(CC) -o server server.c socket_wrapper.o $(CFLAGS) -pthread
 p-server: socket_wrapper.o p-server.c
 	$(CC) -o p-server p-server.c socket_wrapper.o $(CFLAGS)
 p-client: socket_wrapper.o p-client.c

README.md

@@ -15,9 +15,16 @@ Server OPTION and arguments:
 - `-h`        :print help message.
 
 Available macro:
-- DEFAULT_SERVER_PORT : 9091
-- DEFAULT_MAX_PATH_SIZE : 256(must be less than 1000)
-- TIMEOUT : 5(second unit)
+For server
+- DEFAULT_MAX_LISTEN_SOCKET: 16
+- DEFAULT_SERVER_PORT: 9091
+- DEFAULT_MAX_PATH_SIZE: 256(must be less than 1000)
 - USE_SENDFILE
-    - DEFAULT_SEND_FILE_CHUNK_SIZE : 0x100000(1MB)
+    - DEFAULT_SEND_FILE_CHUNK_SIZE: 0x100000(1MB)
+- DEFAULT_WORK_QUEUE_SIZE: 10
+- DEFAULT_MAX_THREAD_NUMBER: 10
+For client
 - SLOW_CLIENT(second unit)
+- DEFAULT_PROGRESS_BAR_WIDTH: 30
+For both
+- TIMEOUT: 5(second unit)

client.c

@@ -9,7 +9,7 @@
 #include <arpa/inet.h>
 #include <errno.h>
 #include <sys/stat.h>
-
+#include <stdbool.h>
 #include <assert.h>
 #include <fcntl.h>
 #include "socket_wrapper.h"
@@ -28,6 +28,7 @@ static const int TIMEOUT = 5;
 static const int TIMEOUT = DEFAULT_TIMEOUT;
 #endif
 
+static const int PROGRESS_BAR_WIDTH = 30;
 /*========
  *Operation
  *========*/
@@ -40,13 +41,43 @@ int sendReadOp(int sock,const char * filename){
         perror("readop send fail");
         return -1;
     }
+    #ifdef SLOW_CLIENT
+    sleep(SLOW_CLIENT);
+    #endif
     if(send(sock,filename,op.file_url_size,0)<0){
         perror("readop filename send fail");
         return -1;
     }
     return 0;
 }
-
+/** 
+ * @arg cur_progress : it is percentage.
+*/
+bool isProgressBarNeedUpdate(size_t offset,size_t total,double cur_progress){
+    return ((double)offset / (double)total) > (cur_progress / 100.0);
+}
+void DisplayProgressBar(size_t offset,size_t total,double cur_progress){
+    char buf[PROGRESS_BAR_WIDTH];
+    size_t i;
+    size_t cur_pos = (cur_progress / 100.0 * PROGRESS_BAR_WIDTH); //must be less than SIZE_MAX. other value is undefined behavior.
+    for (i = 0; i < PROGRESS_BAR_WIDTH; i++)
+    {
+        if (i < cur_pos)
+            buf[i] = '=';
+        else if(i == cur_pos)
+            buf[i] = '>';
+        else buf[i] = ' ';
+    }
+    printf("\r[%s]: %.2f%% bytes: %ld/%ld bytes",buf,cur_progress,total,offset);
+}
+void DisplayProgressBar100Percent(size_t total){
+    size_t i;
+    char buf[PROGRESS_BAR_WIDTH];
+    for (i = 0; i < PROGRESS_BAR_WIDTH; i++){
+        buf[i] = '=';
+    }
+    printf("\r[%s]: 100%% bytes: %ld/%ld bytes\n",buf,total,total);
+}
 int recvFile(int sock, const char * filename,size_t file_size){
     int fd;
     size_t count = 0;
@@ -80,14 +111,14 @@ int recvFile(int sock, const char * filename,size_t file_size){
             return_value = -1;
             goto END;
         }
-        if( ((double)count / (double)file_size) * 100.0 > ((double)cur_progress) ){
-            printf("\rprogress : %d%% current bytes: %ld bytes",cur_progress,count);
+        if( isProgressBarNeedUpdate(count,file_size,cur_progress) ){
+            DisplayProgressBar(count,file_size,cur_progress);
             cur_progress = (int)((((double)count / (double)file_size)) * 100.0 + 1.0);
             fflush(stdout);
         }
         count += readed;
     }
-    printf("\rprogress : 100.00%% current bytes: %ld bytes\n",count);
+    DisplayProgressBar100Percent(file_size);
 END:
     free(buf);
     return return_value;
@@ -130,10 +161,10 @@ int main(int argc, const char *argv[]){
     struct sockaddr_in addr;
     const char * filename;
     const char * server_name;
-    in_port_t server_port;
+    in_port_t server_port = 0;
     int sock, err;
     if (argc != 4){
-        fprintf(stderr,"invaild arguments number.");
+        fprintf(stderr,"USAUE: %s SERVERNAME PORT FILENAME\n",argv[0]);
         return 1;
     }
     server_name = argv[1];
@@ -149,7 +180,7 @@ int main(int argc, const char *argv[]){
         perror("sock create fail");
         return 1;
     }
-    err = getsockaddrbyname(AF_INET,SOCK_STREAM,0,server_name,((struct sockaddr *)&addr));
+    err = getsockaddrbyname(AF_INET,SOCK_STREAM,0,server_name,(struct sockaddr *)&addr);
     if (err != 0){
         int check;
         fprintf(stderr,"netdb fail: %s\n",gai_strerror(err));
@@ -166,7 +197,6 @@ int main(int argc, const char *argv[]){
     addr.sin_port = htons(server_port);
     if(connect(sock,(struct sockaddr *)&addr,sizeof(addr)) < 0){
         perror("connect failed");
-        close(sock);
         return 1;
     }
     if(sendReadOp(sock,filename) == 0){

p-client.c

@@ -9,9 +9,10 @@
 #include <arpa/inet.h>
 #include <errno.h>
 #include <sys/stat.h>
-
+#include <stdbool.h>
 #include <assert.h>
 #include <fcntl.h>
+
 #include "socket_wrapper.h"
 
 #ifdef USE_SENDFILE
@@ -27,6 +28,11 @@ static const int TIMEOUT = 5;
 #else
 static const int TIMEOUT = DEFAULT_TIMEOUT;
 #endif
+#ifndef DEFAULT_PROGRESS_BAR_WIDTH
+static const int PROGRESS_BAR_WIDTH = 30;
+#else
+static const int PROGRESS_BAR_WIDTH = DEFAULT_PROGRESS_BAR_WIDTH;
+#endif
 /*========
  *Operation
  *========*/
@@ -48,12 +54,39 @@ int sendReadOp(int sock,const char * filename){
     }
     return 0;
 }
-
+/** 
+ * @arg cur_progress : it is percentage.
+*/
+bool isProgressBarNeedUpdate(size_t offset,size_t total,double cur_progress){
+    return ((double)offset / (double)total) > (cur_progress / 100.0);
+}
+void DisplayProgressBar(size_t offset,size_t total,double cur_progress){
+    char buf[PROGRESS_BAR_WIDTH];
+    size_t i;
+    size_t cur_pos = (cur_progress / 100.0 * PROGRESS_BAR_WIDTH); //must be less than SIZE_MAX. other value is undefined behavior.
+    for (i = 0; i < PROGRESS_BAR_WIDTH; i++)
+    {
+        if (i < cur_pos)
+            buf[i] = '=';
+        else if(i == cur_pos)
+            buf[i] = '>';
+        else buf[i] = '.';
+    }
+    printf("\r[%s]: %.2f%% bytes: %ld/%ld bytes",buf,cur_progress,total,offset);
+}
+void DisplayProgressBar100Percent(size_t total){
+    size_t i;
+    char buf[PROGRESS_BAR_WIDTH];
+    for (i = 0; i < PROGRESS_BAR_WIDTH; i++){
+        buf[i] = '=';
+    }
+    printf("\r[%s]: 100%% bytes: %ld/%ld bytes\n",buf,total,total);
+}
 int recvFile(int sock, const char * filename,size_t file_size){
     int fd;
     size_t count = 0;
     int i;
-    int cur_progress = 1;
+    double cur_progress = 1;
     int return_value = 0;
     int buf_sz = getBufferSizeFrom(sock);
     uint8_t * buf = malloc(buf_sz*sizeof(*buf));
@@ -82,14 +115,14 @@ int recvFile(int sock, const char * filename,size_t file_size){
             return_value = -1;
             goto END;
         }
-        if( ((double)count / (double)file_size) * 100.0 > ((double)cur_progress) ){
-            printf("\rprogress : %d%% current bytes: %ld bytes",cur_progress,count);
+        if( isProgressBarNeedUpdate(count,file_size,cur_progress) ){
+            DisplayProgressBar(count,file_size,cur_progress);
             cur_progress = (int)((((double)count / (double)file_size)) * 100.0 + 1.0);
             fflush(stdout);
         }
         count += readed;
     }
-    printf("\rprogress : 100.00%% current bytes: %ld bytes\n",count);
+    DisplayProgressBar100Percent(file_size);
 END:
     free(buf);
     return return_value;
@@ -135,7 +168,7 @@ int main(int argc, const char *argv[]){
     in_port_t server_port = 0;
     int sock, err;
     if (argc != 4){
-        fprintf(stderr,"invaild arguments number.");
+        fprintf(stderr,"USAUE: %s SERVERNAME PORT FILENAME\n",argv[0]);
         return 1;
     }
     server_name = argv[1];

p-server.c

@@ -272,8 +272,8 @@ int main(int argc, const char *argv[]){
             perror("accept error");
             return 1;
         }
-        msg = inet_ntop(AF_INET,&addr.sin_addr,ip_buf,sizeof(ip_buf));
-        fprintf(stderr,"Connected on : %s:%d\n",msg == NULL ? "(null)" : msg , ntohs(addr.sin_port));
+        msg = inet_ntop(AF_INET,&client_addr.sin_addr,ip_buf,sizeof(ip_buf));
+        fprintf(stderr,"Connected on : %s:%d\n",msg == NULL ? "(null)" : msg , ntohs(client_addr.sin_port));
         pid = fork();
         if(pid == 0){
             if((fd = read_request(csock,buf,bufsize)) > 0){

pstest.sh

@@ -1,3 +1,3 @@
 #! /bin/bash
 cd server_test
-./p-server
+./server

server.c

@@ -12,8 +12,16 @@
 #include <signal.h>
 #include <assert.h>
 #include <fcntl.h>
-#include "socket_wrapper.h"
+#include <pthread.h>
 
+#include "socket_wrapper.h"
+#include "simple_circular_buffer.h"
+
+#ifndef DEFAULT_MAX_LISTEN_SOCKET
+static const int MAX_LISTEN_SOCKET = 16;
+#else
+static const int MAX_LISTEN_SOCKET = DEFAULT_MAX_LISTEN_SOCKET;
+#endif
 #ifdef USE_SENDFILE
 #include <sys/sendfile.h>
 #ifndef DEFAULT_SEND_FILE_CHUNK_SIZE
@@ -39,6 +47,19 @@ static const int TIMEOUT = 5;
 static const int TIMEOUT = DEFAULT_TIMEOUT;
 #endif
 
+enum{
+#ifndef DEFAULT_WORK_QUEUE_SIZE
+ WORK_QUEUE_SIZE = 10,
+ #else
+ WORK_QUEUE_SIZE = DEFAULT_WORK_QUEUE_SIZE,
+#endif
+#ifndef DEFAULT_MAX_THREAD_NUMBER
+ MAX_THREAD_NUMBER = 10
+ #else
+ MAX_THREAD_NUMBER = DEFAULT_MAX_THREAD_NUMBER
+#endif
+};
+
 /*========
  *Operation
  *========*/
@@ -204,18 +225,83 @@ int parse_args(int argc,const char * argv[] , in_port_t * port){
     }
     return 0;
 }
+//============
+//Simple Thread Pool
+//============
+typedef struct SharedState{
+    //empty if less than 0
+    queue_struct(int,WORK_QUEUE_SIZE) socks;
+    pthread_mutex_t sock_mutex;
+    pthread_cond_t ready;
+    //int progress[MAX_THREAD_NUMBER];
+} shared_state_t;
+
+void init_shared_state(shared_state_t * state) {
+    queue_init(&state->socks);
+    pthread_mutex_init(&state->sock_mutex,NULL);
+    pthread_cond_init(&state->ready,NULL);
+}
+//
+typedef struct WorkerArgument
+{
+    int id;
+    int bufsize;
+    uint8_t * buf;
+
+    //pthread_mutex_t * cond_mutex;
+    //pthread_cond_t cond;
+
+} worker_argument_t;
+
+__attribute__((malloc)) worker_argument_t * create_worker_argument(int id, int bufsize){
+    worker_argument_t * ret = (worker_argument_t *)malloc(sizeof(worker_argument_t));
+    if (ret == NULL) return ret;
+    ret->id = id;
+    ret->bufsize = bufsize;
+    ret->buf = (uint8_t *)malloc(sizeof(*ret->buf)*bufsize);
+    if(ret->buf == NULL){
+        free(ret);
+        ret = NULL;
+    }
+    return ret;
+}
+
+static shared_state_t globalState;
+
+void * worker_proc(void * data){
+    worker_argument_t * args = (worker_argument_t *)data;
+    int fd, csock;
+    for(;;){
+        pthread_mutex_lock(&globalState.sock_mutex);
+        while (queue_isempty(&globalState.socks)){
+            pthread_cond_wait(&globalState.ready,&globalState.sock_mutex);
+        }
+        csock = dequeue(&globalState.socks);
+        pthread_mutex_unlock(&globalState.sock_mutex);
+        if((fd = read_request(csock,args->buf,args->bufsize)) > 0){
+            send_response(csock,fd,args->buf,args->bufsize);
+            close(fd);
+        }
+
+        if(close(csock) < 0)
+            perror("csock close error");
+    }
+}
+
+
+static pthread_t worker_threads[MAX_THREAD_NUMBER];
 
 static int sock;
 void safe_exit(){
     close(sock);
 }
 int main(int argc, const char *argv[]){
-    uint8_t * buf;
     struct sockaddr_in addr;
     struct sockaddr_in client_addr;
     socklen_t client_addr_len = sizeof(client_addr);
     int csock;
     int bufsize;
+    int i;
     in_port_t binding_port_number = SERVER_PORT;
     if (argc > 1){
         int d = parse_args(argc,argv,&binding_port_number);
@@ -234,12 +320,16 @@ int main(int argc, const char *argv[]){
             perror("setsockopt");
         }
     }
+    init_shared_state(&globalState);
     bufsize = getBufferSizeFrom(sock);
-    buf = malloc(bufsize * sizeof(*buf));
-    if (buf == NULL){
-        fprintf(stderr,"lack of memory");
+    for (i = 0; i < MAX_THREAD_NUMBER; i++) {
+        worker_argument_t * args = create_worker_argument(i,bufsize);
+        if (args == NULL) {
+            fprintf(stderr,"malloc: lack of memory");
             return 1;
         }
+        pthread_create(&worker_threads[i],NULL,worker_proc,args);
+    }
     
     addr.sin_addr.s_addr = htonl(INADDR_ANY); /*0.0.0.0 모든 네트워크 인터페이스에 묶임.*/
     addr.sin_family = AF_INET;
@@ -255,27 +345,33 @@ int main(int argc, const char *argv[]){
         fprintf(stderr,"server bind on %s:%d\n",msg ,binding_port_number);
     }
 
-    if(listen(sock,1) < 0){
+    if(listen(sock,MAX_LISTEN_SOCKET) < 0){
         perror("listen failed");
         return 1;
     }
 
     while ((csock = accept(sock, (struct sockaddr *)&client_addr,&client_addr_len)) >= 0)
     {
-        int fd;
         char ip_buf[INET_ADDRSTRLEN];
-        const char * msg = inet_ntop(AF_INET,&addr.sin_addr,ip_buf,sizeof(ip_buf));
+        const char * msg = inet_ntop(AF_INET,&client_addr.sin_addr,ip_buf,sizeof(ip_buf));
         fprintf(stderr,"Connected on : %s:%d\n",msg == NULL ? "(null)" : msg , ntohs(addr.sin_port));
-        if((fd = read_request(csock,buf,bufsize)) > 0){
-            send_response(csock,fd,buf,bufsize);
-            close(fd);
+        for(;;){
+            pthread_mutex_lock(&globalState.sock_mutex);
+            if (queue_isfull(&globalState.socks)){
+                pthread_mutex_unlock(&globalState.sock_mutex);
+#ifdef _GNU_SOURCE
+                pthread_yield();
+#else
+                usleep(400);
+#endif    
+                continue;
+            }
+            else enqueue(&globalState.socks,csock);
+            break;
+        }
+        pthread_mutex_unlock(&globalState.sock_mutex);
+        pthread_cond_signal(&globalState.ready);
     }
     
-        if(close(csock) < 0)
-            perror("csock close error");
-
-    }
-    free(buf);
-    perror("accept error");
     return 1;
 }

simple_circular_buffer.h

@@ -0,0 +1,30 @@
+#ifndef _SIMPLE_CIRCULAR_BUFFER_
+#define _SIMPLE_CIRCULAR_BUFFER_
+
+#include<stddef.h>
+
+#define queue_struct(queue_type,queue_size) struct{\
+queue_type data [(queue_size)+1];\
+size_t begin;\
+size_t end;\
+}
+
+#define queue_size(queue) ((sizeof((queue)->data)/sizeof((queue)->data[0])) - 1)
+#define queue_isempty(queue) ((queue)->begin == (queue)->end)
+#define queue_isfull(queue) ((queue)->begin == (((queue)->end + 1) % queue_size(queue)))
+
+#define queue_init(queue) do{\
+    (queue)->begin = 0;\
+    (queue)->end = 0;\
+}while(0)
+//unchecked
+#define enqueue(queue,element) do{ \
+    (queue)->data[(queue)->end] = (element);\
+    (queue)->end = ((queue)->end + 1) % (queue_size(queue) + 1);\
+}while(0)
+//unchecked
+#define dequeue(queue) \
+(((queue)->begin = ((queue)->begin + 1) % (queue_size(queue) + 1)) ,\
+(queue)->data[((queue)->begin + queue_size(queue)) % (queue_size(queue) + 1)])
+
+#endif 

socket_wrapper.c

@@ -26,21 +26,32 @@ ssize_t timeout_recv(int fd,void * buf,size_t n,int timeout)
 {
     ssize_t ret = 0;
     int poll_ret;
+    int try = 0;
     struct pollfd fd_single;
+    timeout = timeout * 1000;
+    if (n == 0) return 0;
+    for(;;){
         fd_single.fd = fd;
         fd_single.events = POLL_IN;
-    poll_ret = (poll(&fd_single,1,timeout * 1000));
-    if (poll_ret < 0) return -1;
+        poll_ret = (poll(&fd_single,1,timeout));
+        if (poll_ret < 0){  fprintf(stderr,"timeout %d\n",timeout); return -1;}
         else if(poll_ret == 0) return -2;
         if (fd_single.revents & POLLHUP) //We'll treat hangups state like timeouts state.
             return -2;
         if ((fd_single.revents & POLLERR) || (fd_single.revents & POLLNVAL))
             return -1;
-    if (fd_single.revents & POLL_IN)
+        if (fd_single.revents & POLL_IN){
             ret = recv(fd,buf,n,0);
-        assert(ret != 0);
-        return ret;
-    
+            if(ret != 0) return ret;
+            //try 3 times
+            if (try < 3){
+                try++;
+                timeout /= 2;
+                continue;
+            }
+            return -2;
+        }
+    }
     assert(0 && "unreachable");
 }