axis-c-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From billblo...@apache.org
Subject [axis-axis2-c-core] 02/08: Adding new thread pool code
Date Sat, 04 Apr 2020 15:08:08 GMT
This is an automated email from the ASF dual-hosted git repository.

billblough pushed a commit to branch new_thread_pool-25march2009
in repository https://gitbox.apache.org/repos/asf/axis-axis2-c-core.git

commit d0574df33a57ff0946f82c2e67466f5b27812696
Author: Damitha N.M. Kumarage <damitha@apache.org>
AuthorDate: Wed Mar 25 01:10:34 2009 +0000

    Adding new thread pool code
---
 util/include/axutil_thread_pool.h                |  11 +-
 util/include/platforms/unix/axutil_thread_unix.h |   6 +
 util/src/Makefile.am                             |   1 +
 util/src/platforms/unix/thread_unix.c            |  61 +++++++-
 util/src/thread_pool.c                           | 187 +++++++++++++++++++++--
 5 files changed, 251 insertions(+), 15 deletions(-)

diff --git a/util/include/axutil_thread_pool.h b/util/include/axutil_thread_pool.h
index d8bd8a0..ce83341 100644
--- a/util/include/axutil_thread_pool.h
+++ b/util/include/axutil_thread_pool.h
@@ -28,6 +28,9 @@
 #include <axutil_allocator.h>
 #include <axutil_thread.h>
 
