first commit

CS4210/Project 2/common/thread_pool/thread_pool.c

@@ -0,0 +1,155 @@
+#include <stdlib.h>
+#include <stdio.h>
+
+#include "thread_pool.h"
+#include <pthread.h>
+#include "../defs.h"
+
+
+/////////////////////////////////
+// Used by all data structures //
+/////////////////////////////////
+
+static unsigned int length = 0;
+
+static pthread_mutex_t mutex;
+static pthread_cond_t isReady;
+
+///////////////////////
+// Linked List Queue //
+///////////////////////
+
+typedef struct QUEUE {
+    int socket;
+    struct QUEUE* next;
+} qMember;
+
+qMember* qHead = NULL;
+qMember* qTail = NULL;
+
+pthread_mutex_t* get_mutex()
+{
+  return &mutex;
+}
+
+pthread_cond_t* get_is_ready()
+{
+  return &isReady;
+}
+
+void qAdd(int socket) {
+    qMember* newNode = malloc(sizeof(qMember));
+
+    VPRINTF(("Adding socket: %d\n", socket));
+
+    if (!newNode) { printf("Malloc failed, qAdd\n"); exit(1); }
+
+    newNode->next = NULL;
+    newNode->socket = socket;
+
+    pthread_mutex_lock(get_mutex());
+
+    if (!qHead) {
+        qTail = qHead = newNode;
+        length++;
+    } else {
+        qTail->next = newNode;
+        qTail = qTail->next;
+        length++;
+    }
+
+    pthread_mutex_unlock(get_mutex());
+    pthread_cond_signal(get_is_ready());
+    
+}
+
+int qRemove() {
+    int toRet;                                  //Socket to return
+    qMember* temp;
+
+    pthread_mutex_lock(get_mutex());                 //Lock mutex
+
+    while (!qHead) {
+        pthread_cond_wait( get_is_ready(), get_mutex() );  //No requests waiting
+    }
+
+    toRet = qHead->socket;                      //retrieve socket
+    VPRINTF(("Removing socket: %d\n", toRet));
+
+    //Fix the list
+    temp = qHead;
+    qHead = qHead->next;
+    length--;
+
+    //Fine grain lock, release lock before freeing old node
+    pthread_mutex_unlock(get_mutex());
+
+    free(temp);    
+        
+    return toRet;    
+}
+
+/*
+ * Wrappers
+ */
+void addSocket(int socket) {
+    #ifdef LLQ
+        qAdd(socket);
+    #else
+        qAdd(socket);
+    #endif
+}
+
+int removeSocket() {
+    #ifdef LLQ
+        return qRemove();
+    #else
+        return qRemove();
+    #endif
+}
+
+unsigned int size() {
+    return length;
+}
+
+void init_synch() {
+    pthread_mutex_init(get_mutex(), NULL);
+    pthread_cond_init(get_is_ready(),NULL);
+}
+
+void init_thread_pool(unsigned short *port_num, int maxThreads,
+						void *(*producer)(void*),
+						void *(*consumer)(void*),
+                        void *args) 
+{
+    int i;
+
+    if (!port_num) { printf("NULL Port Number\n"); exit(1); }
+
+    init_synch();	
+
+    printf("** Initializing Thread Pool\n");
+    VPRINTF(("*** Making %d threads on port %d\n",maxThreads,port_num));
+    fflush(stdout);
+   
+    //initialize producer
+    producerThread = malloc(sizeof(pthread_t));
+    if (!producerThread) { exit(1); }
+
+    printf("*** Listening on Port: %d\n", *port_num);
+
+    pthread_create(producerThread, NULL, producer, port_num);
+
+    //Initialize consumers
+    consumerList = malloc(sizeof(pthread_t*)*maxThreads);
+
+    for (i=0; i < maxThreads; i++) {
+        consumerList[i] = malloc(sizeof(pthread_t));
+        
+        if (pthread_create(consumerList[i], NULL, consumer, args)) {
+            printf("Error creating thread %d\n",i);
+        }
+
+    }
+
+}