Add rowlock

uoo723 · Jun 22, 2017 · e8e4860 · e8e4860
1 parent f26a3cb
commit e8e4860
Showing 1 changed file with 73 additions and 3 deletions.
diff --git a/src/server.c b/src/server.c
@@ -2,6 +2,7 @@
 #include <signal.h>
 #include <stdlib.h>
 #include <string.h>
+#include <pthread.h>
 
 #include <sys/types.h>
 #include <sys/ipc.h>
@@ -13,10 +14,25 @@
 #include "server.h"
 #include "hashtable.h"
 #include "node.h"
+#include "thpool.h"
+
+typedef threadpool threadpool_t;
+
+typedef struct {
+    unsigned int key;
+    char value[VALUESIZE];
+} args_t;
 
 static key_t key_id;
 static msgbuf_t msg;
 static hashtable_t *hashtable;
+static threadpool_t thpool;
+
+static pthread_rwlock_t rwlock;
+
+static void rcv_put_internal(void *);
+static void rcv_get_internal(void *);
+static void rcv_remove_internal(void *);
 
 static void int_handler(int signo) {
     msgctl(key_id, IPC_RMID, 0);
@@ -35,6 +51,18 @@ void server() {
 		exit(0);
 	}
 
+    if ((thpool = thpool_init(4)) == NULL) {
+        perror("thpool is null ");
+        exit(0);
+    }
+
+    if (pthread_rwlock_init(&rwlock, NULL) < 0) {
+        perror("failed initializing rwlock ");
+        exit(0);
+    }
+
+    args_t *args = NULL;
+
 	while (1) {
 		if (msgrcv(key_id, &msg, MSGSIZE, TYPE_SERVER, 0) < 0) {
 			perror("msgrcv error ");
@@ -43,18 +71,30 @@ void server() {
 
 		switch (msg.type) {
 			case TYPE_REQ_PUT:
-			rcv_put(msg.key, msg.value);
+            args = malloc(sizeof(args_t));
+            args->key = msg.key;
+            strcpy(args->value, msg.value);
+
+            thpool_add_work(thpool, (void *) rcv_put_internal, (void *) args);
 			break;
 
 			case TYPE_REQ_GET:
-			rcv_get(msg.key);
+            args = malloc(sizeof(args_t));
+            args->key = msg.key;
+
+            thpool_add_work(thpool, (void *) rcv_get_internal, (void *) args);
 			break;
 
 			case TYPE_REQ_REMOVE:
-			rcv_remove(msg.key);
+            args = malloc(sizeof(args_t));
+            args->key = msg.key;
+
+            thpool_add_work(thpool, (void *) rcv_remove_internal, (void *) args);
 			break;
 
 			case TYPE_QUIT:
+            thpool_destroy(thpool);
+            pthread_rwlock_destroy(&rwlock);
 			int_handler(2);
 			exit(0);
 			break;
@@ -65,6 +105,36 @@ void server() {
 	}
 }
 
+static void rcv_put_internal(void *arguments) {
+    args_t *args = (args_t *) arguments;
+
+    pthread_rwlock_wrlock(&rwlock);
+    rcv_put(args->key, args->value);
+    pthread_rwlock_unlock(&rwlock);
+
+    free(args);
+}
+
+static void rcv_get_internal(void *arguments) {
+    args_t *args = (args_t *) arguments;
+
+    pthread_rwlock_rdlock(&rwlock);
+    rcv_get(args->key);
+    pthread_rwlock_unlock(&rwlock);
+
+    free(args);
+}
+
+static void rcv_remove_internal(void *arguments) {
+    args_t *args = (args_t *) arguments;
+
+    pthread_rwlock_wrlock(&rwlock);
+    rcv_remove(args->key);
+    pthread_rwlock_unlock(&rwlock);
+
+    free(args);
+}
+
 void rcv_put(unsigned int key, char *value) {
 	ht_set(hashtable, key, value);
 }