+/* maximum number of threads allowed in the pool */
+#define MAXT_IN_POOL 200
+
 #ifdef __cplusplus
 extern "C"
 {
@@ -48,8 +51,8 @@ extern "C"
      * @param data arguments to be passed to the function
      * @return pointer to a thread in ready state.
      */
-    AXIS2_EXTERN axutil_thread_t *AXIS2_CALL
-    axutil_thread_pool_get_thread(
+    AXIS2_EXTERN axis2_status_t AXIS2_CALL
+    axutil_thread_pool_dispatch(
         axutil_thread_pool_t * pool,
         axutil_thread_start_t func,
         void *data);
@@ -95,11 +98,13 @@ extern "C"
     /**
     * Initializes (creates) an thread_pool.
     * @param allocator user defined allocator for the memory allocation.
+    * @param size Size of the thread pool.
     * @return initialized thread_pool. NULL on error.
     */
     AXIS2_EXTERN axutil_thread_pool_t *AXIS2_CALL
     axutil_thread_pool_init(
-        axutil_allocator_t * allocator);
+        axutil_allocator_t * allocator,
+        int size);
 
     /**
      * This function can be used to initialize the environment in case of
diff --git a/util/include/platforms/unix/axutil_thread_unix.h b/util/include/platforms/unix/axutil_thread_unix.h
index 5ee5992..ebd9490 100644
--- a/util/include/platforms/unix/axutil_thread_unix.h
+++ b/util/include/platforms/unix/axutil_thread_unix.h
@@ -56,4 +56,10 @@ struct axutil_thread_mutex_t
     pthread_mutex_t mutex;
 };
 
+struct axutil_thread_cond_t
+{
+    axutil_allocator_t *allocator;
+    pthread_cond_t cond;
+};
+
 #endif                          /* AXIS2_THREAD_UNIX_H */
diff --git a/util/src/Makefile.am b/util/src/Makefile.am
index f9ed14b..e1acf32 100644
--- a/util/src/Makefile.am
+++ b/util/src/Makefile.am
@@ -18,6 +18,7 @@ libaxutil_la_SOURCES = hash.c \
                         network_handler.c \
                         file.c\
                         uuid_gen.c\
+						pool_queue.c\
                         thread_pool.c \
                         property.c \
                         types.c \
diff --git a/util/src/platforms/unix/thread_unix.c b/util/src/platforms/unix/thread_unix.c
index 7fbbc5c..d183488 100644
--- a/util/src/platforms/unix/thread_unix.c
+++ b/util/src/platforms/unix/thread_unix.c
@@ -86,7 +86,7 @@ axutil_threadattr_detach_get(
 }
 
 static void *
-dummy_worker(
+axutil_thread_dummy_worker(
     void *opaque)
 {
     axutil_thread_t *thread = (axutil_thread_t *) opaque;
@@ -129,10 +129,11 @@ axutil_thread_create(
         temp = NULL;
     }
 
-    if ((stat = pthread_create(new->td, temp, dummy_worker, new)) == 0)
+    if ((stat = pthread_create(new->td, temp, axutil_thread_dummy_worker, new)) == 0)
     {
         return new;
     }
+
     return NULL;
 }
 
@@ -345,3 +346,59 @@ axutil_thread_mutex_destroy(
     AXIS2_FREE(mutex->allocator, mutex);
     return AXIS2_SUCCESS;
 }
+
+AXIS2_EXTERN axutil_thread_cond_t *AXIS2_CALL
+axutil_thread_cond_create(
+    axutil_allocator_t * allocator,
+    unsigned int flags)
+{
+    axutil_thread_cond_t *new_cond = NULL;
+
+    new_cond = AXIS2_MALLOC(allocator, sizeof(axutil_thread_cond_t));
+    new_cond->allocator = allocator;
+
+    if (pthread_cond_init(&(new_cond->cond), NULL) != 0)
+    {
+        AXIS2_FREE(allocator, new_cond);
+        return NULL;
+    }
+    return new_cond;
+}
+
+AXIS2_EXTERN axis2_status_t AXIS2_CALL
+axutil_thread_cond_wait(
+    axutil_thread_cond_t *cond,
+    axutil_thread_mutex_t * mutex)
+{
+    if (pthread_cond_wait(&(cond->cond), &(mutex->mutex)) != 0)
+    {
+        return AXIS2_FAILURE;
+    }
+    return AXIS2_SUCCESS;
+}
+
+AXIS2_EXTERN axis2_status_t AXIS2_CALL
+axutil_thread_cond_signal(
+    axutil_thread_cond_t *cond)
+{
+    if (pthread_cond_signal(&(cond->cond)) != 0)
+    {
+        return AXIS2_FAILURE;
+    }
+    return AXIS2_SUCCESS;
+}
+
+
+AXIS2_EXTERN axis2_status_t AXIS2_CALL
+axutil_thread_cond_destroy(
+    axutil_thread_cond_t * cond)
+{
+    if (0 != pthread_cond_destroy(&(cond->cond)))
+    {
+        return AXIS2_FAILURE;
+    }
+    AXIS2_FREE(cond->allocator, cond);
+    return AXIS2_SUCCESS;
+}
+
+
diff --git a/util/src/thread_pool.c b/util/src/thread_pool.c
index 0eb9f95..f87925a 100644
--- a/util/src/thread_pool.c
+++ b/util/src/thread_pool.c
@@ -18,27 +18,164 @@
 #include <axutil_thread_pool.h>
 #include <axutil_env.h>
 #include <axutil_error_default.h>
+#include <axutil_pool_queue.h>
+#include <stdio.h>
+
+typedef enum 
+{
+    ALL_RUN, ALL_EXIT
+} poolstate_t;
 
 struct axutil_thread_pool
 {
     axutil_allocator_t *allocator;
+    /* Mutex to synchronize the read/write operations */
+    axutil_thread_mutex_t *mutex;
+    int poolsize;
+    axutil_thread_cond_t *job_posted; /* dispatcher: "Hey guys, there's a job!" */
+    axutil_thread_cond_t *job_taken; /* a worker: "Got it!" */
+    poolstate_t state; /* Threads check this before getting job. */
+    int arrsz; /* Number of entries in array. */
+    int live; /* Number of live threads in pool (when pool is being destroyed, live<=arrsz)
*/
+    axutil_pool_queue_head_t *queue; /* queue of work orders */
+    struct timeval created; /* When the threadpool was created. */
+    axutil_thread_t **array; /* threads themselves. */
 };
 
+/* Define the life of a working thread.*/
+
+static void * axutil_thread_pool_do_work(
+        axutil_thread_t *axis_thd,
+        void * owning_pool) 
+{
+    /* Convert pointer to owning pool to proper type. */
+    axutil_thread_pool_t *pool = (axutil_thread_pool_t *) owning_pool;
+  
+    /* Remember my creation sequence number */
+    /*int myid = pool->live;*/
+
+    /* When we get a posted job, we copy it into these local vars. */
+    axutil_thread_start_t myjob;
+    void *myarg;  
+    axutil_thread_start_t mycleaner;
+    void *mycleanarg;
+
+    /*pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS,NULL);
+    pthread_cleanup_push(pthread_mutex_unlock, (void *) &pool->mutex);*/
+
+    /* Grab mutex so we can begin waiting for a job */
+    axutil_thread_mutex_lock(pool->mutex);
+  
+    /* Main loop: wait for job posting, do job(s) ... forever */
+    for( ; ; ) 
+    {
+        while(axutil_pool_queue_is_job_available(pool->queue) == 0)
+        {
+	        axutil_thread_cond_wait(pool->job_posted, pool->mutex);
+        }
+  
+        /* We've just woken up and we have the mutex.  Check pool's state */
+        if (ALL_EXIT == pool->state)
+        {
+            break;
+        }
+    
+        /* while we find work to do */
+        axutil_pool_queue_get_work_order(pool->queue, &myjob, &myarg, &mycleaner,
&mycleanarg);
+        axutil_thread_cond_signal(pool->job_taken);
+    
+        /* Yield mutex so other jobs can be posted */
+        axutil_thread_mutex_unlock(pool->mutex);
+    
+        /* Run the job we've taken */
+        /*if(mycleaner)
+        {
+            pthread_cleanup_push(mycleaner,mycleanarg);
+            myjob(axis_thd, myarg);
+            pthread_cleanup_pop(1);
+        }
+        else*/
+        {
+	        myjob(axis_thd, myarg);
+        }
+    
+        /* Grab mutex so we can grab posted job, or (if no job is posted) begin waiting for
next 
+         * posting. 
+         */
+        axutil_thread_mutex_lock(pool->mutex);
+    }
+  
+    /* If we get here, we broke from loop because state is ALL_EXIT. */
+    --pool->live;
+  
+    /* We're not really taking a job ... but this signals the destroyer that one thread has
exited, 
+     * so it can keep on destroying. pthread_cond_signal(&pool->job_taken);
+     */
+  
+    axutil_thread_mutex_unlock(pool->mutex);
+
+    /*pthread_cleanup_pop(1);*/
+    return NULL;
+}  
+
+
 AXIS2_EXTERN axutil_thread_pool_t *AXIS2_CALL
 axutil_thread_pool_init(
-    axutil_allocator_t *allocator)
+    axutil_allocator_t *allocator,
+    int poolsize)
 {
     axutil_thread_pool_t *pool = NULL;
+    int i = 0;
+
+    if ((poolsize <= 0) || (poolsize > MAXT_IN_POOL))
+    {
+        return NULL;
+    }
 
-    pool =
-        (axutil_thread_pool_t *) AXIS2_MALLOC(allocator,
-                                              sizeof(axutil_thread_pool_t));
+    pool = (axutil_thread_pool_t *) AXIS2_MALLOC(allocator, sizeof(axutil_thread_pool_t));
 
     if (!pool)
     {
         return NULL;
     }
     pool->allocator = allocator;
+    pool->mutex = axutil_thread_mutex_create(allocator, AXIS2_THREAD_MUTEX_DEFAULT);
+    pool->job_posted = axutil_thread_cond_create(allocator, AXIS2_THREAD_MUTEX_DEFAULT);
+    pool->job_taken = axutil_thread_cond_create(allocator, AXIS2_THREAD_MUTEX_DEFAULT);
+    pool->arrsz = poolsize;
+    pool->state = ALL_RUN;
+    pool->queue = axutil_pool_queue_make_queue(poolsize);
+    gettimeofday(pool->created, NULL);
+
+    /* create the array of threads within the pool */
+    pool->array = (axutil_thread_t **) AXIS2_MALLOC(allocator, pool->arrsz * sizeof(axutil_thread_t
*));
+    if (!pool->array) 
+    {
+        AXIS2_FREE(allocator, pool);
+        pool = NULL;
+        return NULL;
+    }
+
+    /* bring each thread to life (update counters in loop so threads can
+     * access pool->live to find out their ID#
+     */
+    for (i = 0; i < pool->arrsz; ++i) 
+    {
+        axutil_thread_t *axis2_thd = NULL;
+
+        axis2_thd = axutil_thread_create(allocator, NULL, axutil_thread_pool_do_work, (void
*) pool);
+        if(!axis2_thd)
+        {
+            AXIS2_FREE(allocator, pool);
+            return NULL;
+        }
+
+        pool->array[i] = axis2_thd;
+
+        pool->live = i+1;
+
+        axutil_thread_detach(pool->array[i]); /* automatic cleanup when thread exits.
*/
+    }
 
     return pool;
 }
@@ -57,23 +194,53 @@ axutil_thread_pool_free(
     }
     AXIS2_FREE(pool->allocator, pool);
     return;
-}
 
