diff options
Diffstat (limited to '')
-rw-r--r-- | libraries/luaproc/luaproc.c | 837 |
1 files changed, 837 insertions, 0 deletions
diff --git a/libraries/luaproc/luaproc.c b/libraries/luaproc/luaproc.c new file mode 100644 index 0000000..2ec5b31 --- /dev/null +++ b/libraries/luaproc/luaproc.c | |||
@@ -0,0 +1,837 @@ | |||
1 | /*************************************************** | ||
2 | |||
3 | Copyright 2008 Alexandre Skyrme, Noemi Rodriguez, Roberto Ierusalimschy | ||
4 | |||
5 | Permission is hereby granted, free of charge, to any person obtaining a copy | ||
6 | of this software and associated documentation files (the "Software"), to deal | ||
7 | in the Software without restriction, including without limitation the rights | ||
8 | to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | ||
9 | copies of the Software, and to permit persons to whom the Software is | ||
10 | furnished to do so, subject to the following conditions: | ||
11 | |||
12 | The above copyright notice and this permission notice shall be included in | ||
13 | all copies or substantial portions of the Software. | ||
14 | |||
15 | THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | ||
16 | IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | ||
17 | FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | ||
18 | AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | ||
19 | LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | ||
20 | OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN | ||
21 | THE SOFTWARE. | ||
22 | |||
23 | ***************************************************** | ||
24 | |||
25 | [luaproc.c] | ||
26 | |||
27 | ****************************************************/ | ||
28 | |||
29 | #include <netdb.h> | ||
30 | #include <pthread.h> | ||
31 | #include <stdio.h> | ||
32 | #include <stdlib.h> | ||
33 | #include <string.h> | ||
34 | #include <unistd.h> | ||
35 | #include <lua.h> | ||
36 | #include <lauxlib.h> | ||
37 | #include <lualib.h> | ||
38 | |||
39 | #include "luaproc.h" | ||
40 | #include "list.h" | ||
41 | #include "sched.h" | ||
42 | #include "channel.h" | ||
43 | |||
44 | #define FALSE 0 | ||
45 | #define TRUE 1 | ||
46 | |||
47 | /********* | ||
48 | * globals | ||
49 | *********/ | ||
50 | |||
51 | /* channel operations mutex */ | ||
52 | pthread_mutex_t mutex_channel = PTHREAD_MUTEX_INITIALIZER; | ||
53 | |||
54 | /* recycle list mutex */ | ||
55 | pthread_mutex_t mutex_recycle_list = PTHREAD_MUTEX_INITIALIZER; | ||
56 | |||
57 | /* recycled lua process list */ | ||
58 | list recyclelp = NULL; | ||
59 | |||
60 | /* maximum lua processes to recycle */ | ||
61 | int recyclemax = 0; | ||
62 | |||
63 | /* lua process */ | ||
64 | struct stluaproc { | ||
65 | lua_State *lstate; | ||
66 | int stat; | ||
67 | int args; | ||
68 | channel chan; | ||
69 | int destroyworker; | ||
70 | }; | ||
71 | |||
72 | /****************************** | ||
73 | * library functions prototypes | ||
74 | ******************************/ | ||
75 | /* create a new lua process */ | ||
76 | static int luaproc_create_newproc( lua_State *L ); | ||
77 | /* send a message to a lua process */ | ||
78 | static int luaproc_send( lua_State *L ); | ||
79 | /* receive a message from a lua process */ | ||
80 | static int luaproc_receive( lua_State *L ); | ||
81 | /* create a new channel */ | ||
82 | static int luaproc_create_channel( lua_State *L ); | ||
83 | /* destroy a channel */ | ||
84 | static int luaproc_destroy_channel( lua_State *L ); | ||
85 | /* wait until all luaprocs have finished and exit */ | ||
86 | static int luaproc_exit( lua_State *L ); | ||
87 | /* create a new worker */ | ||
88 | static int luaproc_create_worker( lua_State *L ); | ||
89 | /* destroy a worker */ | ||
90 | static int luaproc_destroy_worker( lua_State *L ); | ||
91 | /* set amount of lua processes that should be recycled (ie, reused) */ | ||
92 | static int luaproc_recycle_set( lua_State *L ); | ||
93 | |||
94 | /* luaproc function registration array - main (parent) functions */ | ||
95 | static const struct luaL_reg luaproc_funcs_parent[] = { | ||
96 | { "newproc", luaproc_create_newproc }, | ||
97 | { "exit", luaproc_exit }, | ||
98 | { "createworker", luaproc_create_worker }, | ||
99 | { "destroyworker", luaproc_destroy_worker }, | ||
100 | { "recycle", luaproc_recycle_set }, | ||
101 | { NULL, NULL } | ||
102 | }; | ||
103 | |||
104 | /* luaproc function registration array - newproc (child) functions */ | ||
105 | static const struct luaL_reg luaproc_funcs_child[] = { | ||
106 | { "newproc", luaproc_create_newproc }, | ||
107 | { "send", luaproc_send }, | ||
108 | { "receive", luaproc_receive }, | ||
109 | { "newchannel", luaproc_create_channel }, | ||
110 | { "delchannel", luaproc_destroy_channel }, | ||
111 | { "createworker", luaproc_create_worker }, | ||
112 | { "destroyworker", luaproc_destroy_worker }, | ||
113 | { "recycle", luaproc_recycle_set }, | ||
114 | { NULL, NULL } | ||
115 | }; | ||
116 | |||
117 | static void registerlib( lua_State *L, const char *name, lua_CFunction f ) { | ||
118 | lua_getglobal( L, "package" ); | ||
119 | lua_getfield( L, -1, "preload" ); | ||
120 | lua_pushcfunction( L, f ); | ||
121 | lua_setfield( L, -2, name ); | ||
122 | lua_pop( L, 2 ); | ||
123 | } | ||
124 | |||
125 | static void openlibs( lua_State *L ) { | ||
126 | lua_cpcall( L, luaopen_base, NULL ); | ||
127 | lua_cpcall( L, luaopen_package, NULL ); | ||
128 | registerlib( L, "io", luaopen_io ); | ||
129 | registerlib( L, "os", luaopen_os ); | ||
130 | registerlib( L, "table", luaopen_table ); | ||
131 | registerlib( L, "string", luaopen_string ); | ||
132 | registerlib( L, "math", luaopen_math ); | ||
133 | registerlib( L, "debug", luaopen_debug ); | ||
134 | } | ||
135 | |||
136 | /* return status (boolean) indicating if lua process should be recycled */ | ||
137 | luaproc luaproc_recycle_pop( void ) { | ||
138 | |||
139 | luaproc lp; | ||
140 | node n; | ||
141 | |||
142 | /* get exclusive access to operate on recycle list */ | ||
143 | pthread_mutex_lock( &mutex_recycle_list ); | ||
144 | |||
145 | /* check if there are any lua processes on recycle list */ | ||
146 | if ( list_node_count( recyclelp ) > 0 ) { | ||
147 | /* pop list head */ | ||
148 | n = list_pop_head( recyclelp ); | ||
149 | /* free access to operate on recycle list */ | ||
150 | pthread_mutex_unlock( &mutex_recycle_list ); | ||
151 | /* find associated luaproc */ | ||
152 | lp = (luaproc )list_data( n ); | ||
153 | /* destroy node (but not associated luaproc) */ | ||
154 | list_destroy_node( n ); | ||
155 | /* return associated luaproc */ | ||
156 | return lp; | ||
157 | } | ||
158 | |||
159 | /* free access to operate on recycle list */ | ||
160 | pthread_mutex_unlock( &mutex_recycle_list ); | ||
161 | |||
162 | /* if no lua processes are available simply return null */ | ||
163 | return NULL; | ||
164 | } | ||
165 | |||
166 | /* check if lua process should be recycled and, in case so, add it to the recycle list */ | ||
167 | int luaproc_recycle_push( luaproc lp ) { | ||
168 | |||
169 | node n; | ||
170 | |||
171 | /* get exclusive access to operate on recycle list */ | ||
172 | pthread_mutex_lock( &mutex_recycle_list ); | ||
173 | |||
174 | /* check if amount of lua processes currently on recycle list is greater than | ||
175 | or equal to the maximum amount of lua processes that should be recycled */ | ||
176 | if ( list_node_count( recyclelp ) >= recyclemax ) { | ||
177 | /* free access to operate on recycle list */ | ||
178 | pthread_mutex_unlock( &mutex_recycle_list ); | ||
179 | /* if so, lua process should NOT be recycled and should be destroyed */ | ||
180 | return FALSE; | ||
181 | } | ||
182 | /* otherwise, lua process should be added to recycle list */ | ||
183 | n = list_new_node( lp ); | ||
184 | if ( n == NULL ) { | ||
185 | /* free access to operate on recycle list */ | ||
186 | pthread_mutex_unlock( &mutex_recycle_list ); | ||
187 | /* in case of errors, lua process should be destroyed */ | ||
188 | return FALSE; | ||
189 | } | ||
190 | list_add( recyclelp, n ); | ||
191 | /* free access to operate on recycle list */ | ||
192 | pthread_mutex_unlock( &mutex_recycle_list ); | ||
193 | /* since lua process will be recycled, it should not be destroyed */ | ||
194 | return TRUE; | ||
195 | } | ||
196 | |||
197 | /* create new luaproc */ | ||
198 | luaproc luaproc_new( const char *code, int destroyflag ) { | ||
199 | |||
200 | luaproc lp; | ||
201 | int ret; | ||
202 | /* create new lua state */ | ||
203 | lua_State *lpst = luaL_newstate( ); | ||
204 | /* store the luaproc struct in its own lua state */ | ||
205 | lp = (luaproc )lua_newuserdata( lpst, sizeof( struct stluaproc )); | ||
206 | lua_setfield( lpst, LUA_REGISTRYINDEX, "_SELF" ); | ||
207 | |||
208 | lp->lstate = lpst; | ||
209 | lp->stat = LUAPROC_STAT_IDLE; | ||
210 | lp->args = 0; | ||
211 | lp->chan = NULL; | ||
212 | lp->destroyworker = destroyflag; | ||
213 | |||
214 | /* load standard libraries */ | ||
215 | openlibs( lpst ); | ||
216 | |||
217 | /* register luaproc's own functions */ | ||
218 | luaL_register( lpst, "luaproc", luaproc_funcs_child ); | ||
219 | |||
220 | /* load process' code */ | ||
221 | ret = luaL_loadstring( lpst, code ); | ||
222 | /* in case of errors, destroy recently created lua process */ | ||
223 | if ( ret != 0 ) { | ||
224 | lua_close( lpst ); | ||
225 | return NULL; | ||
226 | } | ||
227 | |||
228 | /* return recently created lua process */ | ||
229 | return lp; | ||
230 | } | ||
231 | |||
232 | /* synchronize worker threads and exit */ | ||
233 | static int luaproc_exit( lua_State *L ) { | ||
234 | sched_join_workerthreads( ); | ||
235 | return 0; | ||
236 | } | ||
237 | |||
238 | /* create a new worker pthread */ | ||
239 | static int luaproc_create_worker( lua_State *L ) { | ||
240 | |||
241 | if ( sched_create_worker( ) != LUAPROC_SCHED_OK ) { | ||
242 | lua_pushnil( L ); | ||
243 | lua_pushstring( L, "error creating worker" ); | ||
244 | return 2; | ||
245 | } | ||
246 | |||
247 | lua_pushboolean( L, TRUE ); | ||
248 | return 1; | ||
249 | } | ||
250 | |||
251 | /* set amount of lua processes that should be recycled (ie, reused) */ | ||
252 | static int luaproc_recycle_set( lua_State *L ) { | ||
253 | |||
254 | node n; | ||
255 | luaproc lp; | ||
256 | int max = luaL_checkint( L, 1 ); | ||
257 | |||
258 | /* check if function argument represents a reasonable value */ | ||
259 | if ( max < 0 ) { | ||
260 | /* in case of errors return nil + error msg */ | ||
261 | lua_pushnil( L ); | ||
262 | lua_pushstring( L, "error setting recycle limit to negative value" ); | ||
263 | return 2; | ||
264 | } | ||
265 | |||
266 | /* get exclusive access to operate on recycle list */ | ||
267 | pthread_mutex_lock( &mutex_recycle_list ); | ||
268 | |||
269 | /* set maximum lua processes that should be recycled */ | ||
270 | recyclemax = max; | ||
271 | |||
272 | /* destroy recycle list excessive nodes (and corresponding lua processes) */ | ||
273 | while ( list_node_count( recyclelp ) > max ) { | ||
274 | /* get first node from recycle list */ | ||
275 | n = list_pop_head( recyclelp ); | ||
276 | /* find associated luaproc */ | ||
277 | lp = (luaproc )list_data( n ); | ||
278 | /* destroy node */ | ||
279 | list_destroy_node( n ); | ||
280 | /* close associated lua_State */ | ||
281 | lua_close( lp->lstate ); | ||
282 | } | ||
283 | |||
284 | /* free access to operate on recycle list */ | ||
285 | pthread_mutex_unlock( &mutex_recycle_list ); | ||
286 | |||
287 | lua_pushboolean( L, TRUE ); | ||
288 | return 1; | ||
289 | } | ||
290 | |||
291 | |||
292 | /* destroy a worker pthread */ | ||
293 | static int luaproc_destroy_worker( lua_State *L ) { | ||
294 | |||
295 | /* new lua process pointer */ | ||
296 | luaproc lp; | ||
297 | |||
298 | /* create new lua process with empty code and destroy worker flag set to true | ||
299 | (ie, conclusion of lua process WILL result in worker thread destruction */ | ||
300 | lp = luaproc_new( "", TRUE ); | ||
301 | |||
302 | /* ensure process creation was successfull */ | ||
303 | if ( lp == NULL ) { | ||
304 | /* in case of errors return nil + error msg */ | ||
305 | lua_pushnil( L ); | ||
306 | lua_pushstring( L, "error destroying worker" ); | ||
307 | return 2; | ||
308 | } | ||
309 | |||
310 | /* increase active luaproc count */ | ||
311 | sched_lpcount_inc(); | ||
312 | |||
313 | /* schedule luaproc */ | ||
314 | if ( sched_queue_proc( lp ) != LUAPROC_SCHED_QUEUE_PROC_OK ) { | ||
315 | printf( "[luaproc] error queueing Lua process\n" ); | ||
316 | /* decrease active luaproc count */ | ||
317 | sched_lpcount_dec(); | ||
318 | /* close lua_State */ | ||
319 | lua_close( lp->lstate ); | ||
320 | /* return nil + error msg */ | ||
321 | lua_pushnil( L ); | ||
322 | lua_pushstring( L, "error destroying worker" ); | ||
323 | return 2; | ||
324 | } | ||
325 | |||
326 | lua_pushboolean( L, TRUE ); | ||
327 | return 1; | ||
328 | } | ||
329 | |||
330 | /* recycle a lua process */ | ||
331 | luaproc luaproc_recycle( luaproc lp, const char *code ) { | ||
332 | |||
333 | int ret; | ||
334 | |||
335 | /* reset struct members */ | ||
336 | lp->stat = LUAPROC_STAT_IDLE; | ||
337 | lp->args = 0; | ||
338 | lp->chan = NULL; | ||
339 | lp->destroyworker = FALSE; | ||
340 | |||
341 | /* load process' code */ | ||
342 | ret = luaL_loadstring( lp->lstate, code ); | ||
343 | |||
344 | /* in case of errors, destroy lua process */ | ||
345 | if ( ret != 0 ) { | ||
346 | lua_close( lp->lstate ); | ||
347 | return NULL; | ||
348 | } | ||
349 | |||
350 | /* return recycled lua process */ | ||
351 | return lp; | ||
352 | } | ||
353 | |||
354 | /* create and schedule a new lua process (luaproc.newproc) */ | ||
355 | static int luaproc_create_newproc( lua_State *L ) { | ||
356 | |||
357 | /* check if first argument is a string (lua code) */ | ||
358 | const char *code = luaL_checkstring( L, 1 ); | ||
359 | |||
360 | /* new lua process pointer */ | ||
361 | luaproc lp; | ||
362 | |||
363 | /* check if existing lua process should be recycled to avoid new creation */ | ||
364 | lp = luaproc_recycle_pop( ); | ||
365 | |||
366 | /* if there is a lua process available on the recycle queue, recycle it */ | ||
367 | if ( lp != NULL ) { | ||
368 | lp = luaproc_recycle( lp, code ); | ||
369 | } | ||
370 | /* otherwise create a new one from scratch */ | ||
371 | else { | ||
372 | /* create new lua process with destroy worker flag set to false | ||
373 | (ie, conclusion of lua process will NOT result in worker thread destruction */ | ||
374 | lp = luaproc_new( code, FALSE ); | ||
375 | } | ||
376 | |||
377 | /* ensure process creation was successfull */ | ||
378 | if ( lp == NULL ) { | ||
379 | /* in case of errors return nil + error msg */ | ||
380 | lua_pushnil( L ); | ||
381 | lua_pushstring( L, "error loading code string" ); | ||
382 | return 2; | ||
383 | } | ||
384 | |||
385 | /* increase active luaproc count */ | ||
386 | sched_lpcount_inc(); | ||
387 | |||
388 | /* schedule luaproc */ | ||
389 | if ( sched_queue_proc( lp ) != LUAPROC_SCHED_QUEUE_PROC_OK ) { | ||
390 | printf( "[luaproc] error queueing Lua process\n" ); | ||
391 | /* decrease active luaproc count */ | ||
392 | sched_lpcount_dec(); | ||
393 | /* close lua_State */ | ||
394 | lua_close( lp->lstate ); | ||
395 | /* return nil + error msg */ | ||
396 | lua_pushnil( L ); | ||
397 | lua_pushstring( L, "error queuing process" ); | ||
398 | return 2; | ||
399 | } | ||
400 | |||
401 | lua_pushboolean( L, TRUE ); | ||
402 | return 1; | ||
403 | } | ||
404 | |||
405 | /* queue a lua process sending a message without a matching receiver */ | ||
406 | void luaproc_queue_sender( luaproc lp ) { | ||
407 | /* add the sending process to this process' send queue */ | ||
408 | list_add( channel_get_sendq( lp->chan ), list_new_node( lp )); | ||
409 | } | ||
410 | |||
411 | /* dequeue a lua process sending a message with a receiver match */ | ||
412 | luaproc luaproc_dequeue_sender( channel chan ) { | ||
413 | |||
414 | node n; | ||
415 | luaproc lp; | ||
416 | |||
417 | if ( list_node_count( channel_get_sendq( chan )) > 0 ) { | ||
418 | /* get first node from channel's send queue */ | ||
419 | n = list_pop_head( channel_get_sendq( chan )); | ||
420 | /* find associated luaproc */ | ||
421 | lp = (luaproc )list_data( n ); | ||
422 | /* destroy node (but not associated luaproc) */ | ||
423 | list_destroy_node( n ); | ||
424 | /* return associated luaproc */ | ||
425 | return lp; | ||
426 | } | ||
427 | |||
428 | return NULL; | ||
429 | } | ||
430 | |||
431 | /* queue a luc process receiving a message without a matching sender */ | ||
432 | void luaproc_queue_receiver( luaproc lp ) { | ||
433 | /* add the receiving process to this process' receive queue */ | ||
434 | list_add( channel_get_recvq( lp->chan ), list_new_node( lp )); | ||
435 | } | ||
436 | |||
437 | /* dequeue a lua process receiving a message with a sender match */ | ||
438 | luaproc luaproc_dequeue_receiver( channel chan ) { | ||
439 | |||
440 | node n; | ||
441 | luaproc lp; | ||
442 | |||
443 | if ( list_node_count( channel_get_recvq( chan )) > 0 ) { | ||
444 | /* get first node from channel's recv queue */ | ||
445 | n = list_pop_head( channel_get_recvq( chan )); | ||
446 | /* find associated luaproc */ | ||
447 | lp = (luaproc )list_data( n ); | ||
448 | /* destroy node (but not associated luaproc) */ | ||
449 | list_destroy_node( n ); | ||
450 | /* return associated luaproc */ | ||
451 | return lp; | ||
452 | } | ||
453 | |||
454 | return NULL; | ||
455 | } | ||
456 | |||
457 | /* moves values between lua states' stacks */ | ||
458 | void luaproc_movevalues( lua_State *Lfrom, lua_State *Lto ) { | ||
459 | |||
460 | int i; | ||
461 | int n = lua_gettop( Lfrom ); | ||
462 | |||
463 | /* move values between lua states' stacks */ | ||
464 | for ( i = 2; i <= n; i++ ) { | ||
465 | lua_pushstring( Lto, lua_tostring( Lfrom, i )); | ||
466 | } | ||
467 | } | ||
468 | |||
469 | /* return the lua process associated with a given lua state */ | ||
470 | luaproc luaproc_getself( lua_State *L ) { | ||
471 | luaproc lp; | ||
472 | lua_getfield( L, LUA_REGISTRYINDEX, "_SELF" ); | ||
473 | lp = (luaproc )lua_touserdata( L, -1 ); | ||
474 | lua_pop( L, 1 ); | ||
475 | return lp; | ||
476 | } | ||
477 | |||
478 | /* send a message to a lua process */ | ||
479 | static int luaproc_send( lua_State *L ) { | ||
480 | |||
481 | channel chan; | ||
482 | luaproc dstlp, self; | ||
483 | const char *chname = luaL_checkstring( L, 1 ); | ||
484 | |||
485 | /* get exclusive access to operate on channels */ | ||
486 | pthread_mutex_lock( &mutex_channel ); | ||
487 | |||
488 | /* wait until channel is not in use */ | ||
489 | while((( chan = channel_search( chname )) != NULL ) && ( pthread_mutex_trylock( channel_get_mutex( chan )) != 0 )) { | ||
490 | pthread_cond_wait( channel_get_cond( chan ), &mutex_channel ); | ||
491 | } | ||
492 | |||
493 | /* free access to operate on channels */ | ||
494 | pthread_mutex_unlock( &mutex_channel ); | ||
495 | |||
496 | /* if channel is not found, return an error to Lua */ | ||
497 | if ( chan == NULL ) { | ||
498 | lua_pushnil( L ); | ||
499 | lua_pushstring( L, "non-existent channel" ); | ||
500 | return 2; | ||
501 | } | ||
502 | |||
503 | /* try to find a matching receiver */ | ||
504 | dstlp = luaproc_dequeue_receiver( chan ); | ||
505 | |||
506 | /* if a match is found, move values to it and (queue) wake it */ | ||
507 | if ( dstlp != NULL ) { | ||
508 | |||
509 | /* move values between Lua states' stacks */ | ||
510 | luaproc_movevalues( L, dstlp->lstate ); | ||
511 | |||
512 | dstlp->args = lua_gettop( dstlp->lstate ) - 1; | ||
513 | |||
514 | if ( sched_queue_proc( dstlp ) != LUAPROC_SCHED_QUEUE_PROC_OK ) { | ||
515 | |||
516 | /* unlock channel access */ | ||
517 | luaproc_unlock_channel( chan ); | ||
518 | |||
519 | /* decrease active luaproc count */ | ||
520 | sched_lpcount_dec(); | ||
521 | |||
522 | /* close lua_State */ | ||
523 | lua_close( dstlp->lstate ); | ||
524 | lua_pushnil( L ); | ||
525 | lua_pushstring( L, "error scheduling process" ); | ||
526 | return 2; | ||
527 | } | ||
528 | |||
529 | /* unlock channel access */ | ||
530 | luaproc_unlock_channel( chan ); | ||
531 | } | ||
532 | |||
533 | /* otherwise queue (block) the sending process */ | ||
534 | else { | ||
535 | |||
536 | self = luaproc_getself( L ); | ||
537 | |||
538 | if ( self != NULL ) { | ||
539 | self->stat = LUAPROC_STAT_BLOCKED_SEND; | ||
540 | self->chan = chan; | ||
541 | } | ||
542 | |||
543 | /* just yield the lua process, channel unlocking will be done by the scheduler */ | ||
544 | return lua_yield( L, lua_gettop( L )); | ||
545 | } | ||
546 | |||
547 | lua_pushboolean( L, TRUE ); | ||
548 | return 1; | ||
549 | } | ||
550 | |||
551 | /* receive a message from a lua process */ | ||
552 | static int luaproc_receive( lua_State *L ) { | ||
553 | |||
554 | channel chan; | ||
555 | luaproc srclp, self; | ||
556 | const char *chname = luaL_checkstring( L, 1 ); | ||
557 | |||
558 | /* get exclusive access to operate on channels */ | ||
559 | pthread_mutex_lock( &mutex_channel ); | ||
560 | |||
561 | /* wait until channel is not in use */ | ||
562 | while((( chan = channel_search( chname )) != NULL ) && ( pthread_mutex_trylock( channel_get_mutex( chan )) != 0 )) { | ||
563 | pthread_cond_wait( channel_get_cond( chan ), &mutex_channel ); | ||
564 | } | ||
565 | |||
566 | /* free access to operate on channels */ | ||
567 | pthread_mutex_unlock( &mutex_channel ); | ||
568 | |||
569 | /* if channel is not found, free access to operate on channels and return an error to Lua */ | ||
570 | if ( chan == NULL ) { | ||
571 | lua_pushnil( L ); | ||
572 | lua_pushstring( L, "non-existent channel" ); | ||
573 | return 2; | ||
574 | } | ||
575 | |||
576 | /* try to find a matching sender */ | ||
577 | srclp = luaproc_dequeue_sender( chan ); | ||
578 | |||
579 | /* if a match is found, get values from it and (queue) wake it */ | ||
580 | if ( srclp != NULL ) { | ||
581 | |||
582 | /* move values between Lua states' stacks */ | ||
583 | luaproc_movevalues( srclp->lstate, L ); | ||
584 | |||
585 | /* return to sender indicanting message was sent */ | ||
586 | lua_pushboolean( srclp->lstate, TRUE ); | ||
587 | srclp->args = 1; | ||
588 | |||
589 | if ( sched_queue_proc( srclp ) != LUAPROC_SCHED_QUEUE_PROC_OK ) { | ||
590 | |||
591 | /* unlock channel access */ | ||
592 | luaproc_unlock_channel( chan ); | ||
593 | |||
594 | /* decrease active luaproc count */ | ||
595 | sched_lpcount_dec(); | ||
596 | |||
597 | /* close lua_State */ | ||
598 | lua_close( srclp->lstate ); | ||
599 | lua_pushnil( L ); | ||
600 | lua_pushstring( L, "error scheduling process" ); | ||
601 | return 2; | ||
602 | } | ||
603 | |||
604 | /* unlock channel access */ | ||
605 | luaproc_unlock_channel( chan ); | ||
606 | |||
607 | return lua_gettop( L ) - 1; | ||
608 | } | ||
609 | |||
610 | /* otherwise queue (block) the receiving process (sync) or return immediatly (async) */ | ||
611 | else { | ||
612 | |||
613 | /* if trying an asynchronous receive, unlock channel access and return an error */ | ||
614 | if ( lua_toboolean( L, 2 )) { | ||
615 | /* unlock channel access */ | ||
616 | luaproc_unlock_channel( chan ); | ||
617 | /* return an error */ | ||
618 | lua_pushnil( L ); | ||
619 | lua_pushfstring( L, "no senders waiting on channel %s", chname ); | ||
620 | return 2; | ||
621 | } | ||
622 | |||
623 | /* otherwise (synchronous receive) simply block process */ | ||
624 | else { | ||
625 | self = luaproc_getself( L ); | ||
626 | |||
627 | if ( self != NULL ) { | ||
628 | self->stat = LUAPROC_STAT_BLOCKED_RECV; | ||
629 | self->chan = chan; | ||
630 | } | ||
631 | |||
632 | /* just yield the lua process, channel unlocking will be done by the scheduler */ | ||
633 | return lua_yield( L, lua_gettop( L )); | ||
634 | } | ||
635 | } | ||
636 | } | ||
637 | |||
638 | LUALIB_API int luaopen_luaproc( lua_State *L ) { | ||
639 | |||
640 | /* register luaproc functions */ | ||
641 | luaL_register( L, "luaproc", luaproc_funcs_parent ); | ||
642 | |||
643 | /* initialize recycle list */ | ||
644 | recyclelp = list_new(); | ||
645 | |||
646 | /* initialize local scheduler */ | ||
647 | sched_init_local( LUAPROC_SCHED_DEFAULT_WORKER_THREADS ); | ||
648 | |||
649 | return 0; | ||
650 | } | ||
651 | |||
652 | /* return a process' status */ | ||
653 | int luaproc_get_status( luaproc lp ) { | ||
654 | return lp->stat; | ||
655 | } | ||
656 | |||
657 | /* set a process' status */ | ||
658 | void luaproc_set_status( luaproc lp, int status ) { | ||
659 | lp->stat = status; | ||
660 | } | ||
661 | |||
662 | /* return a process' state */ | ||
663 | lua_State *luaproc_get_state( luaproc lp ) { | ||
664 | return lp->lstate; | ||
665 | } | ||
666 | |||
667 | /* return the number of arguments expected by a given process */ | ||
668 | int luaproc_get_args( luaproc lp ) { | ||
669 | return lp->args; | ||
670 | } | ||
671 | |||
672 | /* set the number of arguments expected by a given process */ | ||
673 | void luaproc_set_args( luaproc lp, int n ) { | ||
674 | lp->args = n; | ||
675 | } | ||
676 | |||
677 | /* create a new channel */ | ||
678 | static int luaproc_create_channel( lua_State *L ) { | ||
679 | |||
680 | const char *chname = luaL_checkstring( L, 1 ); | ||
681 | |||
682 | /* get exclusive access to operate on channels */ | ||
683 | pthread_mutex_lock( &mutex_channel ); | ||
684 | |||
685 | /* check if channel exists */ | ||
686 | if ( channel_search( chname ) != NULL ) { | ||
687 | /* free access to operate on channels */ | ||
688 | pthread_mutex_unlock( &mutex_channel ); | ||
689 | /* return an error to lua */ | ||
690 | lua_pushnil( L ); | ||
691 | lua_pushstring( L, "channel already exists" ); | ||
692 | return 2; | ||
693 | } | ||
694 | |||
695 | channel_create( chname ); | ||
696 | |||
697 | /* free access to operate on channels */ | ||
698 | pthread_mutex_unlock( &mutex_channel ); | ||
699 | |||
700 | lua_pushboolean( L, TRUE ); | ||
701 | |||
702 | return 1; | ||
703 | |||
704 | } | ||
705 | |||
706 | /* destroy a channel */ | ||
707 | static int luaproc_destroy_channel( lua_State *L ) { | ||
708 | |||
709 | channel chan; | ||
710 | luaproc lp; | ||
711 | node nitr; | ||
712 | pthread_mutex_t *chmutex; | ||
713 | pthread_cond_t *chcond; | ||
714 | const char *chname = luaL_checkstring( L, 1 ); | ||
715 | |||
716 | |||
717 | /* get exclusive access to operate on channels */ | ||
718 | pthread_mutex_lock( &mutex_channel ); | ||
719 | |||
720 | /* wait until channel is not in use */ | ||
721 | while((( chan = channel_search( chname )) != NULL ) && ( pthread_mutex_trylock( channel_get_mutex( chan )) != 0 )) { | ||
722 | pthread_cond_wait( channel_get_cond( chan ), &mutex_channel ); | ||
723 | } | ||
724 | |||
725 | /* free access to operate on channels */ | ||
726 | pthread_mutex_unlock( &mutex_channel ); | ||
727 | |||
728 | /* if channel is not found, return an error to Lua */ | ||
729 | if ( chan == NULL ) { | ||
730 | lua_pushnil( L ); | ||
731 | lua_pushstring( L, "non-existent channel" ); | ||
732 | return 2; | ||
733 | } | ||
734 | |||
735 | /* get channel's mutex and conditional pointers */ | ||
736 | chmutex = channel_get_mutex( chan ); | ||
737 | chcond = channel_get_cond( chan ); | ||
738 | |||
739 | /* search for processes waiting to send a message on this channel */ | ||
740 | while (( nitr = list_pop_head( channel_get_sendq( chan ))) != NULL ) { | ||
741 | |||
742 | lp = (luaproc )list_data( nitr ); | ||
743 | |||
744 | /* destroy node (but not associated luaproc) */ | ||
745 | list_destroy_node( nitr ); | ||
746 | |||
747 | /* return an error so the processe knows the channel was destroyed before the message was sent */ | ||
748 | lua_settop( lp->lstate, 0 ); | ||
749 | lua_pushnil( lp->lstate ); | ||
750 | lua_pushstring( lp->lstate, "channel destroyed while waiting for receiver" ); | ||
751 | lp->args = 2; | ||
752 | |||
753 | /* schedule the process for execution */ | ||
754 | if ( sched_queue_proc( lp ) != LUAPROC_SCHED_QUEUE_PROC_OK ) { | ||
755 | |||
756 | /* decrease active luaproc count */ | ||
757 | sched_lpcount_dec(); | ||
758 | |||
759 | /* close lua_State */ | ||
760 | lua_close( lp->lstate ); | ||
761 | } | ||
762 | } | ||
763 | |||
764 | /* search for processes waiting to receive a message on this channel */ | ||
765 | while (( nitr = list_pop_head( channel_get_recvq( chan ))) != NULL ) { | ||
766 | |||
767 | lp = (luaproc )list_data( nitr ); | ||
768 | |||
769 | /* destroy node (but not associated luaproc) */ | ||
770 | list_destroy_node( nitr ); | ||
771 | |||
772 | /* return an error so the processe knows the channel was destroyed before the message was received */ | ||
773 | lua_settop( lp->lstate, 0 ); | ||
774 | lua_pushnil( lp->lstate ); | ||
775 | lua_pushstring( lp->lstate, "channel destroyed while waiting for sender" ); | ||
776 | lp->args = 2; | ||
777 | |||
778 | /* schedule the process for execution */ | ||
779 | if ( sched_queue_proc( lp ) != LUAPROC_SCHED_QUEUE_PROC_OK ) { | ||
780 | |||
781 | /* decrease active luaproc count */ | ||
782 | sched_lpcount_dec(); | ||
783 | |||
784 | /* close lua_State */ | ||
785 | lua_close( lp->lstate ); | ||
786 | } | ||
787 | } | ||
788 | |||
789 | /* get exclusive access to operate on channels */ | ||
790 | pthread_mutex_lock( &mutex_channel ); | ||
791 | /* destroy channel */ | ||
792 | channel_destroy( chan, chname ); | ||
793 | /* broadcast channel not in use */ | ||
794 | pthread_cond_broadcast( chcond ); | ||
795 | /* unlock channel access */ | ||
796 | pthread_mutex_unlock( chmutex ); | ||
797 | /* destroy channel mutex and conditional */ | ||
798 | pthread_mutex_destroy( chmutex ); | ||
799 | pthread_cond_destroy( chcond ); | ||
800 | /* free memory used by channel mutex and conditional */ | ||
801 | free( chmutex ); | ||
802 | free( chcond ); | ||
803 | /* free access to operate on channels */ | ||
804 | pthread_mutex_unlock( &mutex_channel ); | ||
805 | |||
806 | lua_pushboolean( L, TRUE ); | ||
807 | |||
808 | return 1; | ||
809 | } | ||
810 | |||
811 | /* register luaproc's functions in a lua_State */ | ||
812 | void luaproc_register_funcs( lua_State *L ) { | ||
813 | luaL_register( L, "luaproc", luaproc_funcs_child ); | ||
814 | } | ||
815 | |||
816 | /* return the channel where the corresponding luaproc is blocked at */ | ||
817 | channel luaproc_get_channel( luaproc lp ) { | ||
818 | return lp->chan; | ||
819 | } | ||
820 | |||
821 | /* unlock access to a channel */ | ||
822 | void luaproc_unlock_channel( channel chan ) { | ||
823 | /* get exclusive access to operate on channels */ | ||
824 | pthread_mutex_lock( &mutex_channel ); | ||
825 | /* unlock channel access */ | ||
826 | pthread_mutex_unlock( channel_get_mutex( chan )); | ||
827 | /* signal channel not in use */ | ||
828 | pthread_cond_signal( channel_get_cond( chan )); | ||
829 | /* free access to operate on channels */ | ||
830 | pthread_mutex_unlock( &mutex_channel ); | ||
831 | } | ||
832 | |||
833 | /* return status (boolean) indicating if worker thread should be destroyed after luaproc execution */ | ||
834 | int luaproc_get_destroyworker( luaproc lp ) { | ||
835 | return lp->destroyworker; | ||
836 | } | ||
837 | |||