summaryrefslogtreecommitdiff
path: root/src/thrd
diff options
context:
space:
mode:
Diffstat (limited to 'src/thrd')
-rw-r--r--src/thrd/CVS/Entries8
-rwxr-xr-xsrc/thrd/pool.cpp245
-rwxr-xr-xsrc/thrd/pool.h65
-rw-r--r--src/thrd/thro.cpp1
-rw-r--r--src/thrd/thro.h2
5 files changed, 198 insertions, 123 deletions
diff --git a/src/thrd/CVS/Entries b/src/thrd/CVS/Entries
index bb6bf86..cca2803 100644
--- a/src/thrd/CVS/Entries
+++ b/src/thrd/CVS/Entries
@@ -1,5 +1,5 @@
-/pool.cpp/1.17/Fri Feb 25 01:22:35 2005//
-/pool.h/1.12/Sun Mar 6 20:44:47 2005//
-/thro.cpp/1.7/Sun Mar 6 20:44:47 2005//
-/thro.h/1.6/Sun Mar 6 20:44:47 2005//
+/pool.cpp/1.13/Mon Feb 21 01:55:50 2005//
+/pool.h/1.10/Mon Feb 21 01:55:50 2005//
+/thro.cpp/1.6/Mon Feb 21 01:55:50 2005//
+/thro.h/1.5/Mon Feb 21 01:55:50 2005//
D
diff --git a/src/thrd/pool.cpp b/src/thrd/pool.cpp
index 1f4ace6..d25a873 100755
--- a/src/thrd/pool.cpp
+++ b/src/thrd/pool.cpp
@@ -5,154 +5,217 @@
using namespace std;
+int pool::i_thrd_used = 0;
+
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 = tool::string2int( wrap::CONF->get_elem( "httpd.thread.initpoolsize" ) );
- increase_pool(i_num_avail_threads);
+ i_thrd_pool_size = tool::string2int( wrap::CONF->get_elem( "httpd.thread.initpoolsize" ) );
+ i_thrd_pool_queue = tool::string2int( wrap::CONF->get_elem( "httpd.thread.queuesize" ) );
+ tpool_init( &thread_pool, i_thrd_pool_size, i_thrd_pool_queue);
}
-pool::~pool()
+void
+pool::tpool_init( tpool_t *tpoolp, int num_worker_threads, int max_queue_size)
{
- pthread_mutex_lock(&mut_queue_tasks);
- while (!queue_tasks.empty())
+ int i, rtn;
+ tpool_t tpool;
+
+ // allocate a pool data structure
+ if (( tpool = (tpool_t) malloc( sizeof( struct tpool ) ) ) == 0 )
{
- delete queue_tasks.front();
- queue_tasks.pop();
+ wrap::system_message( POOLERR );
+ exit(-1);
}
- 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);
-}
+ // initialize th fields
+ tpool->num_threads = num_worker_threads;
+ tpool->max_queue_size = max_queue_size;
-int
-pool::increase_pool(int i_num)
-{
- wrap::system_message(POOLFLL + tool::int2string(i_num) +","+tool::int2string(i_num_total_threads)+")");
- int i_max_pool_size = tool::string2int( wrap::CONF->get_elem( "httpd.thread.maxpoolsize" ) );
+ if ( ( tpool->threads = (pthread_t*) malloc( sizeof(pthread_t)*num_worker_threads ) ) == 0 )
+ {
+ wrap::system_message( POOLERR );
+ exit(-1);
+ }
+
+ tpool->cur_queue_size = 0;
+ tpool->queue_head = 0;
+ tpool->queue_tail = 0;
- for ( int i = 0; i < i_num; ++i )
+ if ( ( rtn = pthread_mutex_init( &(tpool->queue_lock), 0 ) ) != 0 )
{
- if ( i_max_pool_size != 0 && i_num_total_threads >= i_max_pool_size )
- {
- wrap::system_message(POOLER2+tool::int2string(i_max_pool_size)+")");
- wrap::system_message(POOLER1+tool::int2string(i)+")");
- return i;
- }
+ string s_err( "pthread_mutex_init " );
+ s_err.append( strerror( rtn ) );
+
+ wrap::system_message( s_err );
- ++i_num_total_threads;
- pthread_t p_pthread;
- pthread_create(&p_pthread, 0, wait_for_task, (void*) this );
+ exit(-1);
}
+ else if ( ( rtn = pthread_cond_init( &(tpool->queue_not_empty), 0 ) ) != 0 )
+ {
+ string s_err( "pthread_cond_init (1): " );
+ s_err.append( strerror( rtn ) );
- return i_num;
-}
+ wrap::system_message( s_err );
-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);
+ exit(1);
+ }
+ else if ( ( rtn = pthread_cond_init( &(tpool->queue_not_full), 0 ) ) != 0 )
+ {
+ string s_err( "pthread_cond_init (2): " );
+ s_err.append( strerror( rtn ) );
- pthread_cond_signal(&cond_new_task);
+ wrap::system_message( s_err );
+ exit(1);
+ }
+ else if ( ( rtn = pthread_cond_init( &(tpool->queue_empty), 0 ) ) != 0 )
+ {
+ string s_err( "pthread_mutex_init " );
+ s_err.append( strerror( rtn ) );
+
+ wrap::system_message( s_err );
+
+ exit(1);
+ }
+
+ // create threads
+ for ( i = 0; i < num_worker_threads; ++i )
+ pthread_create( &(tpool->threads[i]) , 0, tpool_thread, (void*)tpool );
+
+ *tpoolp = tpool;
}
void*
-pool::wait_for_task( void* p_void )
+pool::tpool_thread( void* p_void )
{
- pool* p_pool = static_cast<pool*>(p_void);
+ tpool_t tpool = (tpool_t) p_void;
+ tpool_work_t *my_workp;
- for (;;)
+ for( pthread_mutex_lock( &(tpool->queue_lock) );;
+ pthread_mutex_lock( &(tpool->queue_lock) ), --i_thrd_used )
{
#ifdef NCURSES
- p_pool->print_pool_size();
+ print_threads(i_thrd_used);
#endif
- pthread_mutex_lock(&p_pool->mut_threads);
- pthread_cond_wait(&p_pool->cond_new_task, &p_pool->mut_threads);
+ while (tpool->cur_queue_size == 0)
+ pthread_cond_wait( &(tpool->queue_not_empty), &(tpool->queue_lock) );
- 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);
+ my_workp = tpool->queue_head;
+ tpool->cur_queue_size--;
- 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);
+ if ( tpool->cur_queue_size == 0)
+ tpool->queue_head = tpool->queue_tail = 0;
- pthread_mutex_unlock(&p_pool->mut_threads);
+ else
+ tpool->queue_head = my_workp->next;
- (*(p_task->p_func))(p_task->p_void);
- delete p_task;
+ if ( tpool->cur_queue_size == ( tpool->max_queue_size - 1 ) )
+ pthread_cond_signal( &(tpool->queue_not_full) );
- 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->cur_queue_size == 0 )
+ pthread_cond_signal( &(tpool->queue_empty) );
- return 0;
-}
+ pthread_mutex_unlock( &(tpool->queue_lock) );
-void
-pool::run(void* p_void)
-{
- add_task(run_func, p_void);
+ (*(my_workp->routine))(my_workp->p_void);
+
+ free(my_workp);
+ }
}
-void
-pool::run_func(void *p_void)
+void pool::run_func( void *p_void )
{
- int* p_sock = static_cast<int*>(p_void);
- wrap::SOCK->read_write(p_sock);
+ int* p_sock = (int*)p_void;
+ wrap::SOCK->read_write( p_sock );
delete p_sock;
}
-bool
-pool::allow_user_login()
+int
+pool::tpool_add_work( tpool_t tpool, void(*routine)(void*), void* p_void ) ///
{
- pthread_mutex_lock(&mut_num_avail_threads);
- if ( i_num_avail_threads < 2 )
+ tpool_work_t *workp;
+ pthread_mutex_lock( &(tpool->queue_lock) );
+
+ if ( ++i_thrd_used == tpool->num_threads )
{
int i_max_pool_size = tool::string2int( wrap::CONF->get_elem( "httpd.thread.maxpoolsize" ) );
- if ( i_max_pool_size != 0 && i_max_pool_size == i_num_total_threads )
+ if ( i_max_pool_size != 0 && i_thrd_used > i_max_pool_size )
{
- pthread_mutex_unlock(&mut_num_avail_threads);
- return false;
+ wrap::system_message(POOLER2+tool::int2string(i_thrd_used)+")");
}
+ else
+ {
+ int i_size = tpool->num_threads + 1;
+
+ wrap::system_message(POOLFLL+tool::int2string(i_size)+")");
+
+ tpool->threads = (pthread_t*)realloc((void*)tpool->threads, sizeof(pthread_t)*tpool->num_threads);
+
+ for ( int i = tpool->num_threads; i < i_size; ++i )
+ pthread_create( &(tpool->threads[i]) , 0, tpool_thread, (void*)tpool );
+
+ i_thrd_pool_size = tpool->num_threads = i_size;
+#ifdef NCURSES
+
+ print_pool_size();
+#endif
+
+ }
+ }
+
+#ifdef NCURSES
+ print_threads(i_thrd_used);
+#endif
+
+ while (tpool->cur_queue_size == tpool->max_queue_size)
+ pthread_cond_wait( &(tpool->queue_not_full), &(tpool->queue_lock) );
+
+ // allocate work structure:
+ workp = (tpool_work_t*) malloc( sizeof( tpool_work_t ) );
+
+ workp->routine = routine;
+ workp->p_void = p_void;
+ workp->next = 0;
+
+ if (tpool->cur_queue_size == 0 )
+ {
+ tpool->queue_tail = tpool->queue_head = workp;
}
- pthread_mutex_unlock(&mut_num_avail_threads);
+ else
+ {
+ (tpool->queue_tail)->next = workp;
+ tpool->queue_tail = workp;
+ }
+
+ tpool->cur_queue_size++;
+ pthread_cond_signal( &tpool->queue_not_empty );
+ pthread_mutex_unlock( &(tpool->queue_lock) );
- return true;
+ return 0;
}
#ifdef NCURSES
void
+pool::print_threads(int i_thrd_used)
+{
+ if ( wrap::NCUR->is_ready() )
+ {
+ mvprintw( NCUR_POOL_RUNNING_X,NCUR_POOL_RUNNING_Y, "In use: %d ", i_thrd_used);
+ refresh();
+ }
+}
+
+void
pool::print_pool_size()
{
if ( wrap::NCUR->is_ready() )
{
- pthread_mutex_lock(&mut_num_avail_threads);
- mvprintw( NCUR_POOL_WAIT_X,NCUR_POOL_WAIT_Y, "Wait/Tot: %d/%d ", i_num_avail_threads, i_num_total_threads);
- mvprintw( NCUR_POOL_RUNNING_X,NCUR_POOL_RUNNING_Y, "Running: %d ", i_num_total_threads-i_num_avail_threads);
- pthread_mutex_unlock(&mut_num_avail_threads);
+ mvprintw( NCUR_POOL_SIZE_X,NCUR_POOL_SIZE_Y, "Size: %d %d", i_thrd_pool_size, i_thrd_pool_queue );
refresh();
}
}
#endif
+
#endif
diff --git a/src/thrd/pool.h b/src/thrd/pool.h
index ab03b57..71ecaa9 100755
--- a/src/thrd/pool.h
+++ b/src/thrd/pool.h
@@ -3,52 +3,65 @@
#ifndef POOL_H
#define POOL_H
-#include <queue>
-
using namespace std;
-class pool
+class pool
{
private:
- friend class thro;
+ static int i_thrd_used;
- struct task
+ typedef struct tpool_work
{
- void(*p_func)(void*);
+ void (*routine)(void*); ///
void *p_void;
+ struct tpool_work *next;
+ }
+ tpool_work_t;
+
+ typedef struct tpool
+ {
+ // pool characteristics:
+ int num_threads;
+ int max_queue_size;
- task(void(*p_func)(void*), void *p_void)
- {
- this->p_func = p_func;
- this->p_void = p_void;
- }
- };
+ // pool state
+ pthread_t *threads;
+ int cur_queue_size;
- pthread_mutex_t mut_threads;
- pthread_mutex_t mut_queue_tasks;
- pthread_mutex_t mut_num_avail_threads;
- pthread_cond_t cond_new_task;
+ tpool_work_t *queue_head;
+ tpool_work_t *queue_tail;
- int i_num_avail_threads;
- int i_num_total_threads;
+ pthread_mutex_t queue_lock;
+ pthread_cond_t queue_not_empty;
+ pthread_cond_t queue_not_full;
+ pthread_cond_t queue_empty;
+ }
+ *tpool_t;
- queue<task*> queue_tasks;
+ int i_thrd_pool_size;
+ int i_thrd_pool_queue;
- int increase_pool(int i_num);
- void add_task( void(*p_func)(void*), void* p_void );
- static void* wait_for_task(void *p_void);
- static void run_func(void *p_void);
+ tpool_t thread_pool;
+
+ void tpool_init( tpool_t *tpoolp, int num_worker_threads, int max_queue_size);
+ int tpool_add_work( tpool_t tpool, void(*routine)(void*), void* p_void );
+ static void* tpool_thread( void *p_void);
+ static void run_func( void *p_void );
public:
pool();
- ~pool();
- void run(void* p_void);
- bool allow_user_login();
+ // inline (speed)!
+ void run( void *p_void )
+ {
+ tpool_add_work( thread_pool, run_func, p_void );
+ }
#ifdef NCURSES
void print_pool_size();
+ static void print_threads(int i_thrd_used);
#endif
+
};
#endif
diff --git a/src/thrd/thro.cpp b/src/thrd/thro.cpp
index 8b3f1ba..e35520c 100644
--- a/src/thrd/thro.cpp
+++ b/src/thrd/thro.cpp
@@ -23,7 +23,6 @@ thro::run( void *p_void )
{
elem.p_thro = this;
elem.p_void = p_void;
- //wrap::POOL->add_task(start_, &elem);
pthread_create( &pthread, NULL, start_, &elem );
}
diff --git a/src/thrd/thro.h b/src/thrd/thro.h
index 8e7e0cf..783cd0d 100644
--- a/src/thrd/thro.h
+++ b/src/thrd/thro.h
@@ -17,7 +17,7 @@ private:
}
elem;
- static void *start_( void *p_void );
+ static void* start_( void *p_void );
public:
thro( );
generated by cgit v1.2.3 (git 2.25.1) at 2026-06-08 12:11:04 +0000