1 files changed, 164 insertions, 130 deletions

diff --git a/pool.cpp b/pool.cpp
index a1517f4..470ea24 100644..100755
--- a/pool.cpp
+++ b/pool.cpp
@@ -1,144 +1,153 @@
-#ifndef POOL_CPP
-#define POOL_CPP
+// class pool implementation.
+
+#ifndef POOL_CXX
+#define POOL_CXX
 
 #include "pool.h"
 
+#include "s_conf.h"
+#include "s_mutx.h"
+#include "s_tool.h"
+#include "thrd.h"
+
 using namespace std;
 
 pool::pool()
 {
-  pthread_mutex_init(&mut_threads, 0);
-  pthread_mutex_init(&mut_queue_tasks, 0);
-  pthread_mutex_init(&mut_num_avail_threads, 0);
-  pthread_cond_init(&cond_new_task, 0);
-
-  i_num_total_threads = 0;
-  i_num_avail_threads = s_tool::string2int( s_conf::get
-                          ().get_val( "THRDPOOL" ) );
-  increase_pool(i_num_avail_threads);
+ i_thrd_pool_size  = s_tool::string2int( s_conf::get().get_val( "THRDPOOL" ) );
+ i_thrd_pool_queue = s_tool::string2int( s_conf::get().get_val( "THRDQUEU" ) );
+
+ tpool_init( &thread_pool, i_thrd_pool_size, i_thrd_pool_queue, 0 );
 }
 
 pool::~pool()
 {
-  pthread_mutex_lock(&mut_queue_tasks);
-  while (!queue_tasks.empty())
-  {
-    delete queue_tasks.front();
-    queue_tasks.pop();
-  }
-  pthread_mutex_unlock(&mut_queue_tasks);
-
-  pthread_mutex_destroy(&mut_threads);
-  pthread_mutex_destroy(&mut_queue_tasks);
-  pthread_mutex_destroy(&mut_num_avail_threads);
-  pthread_cond_destroy(&cond_new_task);
+ // tpool_destroy ...
 }
 
-int
-pool::increase_pool(int i_num)
+void
+pool::tpool_init( tpool_t *tpoolp, int num_worker_threads, int max_queue_size, int do_not_block_when_full )
 {
-#ifdef VERBOSE
-  pthread_mutex_lock  ( &s_mutx::get
-                          ().mut_stdout );
-  cout << POOLFLL + s_tool::int2string(i_num) +","+s_tool::int2string(i_num_total_threads)+")" << endl;
-  pthread_mutex_unlock( &s_mutx::get
-                          ().mut_stdout );
-#endif
+ int i, rtn;
+ tpool_t tpool;
+
+ // allocate a pool data structure
+ if (( tpool = (tpool_t) malloc( sizeof( struct tpool ) ) ) == NULL )
+ {
+  pthread_mutex_lock  ( &s_mutx::get().mut_stdout );
+  cerr << "malloc" << endl;
+  pthread_mutex_unlock( &s_mutx::get().mut_stdout );
+  exit(-1);
+ }
+
+ // initialize th fields
+ tpool->num_threads    = num_worker_threads;
+ tpool->max_queue_size = max_queue_size;
+ tpool->do_not_block_when_full = do_not_block_when_full;
+
+ if ( ( tpool->threads = (pthread_t*) malloc( sizeof( pthread_t ) *num_worker_threads ) ) == NULL )
+ {
+  pthread_mutex_lock  ( &s_mutx::get().mut_stdout );
+  cerr << "malloc" << endl;
+  pthread_mutex_unlock( &s_mutx::get().mut_stdout );
+  exit(-1);
+ }
+
+ tpool->cur_queue_size = 0;
+ tpool->queue_head     = NULL;
+ tpool->queue_tail     = NULL;
+ tpool->queue_closed   = 0;
+ tpool->shutdown       = 0;
+
+ if ( ( rtn = pthread_mutex_init( &(tpool->queue_lock), NULL ) ) != 0 )
+ {
+  pthread_mutex_lock  ( &s_mutx::get().mut_stdout );
+  cerr << "pthread_mutex_init " << strerror( rtn ) << endl;
+  pthread_mutex_unlock( &s_mutx::get().mut_stdout );
+  exit(-1);
+ }
+
+ else if ( ( rtn = pthread_cond_init( &(tpool->queue_not_empty), NULL ) ) != 0 )
+ {
+  pthread_mutex_lock  ( &s_mutx::get().mut_stdout );
+  cerr << "pthread_cond_init " << strerror( rtn ) << endl;
+  pthread_mutex_unlock( &s_mutx::get().mut_stdout );
+  exit(-1);
+ }
+
+ else if ( ( rtn = pthread_cond_init( &(tpool->queue_not_full), NULL ) ) != 0 )
+ {
+  pthread_mutex_lock  ( &s_mutx::get().mut_stdout );
+  cerr << "pthread_cond_init " << strerror( rtn ) << endl;
+  pthread_mutex_unlock( &s_mutx::get().mut_stdout );
+  exit(-1);
+ }
+
+ else if ( ( rtn = pthread_cond_init( &(tpool->queue_empty), NULL ) ) != 0 )
+ {
+  pthread_mutex_lock  ( &s_mutx::get().mut_stdout );
+  cerr << "pthread_cond_init " << strerror( rtn ) << endl;
+  pthread_mutex_unlock( &s_mutx::get().mut_stdout );
+  exit(-1);
+ }
+ // create threads
+ for ( i = 0; i < num_worker_threads; i++ )
+  pthread_create( &(tpool->threads[i]) , NULL, tpool_thread, (void*)tpool );
+
+ *tpoolp = tpool;
+}
 
