Ticket #192: tqueue.deprecated.patch
| File tqueue.deprecated.patch, 20.2 KB (added by jkeenan, 13 years ago) |
|---|
-
src/pmc/tqueue.pmc
1 /*2 Copyright (C) 2001-2008, Parrot Foundation.3 $Id$4 5 =head1 NAME6 7 src/pmc/tqueue.pmc - Threadsafe Queue8 9 =head1 DESCRIPTION10 11 Threadsafe queue class for inter thread communication. If you have an12 unthreaded program then please use an Array-like PMC.13 14 new P0, 'TQueue'15 push P0, some16 new P2, 'ParrotThread'17 ...18 19 and in other thread (at least, when shared PMCs work :)20 21 shift P1, P022 23 Note: The TQueue must always be emptied before program exit.24 25 =head2 Methods26 27 =over 428 29 =cut30 31 */32 33 #include "parrot/parrot.h"34 35 pmclass TQueue need_ext is_shared {36 ATTR struct QUEUE *queue;37 ATTR INTVAL thread_count;38 39 /*40 41 =item C<void init()>42 43 Initializes the queue.44 45 =cut46 47 */48 49 VTABLE void init() {50 Parrot_TQueue_attributes* attrs =51 mem_allocate_zeroed_typed(Parrot_TQueue_attributes);52 53 attrs->thread_count = 0;54 attrs->queue = queue_init(0);55 PMC_data(SELF) = attrs;56 57 PObj_custom_mark_destroy_SETALL(SELF);58 }59 60 /*61 62 =item C<PMC *clone()>63 64 Returns the queue itself. No copy is made.65 66 =cut67 68 */69 70 VTABLE PMC *clone() {71 /* XXX fake a shared PMC */72 return SELF;73 }74 75 /*76 77 =item C<void mark()>78 79 Marks all the threads in the queue as live.80 81 =cut82 83 */84 85 VTABLE void mark() {86 QUEUE *queue;87 QUEUE_ENTRY *entry;88 89 GET_ATTR_queue(INTERP, SELF, queue);90 91 queue_lock(queue);92 entry = queue->head;93 94 while (entry) {95 pobject_lives(INTERP, (PObj *)entry->data);96 97 if (entry == queue->tail)98 break;99 100 entry = entry->next;101 }102 103 queue_unlock(queue);104 }105 106 /*107 108 =item C<void destroy()>109 110 Destroys the queue.111 112 =cut113 114 */115 116 VTABLE void destroy() {117 QUEUE *queue;118 GET_ATTR_queue(INTERP, SELF, queue);119 120 if (queue) {121 #if 0122 /*123 * wait til queue is empty124 * XXX implement a time wait and PANIC if queue125 * isn't empty after some TIMEOUT126 */127 while (SELF.elements()) {128 queue_lock(queue);129 queue_wait(queue);130 queue_unlock(queue);131 }132 #endif133 mem_sys_free(queue);134 }135 mem_sys_free(PMC_data(SELF));136 }137 138 /*139 140 =item C<INTVAL defined()>141 142 Returns whether there are any threads in the queue.143 144 =cut145 146 */147 148 VTABLE INTVAL defined() {149 return SELF.get_integer() != 0;150 }151 152 /*153 154 =item C<INTVAL get_integer()>155 156 =cut157 158 */159 160 VTABLE INTVAL get_integer() {161 162 INTVAL thread_count;163 GET_ATTR_thread_count(INTERP, SELF, thread_count);164 return thread_count;165 }166 167 /*168 169 =item C<INTVAL elements()>170 171 Returns the number of threads in the queue.172 173 =cut174 175 */176 177 VTABLE INTVAL elements() {178 return SELF.get_integer();179 }180 181 /*182 183 =item C<void push_pmc(PMC *item)>184 185 Adds the thread C<*item> to the end of the queue.186 187 =cut188 189 */190 191 void push_pmc(PMC *item) {192 QUEUE_ENTRY * const entry = mem_allocate_typed(QUEUE_ENTRY);193 QUEUE * queue;194 INTVAL thread_count;195 196 GET_ATTR_queue(INTERP, SELF, queue);197 198 /*199 * if item isn't shared nor const, then make200 * a shared item201 */202 if (!(item->vtable->flags &203 (VTABLE_IS_CONST_FLAG | VTABLE_IS_SHARED_FLAG)))204 VTABLE_share(INTERP, item);205 206 GC_WRITE_BARRIER(INTERP, SELF, NULL, item);207 208 entry->data = item;209 entry->type = QUEUE_ENTRY_TYPE_NONE;210 211 /* s. tsq.c:queue_push */212 queue_lock(queue);213 214 GET_ATTR_thread_count(INTERP, SELF, thread_count);215 ++thread_count;216 SET_ATTR_thread_count(INTERP, SELF, thread_count);217 218 /* Is there something in the queue? */219 if (queue->tail) {220 queue->tail->next = entry;221 queue->tail = entry;222 }223 else {224 queue->head = entry;225 queue->tail = entry;226 }227 228 /* signal all waiters */229 queue_broadcast(queue);230 queue_unlock(queue);231 }232 233 /*234 235 =item C<PMC *shift_pmc()>236 237 Removes the first thread from the start of the queue.238 239 =cut240 241 */242 243 VTABLE PMC *shift_pmc() {244 QUEUE *queue;245 QUEUE_ENTRY *entry;246 PMC *ret;247 INTVAL thread_count;248 249 GET_ATTR_queue(INTERP, SELF, queue);250 queue_lock(queue);251 252 while (queue->head == NULL) {253 queue_wait(queue);254 }255 256 entry = nosync_pop_entry(queue);257 GET_ATTR_thread_count(INTERP, SELF, thread_count);258 --thread_count;259 SET_ATTR_thread_count(INTERP, SELF, thread_count);260 261 queue_unlock(queue);262 263 ret = (PMC *)entry->data;264 mem_sys_free(entry);265 266 return ret;267 }268 }269 270 /*271 272 =back273 274 =cut275 276 */277 278 /*279 * Local variables:280 * c-file-style: "parrot"281 * End:282 * vim: expandtab shiftwidth=4:283 */ -
src/pmc/pmc.num
75 75 76 76 # other 77 77 78 tqueue.pmc 50 78 parrotclass.pmc 50 79 parrotobject.pmc 51 79 80 80 parrotclass.pmc 51 81 parrotobject.pmc 52 82 83 os.pmc 53 84 file.pmc 54 81 os.pmc 52 82 file.pmc 53 -
MANIFEST
1 1 # ex: set ro: 2 2 # $Id$ 3 3 # 4 # generated by tools/dev/mk_manifest_and_skip.pl Wed Feb 25 17:00:322009 UT4 # generated by tools/dev/mk_manifest_and_skip.pl Thu Feb 26 01:36:56 2009 UT 5 5 # 6 6 # See tools/dev/install_files.pl for documentation on the 7 7 # format of this file. … … 717 717 examples/pir/readline.pir [examples] 718 718 examples/pir/substr.pir [examples] 719 719 examples/pir/sudoku.pir [examples] 720 examples/pir/thr-primes.pir [examples]721 720 examples/pir/uniq.pir [examples] 722 721 examples/sdl/anim_image.pir [examples] 723 722 examples/sdl/anim_image_dblbuf.pir [examples] … … 2322 2321 src/pmc/sub.pmc [devel]src 2323 2322 src/pmc/task.pmc [devel]src 2324 2323 src/pmc/timer.pmc [devel]src 2325 src/pmc/tqueue.pmc [devel]src2326 2324 src/pmc/undef.pmc [devel]src 2327 2325 src/pmc/unmanagedstruct.pmc [devel]src 2328 2326 src/pmc_freeze.c [] … … 2770 2768 t/pmc/task.t [test] 2771 2769 t/pmc/threads.t [test] 2772 2770 t/pmc/timer.t [test] 2773 t/pmc/tqueue.t [test]2774 2771 t/pmc/undef.t [test] 2775 2772 t/pmc/unmanagedstruct.t [test] 2776 2773 t/postconfigure/01-options.t [test] -
editor/pir-mode.el
156 156 "ParrotRunningThread" "ParrotThread" "Pointer" "Random" "Ref" 157 157 "ResizableBooleanArray" "ResizableFloatArray" "ResizableIntegerArray" 158 158 "ResizablePMCArray" "ResizableStringArray" "RetContinuation" 159 "Role" "Scalar" "SharedRef" "Slice" "String" "Sub" "Super" "TQueue"159 "Role" "Scalar" "SharedRef" "Slice" "String" "Sub" "Super" 160 160 "Timer" "UnManagedStruct" "Undef" "VtableCache")) 161 161 162 162 (defvar pir-ops -
t/pmc/tqueue.t
1 #! parrot2 # Copyright (C) 2001-2005, Parrot Foundation.3 # $Id$4 5 =head1 NAME6 7 t/pmc/tqueue.t - Thread Queue8 9 =head1 SYNOPSIS10 11 % prove t/pmc/tqueue.t12 13 =head1 DESCRIPTION14 15 Tests the thread queue.16 17 =cut18 19 .sub main :main20 .include "include/test_more.pir"21 plan(5)22 thread_safe_queue_tests()23 .end24 25 .sub thread_safe_queue_tests26 .local int i, is_ok27 .local pmc tq, pInt28 29 new tq, ['TQueue']30 ok(1, "didn't crash")31 32 i = tq33 is_ok = i == 034 ok(is_ok, "int assignment gets # of elements in empty queue")35 36 pInt = new ['Integer']37 pInt = 238 push tq, pInt39 pInt = new ['Integer']40 pInt = 341 push tq, pInt42 i = tq43 is_ok = i == 244 ok(is_ok, "int assignment gets # of elements in non-empty queue")45 46 shift pInt, tq47 i = pInt48 is_ok = i == 249 ok(is_ok, "int retrieval works")50 shift pInt, tq51 i = pInt52 is_ok = i == 353 ok(is_ok, "int retrieval works")54 .end55 56 # Local Variables:57 # mode: pir58 # cperl-indent-level: 459 # fill-column: 10060 # End:61 # vim: expandtab shiftwidth=4 ft=pir: -
t/pmc/threads.t
46 46 } 47 47 } 48 48 if ( $platforms{$^O} ) { 49 plan tests => 20;49 plan tests => 15; 50 50 } 51 51 else { 52 52 plan skip_all => "No threading yet or test not enabled for '$^O'"; … … 288 288 500500 289 289 OUTPUT 290 290 291 292 pir_output_like( <<'CODE', <<'OUTPUT', "detach" );293 .sub main :main294 .local pmc foo295 .local pmc queue296 .local pmc thread297 foo = get_global '_foo'298 queue = new ['TQueue'] # flag for when the thread is done299 thread = new ['ParrotThread']300 thread.'run_clone'(foo, queue)301 302 thread.'detach'()303 wait:304 defined $I0, queue305 if $I0 == 0 goto wait306 print "done\n"307 .end308 309 .sub _foo310 .param pmc queue311 print "thread\n"312 sleep 0.1313 $P1 = new ['Integer']314 push queue, $P1315 .end316 CODE317 /(done\nthread\n)|(thread\ndone\n)/318 OUTPUT319 320 321 291 pir_output_is( <<'CODE', <<'OUTPUT', "share a PMC" ); 322 292 .sub main :main 323 293 .local pmc foo … … 354 324 21 355 325 OUTPUT 356 326 357 pir_output_is( <<'CODE', <<'OUT', "multi-threaded" );358 .sub main :main359 .local pmc queue360 queue = new ['TQueue']361 .local pmc tmpInt362 tmpInt = new ['Integer']363 tmpInt = 1364 push queue, tmpInt365 tmpInt = new ['Integer']366 tmpInt = 2367 push queue, tmpInt368 tmpInt = new ['Integer']369 tmpInt = 3370 push queue, tmpInt371 372 .local pmc thread373 thread = new ['ParrotThread']374 .local pmc foo375 foo = get_global '_foo'376 thread.'run_clone'(foo, queue)377 thread.'join'()378 print "done main\n"379 .end380 381 .sub _foo382 .param pmc queue383 $I0 = queue384 print $I0385 print "\n"386 loop:387 $I0 = queue388 if $I0 == 0 goto done389 shift $P0, queue390 print $P0391 print "\n"392 branch loop393 done:394 print "done thread\n"395 .end396 CODE397 3398 1399 2400 3401 done thread402 done main403 OUT404 405 327 pir_output_is( <<'CODE', <<'OUT', "sub name lookup in new thread" ); 406 328 .sub check 407 329 $P0 = get_global ['Foo'], 'foo' … … 1016 938 42 1017 939 OUTPUT 1018 940 1019 pir_output_is( <<'CODE', <<'OUT', 'multi-threaded strings via SharedRef' );1020 .sub main :main1021 .local pmc queue1022 .local pmc tmp_string1023 .local pmc shared_ref1024 1025 queue = new ['TQueue']1026 tmp_string = new ['String']1027 tmp_string = "ok 1\n"1028 shared_ref = new ['SharedRef'], tmp_string1029 push queue, shared_ref1030 tmp_string = new ['String']1031 tmp_string = "ok 2\n"1032 shared_ref = new ['SharedRef'], tmp_string1033 push queue, shared_ref1034 tmp_string = new ['String']1035 tmp_string = "ok 3\n"1036 shared_ref = new ['SharedRef'], tmp_string1037 push queue, shared_ref1038 1039 .local pmc thread1040 .local pmc foo1041 1042 thread = new ['ParrotThread']1043 foo = get_global '_foo'1044 thread.'run_clone'(foo, queue)1045 thread.'join'()1046 print "done main\n"1047 .end1048 1049 .sub _foo1050 .param pmc queue1051 $I0 = queue1052 print $I01053 print "\n"1054 loop:1055 $I0 = queue1056 if $I0 == 0 goto done1057 shift $P0, queue1058 print $P01059 branch loop1060 done:1061 print "done thread\n"1062 .end1063 CODE1064 31065 ok 11066 ok 21067 ok 31068 done thread1069 done main1070 OUT1071 1072 SKIP: {1073 skip( "no shared Strings yet", 2 );1074 pasm_output_is( <<'CODE', <<'OUT', "thread safe queue strings 1" );1075 new P10, ['TQueue']1076 print "ok 1\n"1077 set I0, P101078 print I01079 print "\n"1080 new P7, ['String']1081 set P7, "ok 2\n"1082 push P10, P71083 new P7, ['String']1084 set P7, "ok 3\n"1085 push P10, P71086 set I0, P101087 print I01088 print "\n"1089 1090 shift P8, P101091 print P81092 shift P8, P101093 print P81094 end1095 CODE1096 ok 11097 01098 21099 ok 21100 ok 31101 OUT1102 1103 pasm_output_is( <<'CODE', <<'OUT', "multi-threaded strings" );1104 new P10, ['TQueue']1105 new P7, ['String']1106 set P7, "ok 1\n"1107 push P10, P71108 new P7, ['String']1109 set P7, "ok 2\n"1110 push P10, P71111 new P7, ['String']1112 set P7, "ok 3\n"1113 push P10, P71114 set P6, P101115 1116 get_global P5, "_foo"1117 new P2, ['ParrotThread']1118 callmethod "thread3"1119 set I5, P21120 getinterp P21121 callmethod "join"1122 print "done main\n"1123 end1124 1125 .pcc_sub _foo:1126 set I0, P61127 print I01128 print "\n"1129 loop:1130 set I0, P61131 unless I0, ex1132 shift P8, P61133 print P81134 branch loop1135 ex:1136 print "done thread\n"1137 returncc1138 CODE1139 31140 ok 11141 ok 21142 ok 31143 done thread1144 done main1145 OUT1146 }1147 1148 941 # Local Variables: 1149 942 # mode: cperl 1150 943 # cperl-indent-level: 4 -
t/steps/auto_pmc-01.t
184 184 fixedstringarray.pmc 185 185 hash.pmc 186 186 orderedhash.pmc 187 tqueue.pmc188 187 os.pmc 189 188 file.pmc 190 189 addrregistry.pmc -
t/op/gc.t
6 6 use warnings; 7 7 use lib qw( . lib ../lib ../../lib ); 8 8 use Test::More; 9 use Parrot::Test tests => 20;9 use Parrot::Test tests => 19; 10 10 11 11 =head1 NAME 12 12 … … 474 474 ok 475 475 OUTPUT 476 476 477 pasm_output_is( <<'CODE', <<OUTPUT, "write barrier 4 - tqueue" );478 null I2479 set I3, 100480 lp3:481 null I0482 set I1, 10483 new P5, 'TQueue'484 new P0, 'Integer'485 needs_destroy P0486 # force partial sweep487 # P5 should now be black488 sweep 0489 # store white queue P1 in black P5 - needs a barrier490 new P1, 'TQueue'491 push P5, P1492 null P1493 new P0, 'Integer'494 needs_destroy P0495 # force sweep496 sweep 0497 shift P1, P5498 push P5, P1499 lp1:500 new P0, 'Integer'501 needs_destroy P0502 # force sweep503 sweep 0504 set P0, I0505 new P2, 'TQueue'506 push P2, P0507 push P1, P2508 new P3, 'Undef'509 new P4, 'Undef'510 inc I0511 lt I0, I1, lp1512 513 null I0514 shift P1, P5515 lp2:516 shift P2, P1517 shift P2, P2518 eq P2, I0, ok519 print "nok\n"520 print "I0: "521 print I0522 print " P2: "523 print P2524 print " type: "525 typeof S0, P2526 print S0527 print " I2: "528 print I2529 print "\n"530 exit 1531 ok:532 inc I0533 lt I0, I1, lp2534 inc I2535 lt I2, I3, lp3536 print "ok\n"537 end538 539 CODE540 ok541 OUTPUT542 543 477 pir_output_is( <<'CODE', <<'OUTPUT', "verify pmc proxy object marking" ); 544 478 .sub main :main 545 479 .local pmc cl, s, t -
examples/pir/thr-primes.pir
1 # Copyright (C) 2001-2008, Parrot Foundation.2 # $Id$3 4 =head1 NAME5 6 examples/pir/thr-primes.pir - Threads7 8 =head1 SYNOPSIS9 10 % ./parrot examples/pir/thr-primes.pir11 12 =head1 DESCRIPTION13 14 A threaded primes example.15 16 From C<perldoc perlthrtut>:17 18 1 #!/usr/bin/perl -w19 2 # prime-pthread, courtesy of Tom Christiansen20 321 4 use strict;22 523 6 use threads;24 7 use Thread::Queue;25 826 9 my $stream = new Thread::Queue;27 10 my $kid = new threads(\&check_num, $stream, 2);28 1129 12 for my $i ( 3 .. 1000 ) {30 13 $stream->enqueue($i);31 14 }32 1533 16 $stream->enqueue(undef);34 17 $kid->join;35 1836 19 sub check_num {37 20 my ($upstream, $cur_prime) = @_;38 21 my $kid;39 22 my $downstream = new Thread::Queue;40 23 while (my $num = $upstream->dequeue) {41 24 next unless $num % $cur_prime;42 25 if ($kid) {43 26 $downstream->enqueue($num);44 27 } else {45 28 print "Found prime $num\n";46 29 $kid = new threads(\&check_num, $downstream, $num);47 30 }48 31 }49 32 $downstream->enqueue(undef) if $kid;50 33 $kid->join if $kid;51 34 }52 53 =cut54 55 # translate to PIR by leo56 57 # Runs here (i386/linux 256MB mem) w.58 # ARENA_GC_FLAGS = 1 MAX=500 (~ 95 threads)59 # ARENA_GC_FLAGS = 0 MAX=1000 (~ 168 threads)60 61 62 .sub _main63 .param pmc argv64 .const int MAX = 50065 .local int max66 .local pmc kid67 .local pmc Check_num68 .local pmc stream69 .local int argc70 argc = argv71 max = MAX72 if argc < 2 goto no_arg73 $S0 = argv[1]74 max = $S075 no_arg:76 77 #sweepoff78 # 9 my $stream = new Thread::Queue;79 stream = new 'TQueue'80 # 10 my $kid = new threads(\&check_num, $stream, 2);81 Check_num = get_global "_check_num"82 kid = new 'ParrotThread'83 $P2 = new 'Integer'84 $P2 = 285 kid.'run_clone'(Check_num, Check_num, stream, $P2)86 87 # 12 for my $i ( 3 .. 1000 ) {88 .local int i89 i = 390 lp:91 # 13 $stream->enqueue($i);92 $P3 = new 'Integer'93 $P3 = i94 push stream, $P395 inc i96 if i <= max goto lp97 # 14 }98 99 # 16 $stream->enqueue(undef);100 $P4 = new 'Undef'101 push stream, $P4102 103 # 17 $kid->join;104 kid.'join'()105 .end106 107 # 19 sub check_num {108 # 20 my ($upstream, $cur_prime) = @_;109 # XXX still no comments inside pcc param block110 .sub _check_num111 .param pmc sub112 .param pmc upstream113 .param pmc cur_prime114 115 # 21 my $kid;116 .local pmc kid117 kid = new 'Undef'118 # 22 my $downstream = new Thread::Queue;119 .local pmc downstream120 downstream = new 'TQueue'121 # 23 while (my $num = $upstream->dequeue) {122 .local pmc Num # num is a reserved word123 lp:124 shift Num, upstream125 $I0 = defined Num126 unless $I0 goto ewhile127 # 24 next unless $num % $cur_prime;128 $P0 = new 'Integer'129 $P0 = Num % cur_prime130 unless $P0 goto lp131 # 25 if ($kid) {132 $I1 = defined kid133 unless $I1 goto no_kid1134 # 26 $downstream->enqueue($num);135 push downstream, Num136 goto lp137 # 27 } else {138 no_kid1:139 # 28 print "Found prime $num\n";140 print "Found prime "141 print Num142 print "\n"143 144 # 29 $kid = new threads(\&check_num, $downstream, $num);145 kid = new 'ParrotThread'146 kid.'run_clone'(sub, sub, downstream, Num)147 goto lp148 # 31 }149 ewhile:150 151 # 32 $downstream->enqueue(undef) if $kid;152 $I1 = defined kid153 unless $I1 goto no_kid2154 155 $P4 = new 'Undef'156 push downstream, $P4157 158 # 33 $kid->join if $kid;159 kid.'join'()160 161 no_kid2:162 # 34 }163 # sleep 1 # turn on for watching memory usage164 .end165 166 # Local Variables:167 # mode: pir168 # fill-column: 100169 # End:170 # vim: expandtab shiftwidth=4 ft=pir:
