Nilorea Library

- Main Page
- Modules
- Data Structures
- Files

- File List
- Globals

All Data Structures Files Functions Variables Typedefs Macros Groups
n_thread_pool.c
Go to the documentation of this file.
1
7 #include "n_thread_pool.h"
8
9
15 void *thread_pool_processing_function( void *param )
16 {
17 THREAD_POOL_NODE *node = (THREAD_POOL_NODE *)param ;
18
19 pthread_mutex_lock( &node -> lock );
20 int state = node -> state ;
21 pthread_mutex_unlock( &node -> lock );
22
23 while( state != EXITING_THREAD )
24 {
25 sem_wait( &node -> th_start );
26
27 pthread_mutex_lock( &node -> lock );
28 state = node -> state ;
29
30 if( node -> state != IDLE_THREAD && node -> param != NULL )
31 {
32 node -> state = RUNNING_THREAD ;
33 pthread_mutex_unlock( &node -> lock );
34 node -> func( node -> param ) ;
35 pthread_mutex_lock( &node -> lock );
36 }
37
38 node -> func = NULL ;
39 node -> param = NULL ;
40 node -> state = IDLE_THREAD ;
41 int type = node -> type ;
42 node -> type = -1 ;
43
44 pthread_mutex_unlock( &node -> lock );
45
46 if( type == SYNCED_PROC )
47 sem_post( &node -> th_end );
48 }
49
50 n_log( LOG_NOTICE , "Thread %d exited" , node -> thr );
51
52 return NULL ;
53 }
54
55
56
62 THREAD_POOL *new_thread_pool( int nbmaxthr )
63 {
64 THREAD_POOL *thread_pool = NULL ;
65
66 Malloc( thread_pool , THREAD_POOL , 1 );
67 if( !thread_pool )
68 return NULL ;
69
70 thread_pool -> max_threads = nbmaxthr ;
71
72 thread_pool -> thread_list = (THREAD_POOL_NODE **)malloc( nbmaxthr * sizeof(THREAD_POOL_NODE *) );
73 if( !thread_pool -> thread_list )
74 {
75 Free( thread_pool );
76 return NULL ;
77 }
78
79 for( int it = 0 ; it < nbmaxthr ; it ++ )
80 {
81 thread_pool -> thread_list[ it ] = (THREAD_POOL_NODE *)malloc( sizeof(THREAD_POOL_NODE ) );
82 thread_pool -> thread_list[ it ] -> type = -1 ;
83 thread_pool -> thread_list[ it ] -> state = IDLE_THREAD ;
84 if( sem_init( &thread_pool -> thread_list[ it ] -> th_start , 0 , 0 ) == -1 )
85 {
86 fprintf( stderr , "sem_init failed : %s on &thread_pool -> thread_list[ %d ] -> th_start\n" , strerror ( errno ) , it );
87 return NULL;
88 }
89 if( sem_init( &thread_pool -> thread_list[ it ] -> th_end , 0 , 0 ) == -1 )
90 {
91 fprintf( stderr , "sem_init failed : %s on &thread_pool -> thread_list[ %d] -> th_end\n" , strerror ( errno ) , it );
92 return NULL;
93 }
94 thread_pool -> thread_list[ it ] -> func = NULL ;
95 thread_pool -> thread_list[ it ] -> param = NULL ;
96
97 pthread_mutex_init( &thread_pool -> thread_list[ it ] -> lock , NULL );
98
99 if( pthread_create( &thread_pool -> thread_list[ it ] -> thr , NULL, thread_pool_processing_function , (void *)thread_pool -> thread_list[ it ] ) != 0 )
100 {
101 fprintf( stderr , "pthread_create failed : %s for it %d\n" , strerror ( errno ) , it );
102 return NULL;
103 }
104 }
105 return thread_pool ;
106 }
107
108
109
110
111
120 int add_threaded_process( THREAD_POOL *thread_pool , void *(*func_ptr)(void *param) , void * param , int mode )
121 {
122
123 if( !thread_pool )
124 return FALSE ;
125
126 if( !thread_pool -> thread_list )
127 return FALSE ;
128
129 int it = 0 ;
130 while( it < thread_pool -> max_threads )
131 {
132 pthread_mutex_lock( &thread_pool -> thread_list[ it ] -> lock );
133 if( thread_pool -> thread_list[ it ] -> state == IDLE_THREAD )
134 {
135 break ;
136 }
137 pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
138 it ++ ;
139 }
140 if( it == thread_pool -> max_threads )
141 {
142 n_log( LOG_ERR , "Thread pool is full: %d process already running" , it );
143 return FALSE ;
144 }
145
146 switch( mode )
147 {
148 case DIRECT_PROC :
149 thread_pool -> thread_list[ it ] -> func = func_ptr ;
150 thread_pool -> thread_list[ it ] -> param = param ;
151 thread_pool -> thread_list[ it ] -> state = BUSY_THREAD ;
152 thread_pool -> thread_list[ it ] -> type = DIRECT_PROC ;
153 pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
154 sem_post( &thread_pool -> thread_list[ it ] -> th_start );
155 break ;
156 case SYNCED_PROC :
157 thread_pool -> thread_list[ it ] -> func = func_ptr ;
158 thread_pool -> thread_list[ it ] -> param = param ;
159 thread_pool -> thread_list[ it ] -> state = BUSY_THREAD ;
160 thread_pool -> thread_list[ it ] -> type = SYNCED_PROC ;
161 pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
162 break ;
163 default:
164 n_log( LOG_ERR , "Unknown mode %d for thread %d" , mode , it );
165 pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
166 return FALSE ;
167 }
168 return TRUE ;
169 }
170
171
172
178 int start_threaded_pool( THREAD_POOL *thread_pool )
179 {
180 if( !thread_pool )
181 return FALSE ;
182
183 if( !thread_pool -> thread_list )
184 return FALSE ;
185
186 for( int it = 0 ; it < thread_pool -> max_threads ; it ++ )
187 {
188 int to_run = 0 ;
189 pthread_mutex_lock( &thread_pool -> thread_list[ it ] -> lock );
190 if( thread_pool -> thread_list[ it ] -> type == SYNCED_PROC && thread_pool -> thread_list[ it ] -> state == BUSY_THREAD )
191 {
192 to_run = 1 ;
193 }
194 pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
195 if( to_run == 1 )
196 {
197 if( sem_post( &thread_pool -> thread_list[ it ] -> th_start ) != 0 )
198 return FALSE ;
199 }
200 }
201 return TRUE ;
202 }
203
204
205
211 int wait_for_threaded_pool( THREAD_POOL *thread_pool )
212 {
213 if( !thread_pool )
214 return FALSE ;
215
216 if( !thread_pool -> thread_list )
217 return FALSE ;
218
219 for( int it = 0 ; it < thread_pool -> max_threads ; it ++ )
220 {
221 int to_check = 0 ;
222 pthread_mutex_lock( &thread_pool -> thread_list[ it ] -> lock );
223 if( thread_pool -> thread_list[ it ] -> type == SYNCED_PROC && thread_pool -> thread_list[ it ] -> state != IDLE_THREAD )
224 {
225 to_check = 1 ;
226 }
227 pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
228 if( to_check == 1 )
229 {
230 sem_wait( &thread_pool -> thread_list[ it ] -> th_end );
231 }
232 pthread_mutex_lock( &thread_pool -> thread_list[ it ] -> lock );
233 thread_pool -> thread_list[ it ] -> state = IDLE_THREAD ;
234 pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
235 }
236 return TRUE ;
237 }