-  int i_max_pool_size = s_tool::string2int( s_conf::get
-                          ().get_val( "THRDPMAX" ) );
+void*
+pool::tpool_thread( void* arg )
+{
+ tpool_t tpool = (tpool_t) arg;
+ tpool_work_t *my_workp;
 
-  for ( int i = 0; i < i_num; ++i )
-  {
-    if ( i_max_pool_size != 0 && i_num_total_threads >= i_max_pool_size )
-    {
-#ifdef VERBOSE
-      pthread_mutex_lock  ( &s_mutx::get
-                              ().mut_stdout );
-      cout << POOLER2+s_tool::int2string(i_max_pool_size)+")" << endl;
-      cout << POOLER1+s_tool::int2string(i)+")" << endl;
-      pthread_mutex_unlock( &s_mutx::get
-                              ().mut_stdout );
-#endif
+ while( true )
+ {
+  pthread_mutex_lock( &(tpool->queue_lock) );
 
-      return i;
-    }
+  while ( (tpool->cur_queue_size == 0) && (!tpool->shutdown) )
+   pthread_cond_wait( &(tpool->queue_not_empty), &(tpool->queue_lock) );
 
-    ++i_num_total_threads;
-    pthread_t p_pthread;
-    pthread_create(&p_pthread, 0, wait_for_task, (void*) this );
+  if (tpool->shutdown)
+  {
+   pthread_mutex_unlock( &(tpool->queue_lock) );
+   pthread_exit( NULL );
   }
 
-  return i_num;
-}
-
-void
-pool::add_task( void(*p_func)(void*), void* p_void )
-{
-  pthread_mutex_lock(&mut_queue_tasks);
-  queue_tasks.push(new task(p_func, p_void));
-  pthread_mutex_unlock(&mut_queue_tasks);
+  my_workp = tpool->queue_head;
+  tpool->cur_queue_size--;
 
-  pthread_cond_signal(&cond_new_task);
+  if ( tpool->cur_queue_size == 0)
+   tpool->queue_head = tpool->queue_tail = NULL;
 
-}
+  else
+   tpool->queue_head = my_workp->next;
 
-void*
-pool::wait_for_task( void* p_void )
-{
-  pool* p_pool = static_cast<pool*>(p_void);
-
-  for (;;)
-  {
-    pthread_mutex_lock(&p_pool->mut_threads);
-    pthread_cond_wait(&p_pool->cond_new_task, &p_pool->mut_threads);
-
-    pthread_mutex_lock(&p_pool->mut_num_avail_threads);
-    if ( --p_pool->i_num_avail_threads < 5 )
-    {
-      int i_size = 9 - p_pool->i_num_avail_threads;
-      i_size = p_pool->increase_pool(i_size);
-      p_pool->i_num_avail_threads += i_size;
-    }
-    pthread_mutex_unlock(&p_pool->mut_num_avail_threads);
-
-    pthread_mutex_lock(&p_pool->mut_queue_tasks);
-    task* p_task = p_pool->queue_tasks.front();
-    p_pool->queue_tasks.pop();
-    pthread_mutex_unlock(&p_pool->mut_queue_tasks);
-
-    pthread_mutex_unlock(&p_pool->mut_threads);
-
-    (*(p_task->p_func))(p_task->p_void);
-    delete p_task;
-
-    pthread_mutex_lock(&p_pool->mut_num_avail_threads);
-    p_pool->i_num_avail_threads++;
-    pthread_mutex_unlock(&p_pool->mut_num_avail_threads);
-  }
+  if ( ( ! tpool->do_not_block_when_full ) &&
+       (   tpool->cur_queue_size == ( tpool->max_queue_size - 1 ) ) )
+   pthread_cond_signal( &(tpool->queue_not_full) ); 
 
-  return 0;
-}
+  if ( tpool->cur_queue_size == 0 )
+   pthread_cond_signal( &(tpool->queue_empty) );
 
-void
-pool::run(void* p_void)
-{
-  add_task(run_func, p_void);
+  pthread_mutex_unlock( &(tpool->queue_lock) );
+  (*(my_workp->routine))(my_workp->arg);
+  free((void*)my_workp);
+ }
 }
 
-void
-pool::run_func(void *p_void)
+void pool::run_func( void *v_pointer )
 {
 #ifdef VERBOSE
-  pthread_mutex_lock  ( &s_mutx::get
-                          ().mut_stdout );
+  pthread_mutex_lock  ( &s_mutx::get().mut_stdout );
   cout << THREADS << endl;
-  pthread_mutex_unlock( &s_mutx::get
-                          ().mut_stdout );
+  pthread_mutex_unlock( &s_mutx::get().mut_stdout );
 #endif
 
   // recasting the client thread object.
-  thrd *t = (thrd*) p_void;
+  thrd *t = (thrd*) v_pointer;
 
   // start parsing the client request and sending response's back.
   t-> run ();
@@ -146,35 +155,60 @@ pool::run_func(void *p_void)
   // close the client socket.
   t->~thrd();
 
-  free(p_void);
+  free(v_pointer);
 
 #ifdef VERBOSE
-
-  pthread_mutex_lock  ( &s_mutx::get
-                          ().mut_stdout );
+  pthread_mutex_lock  ( &s_mutx::get().mut_stdout );
   cout << THREADE << endl;
-  pthread_mutex_unlock( &s_mutx::get
-                          ().mut_stdout );
+  pthread_mutex_unlock( &s_mutx::get().mut_stdout );
 #endif
 }
 
-bool
-pool::allow_user_login()
+int
+pool::tpool_add_work( tpool_t tpool, void(*routine)(void*), void* arg ) ///
 {
-  pthread_mutex_lock(&mut_num_avail_threads);
-  if ( i_num_avail_threads < 2 )
-  {
-    int i_max_pool_size = s_tool::string2int( s_conf::get
-                            ().get_val( "THRDPMAX" ) );
-    if ( i_max_pool_size != 0 && i_max_pool_size == i_num_total_threads )
-    {
-      pthread_mutex_unlock(&mut_num_avail_threads);
-      return false;
-    }
-  }
-  pthread_mutex_unlock(&mut_num_avail_threads);
-
-  return true;
+ tpool_work_t *workp;
+ pthread_mutex_lock( &(tpool->queue_lock) );
+
+ if( ( tpool->cur_queue_size == tpool->max_queue_size ) &&
+       tpool->do_not_block_when_full )
+ {
+  pthread_mutex_unlock( &(tpool->queue_lock) );
+  return -1;
+ }
+
+ while( (     tpool->cur_queue_size == tpool->max_queue_size ) &&
+        ( ! ( tpool->shutdown       || tpool->queue_closed   ) ) )
+  pthread_cond_wait( &(tpool->queue_not_full), &(tpool->queue_lock) );
+
+ if( tpool->shutdown || tpool->queue_closed )
+ {
+  pthread_mutex_unlock( &tpool->queue_lock );
+  return -1;
+ }
+
+ // allocate work structure:
+ workp = (tpool_work_t*) malloc( sizeof( tpool_work_t ) );
+
+ workp->routine = routine;
+ workp->arg     = arg;
+ workp->next    = NULL;
+  
+ if( tpool->cur_queue_size == 0 )
+ {
+  tpool->queue_tail = tpool->queue_head = workp;
+  pthread_cond_signal( &(tpool->queue_not_empty) );
+ } 
+
+ else
+ {
+  tpool->queue_tail->next = workp;
+  tpool->queue_tail       = workp;
+ } 
+
+ tpool->cur_queue_size++;
+ pthread_mutex_unlock( &(tpool->queue_lock) );
+ return 1;
 }
 
 #endif