-AXIS2_EXTERN axutil_thread_t *AXIS2_CALL
-axutil_thread_pool_get_thread(
+  }
+
+AXIS2_EXTERN axis2_status_t AXIS2_CALL
+axutil_thread_pool_dispatch(
     axutil_thread_pool_t *pool,
     axutil_thread_start_t func,
     void *data)
 {
+    /*int old_cancel;*/
+
     if (!pool)
     {
-        return NULL;
+        return AXIS2_FAILURE;
     }
     if (!pool->allocator)
     {
-        return NULL;
+        return AXIS2_FAILURE;
+    }
+
+    if(pool == (axutil_thread_pool_t *) data)
+    {
     }
-    return axutil_thread_create(pool->allocator, NULL, func, data);
+    else
+    {
+        /*pthread_cleanup_push(pthread_mutex_unlock, (void *) &pool->mutex);*/
+      
+        /* Grab the mutex */
+        axutil_thread_mutex_lock(pool->mutex);
+
+        while(!axutil_pool_queue_can_accept_work(pool->queue))
+        {
+            axutil_thread_cond_signal(pool->job_posted);
+            axutil_thread_cond_wait(pool->job_taken, pool->mutex);
+        }
+      
+        /* Finally, there's room to post a job. Do so and signal workers. */
+        axutil_pool_queue_add_work_order(pool->queue, func, data, NULL, NULL);
+
+        axutil_thread_cond_signal(pool->job_posted);
+      
+        /* Yield mutex so a worker can pick up the job */
+        axutil_thread_mutex_unlock(pool->mutex);
+        /*pthread_cleanup_pop(1);*/
+    }
+
+    return AXIS2_SUCCESS;
 }
 
 AXIS2_EXTERN axis2_status_t AXIS2_CALL


Mime
View raw message