src/scheduler.c

   1 /*
   2 Copyright (C) 2007-2009, Parrot Foundation.
   3 $Id$
   4
   5 =head1 NAME
   6
   7 src/scheduler.c - The core routines for the concurrency scheduler
   8
   9 =head1 DESCRIPTION
  10
  11 Each interpreter has a concurrency scheduler element in its core struct. The
  12 scheduler is responsible for receiveing, dispatching, and monitoring events,
  13 exceptions, async I/O, and concurrent tasks (threads).
  14
  15 =cut
  16
  17 */
  18
  19 #include "parrot/parrot.h"
  20 #include "parrot/scheduler_private.h"
  21
  22 #include "pmc/pmc_scheduler.h"
  23 #include "pmc/pmc_task.h"
  24 #include "pmc/pmc_timer.h"
  25 #include "pmc/pmc_context.h"
  26
  27 #include "scheduler.str"
  28
  29 #define CX_DEBUG 0
  30
  31 /* HEADERIZER HFILE: include/parrot/scheduler.h */
  32
  33 /* HEADERIZER BEGIN: static */
  34 /* Don't modify between HEADERIZER BEGIN / HEADERIZER END.  Your changes will be lost. */
  35
  36 static void scheduler_process_messages(PARROT_INTERP,
  37     ARGMOD(PMC *scheduler))
  38         __attribute__nonnull__(1)
  39         __attribute__nonnull__(2)
  40         FUNC_MODIFIES(*scheduler);
  41
  42 static void scheduler_process_wait_list(PARROT_INTERP,
  43     ARGMOD(PMC *scheduler))
  44         __attribute__nonnull__(1)
  45         __attribute__nonnull__(2)
  46         FUNC_MODIFIES(*scheduler);
  47
  48 #define ASSERT_ARGS_scheduler_process_messages __attribute__unused__ int _ASSERT_ARGS_CHECK = (\
  49        PARROT_ASSERT_ARG(interp) \
  50     , PARROT_ASSERT_ARG(scheduler))
  51 #define ASSERT_ARGS_scheduler_process_wait_list __attribute__unused__ int _ASSERT_ARGS_CHECK = (\
  52        PARROT_ASSERT_ARG(interp) \
  53     , PARROT_ASSERT_ARG(scheduler))
  54 /* Don't modify between HEADERIZER BEGIN / HEADERIZER END.  Your changes will be lost. */
  55 /* HEADERIZER END: static */
  56
  57 /*
  58
  59 =head2 Scheduler Interface Functions
  60
  61 Functions to interface with the concurrency scheduler.
  62
  63 =over 4
  64
  65 =item C<void Parrot_cx_init_scheduler(PARROT_INTERP)>
  66
  67 Initalize the concurrency scheduler for the interpreter.
  68
  69 =cut
  70
  71 */
  72
  73 void
  74 Parrot_cx_init_scheduler(PARROT_INTERP)
  75 {
  76     ASSERT_ARGS(Parrot_cx_init_scheduler)
  77     if (!interp->parent_interpreter) {
  78         PMC *scheduler;
  79
  80         /* Add the very first interpreter to the list of interps. */
  81         pt_add_to_interpreters(interp, NULL);
  82
  83         scheduler = pmc_new(interp, enum_class_Scheduler);
  84         scheduler = VTABLE_share_ro(interp, scheduler);
  85
  86         interp->scheduler = scheduler;
  87     }
  88 }
  89
  90 /*
  91
  92 =item C<void Parrot_cx_check_tasks(PARROT_INTERP, PMC *scheduler)>
  93
  94 If a wake request has been received, handle tasks.
  95
  96 =cut
  97
  98 */
  99
 100 void
 101 Parrot_cx_check_tasks(PARROT_INTERP, ARGMOD(PMC *scheduler))
 102 {
 103     ASSERT_ARGS(Parrot_cx_check_tasks)
 104     if (SCHEDULER_wake_requested_TEST(scheduler))
 105         Parrot_cx_handle_tasks(interp, interp->scheduler);
 106 }
 107
 108 /*
 109
 110 =item C<void Parrot_cx_handle_tasks(PARROT_INTERP, PMC *scheduler)>
 111
 112 Handle the pending tasks in the scheduler's task list. Returns when there are
 113 no more pending tasks. Returns 0 to terminate the scheduler runloop, or 1 to
 114 continue the runloop.
 115
 116 =cut
 117
 118 */
 119
 120 PARROT_EXPORT
 121 void
 122 Parrot_cx_handle_tasks(PARROT_INTERP, ARGMOD(PMC *scheduler))
 123 {
 124     ASSERT_ARGS(Parrot_cx_handle_tasks)
 125     SCHEDULER_wake_requested_CLEAR(scheduler);
 126     Parrot_cx_refresh_task_list(interp, scheduler);
 127
 128     while (VTABLE_get_integer(interp, scheduler) > 0) {
 129         PMC * const task = VTABLE_pop_pmc(interp, scheduler);
 130         if (!PMC_IS_NULL(task)) {
 131             PMC    * const type_pmc = VTABLE_get_attr_str(interp, task, CONST_STRING(interp, "type"));
 132             STRING * const type     = VTABLE_get_string(interp, type_pmc);
 133
 134             if (Parrot_str_equal(interp, type, CONST_STRING(interp, "callback"))) {
 135                 Parrot_cx_invoke_callback(interp, task);
 136             }
 137             else if (Parrot_str_equal(interp, type, CONST_STRING(interp, "timer"))) {
 138                 Parrot_cx_timer_invoke(interp, task);
 139             }
 140             else if (Parrot_str_equal(interp, type, CONST_STRING(interp, "event"))) {
 141                 PMC * const handler = Parrot_cx_find_handler_for_task(interp, task);
 142                 if (!PMC_IS_NULL(handler)) {
 143                     PMC * const handler_sub = VTABLE_get_attr_str(interp, handler, CONST_STRING(interp, "code"));
 144                     Parrot_pcc_invoke_sub_from_c_args(interp, handler_sub,
 145                             "PP->", handler, task);
 146                 }
 147             }
 148             else {
 149                 Parrot_ex_throw_from_c_args(interp, NULL, EXCEPTION_INVALID_OPERATION,
 150                         "Unknown task type '%Ss'.\n", type);
 151             }
 152
 153             Parrot_cx_delete_task(interp, task);
 154         }
 155
 156         /* If the scheduler was flagged to terminate, make sure you process all
 157          * tasks. */
 158         if (SCHEDULER_terminate_requested_TEST(scheduler))
 159             Parrot_cx_refresh_task_list(interp, scheduler);
 160
 161     } /* end of pending tasks */
 162 }
 163
 164 /*
 165
 166 =item C<void Parrot_cx_refresh_task_list(PARROT_INTERP, PMC *scheduler)>
 167
 168 Tell the scheduler to perform maintenance on its list of active tasks, checking
 169 for completed timers or sleep events, sorting for priority, checking for
 170 messages, etc.
 171
 172 =cut
 173
 174 */
 175
 176 void
 177 Parrot_cx_refresh_task_list(PARROT_INTERP, ARGMOD(PMC *scheduler))
 178 {
 179     ASSERT_ARGS(Parrot_cx_refresh_task_list)
 180     scheduler_process_wait_list(interp, scheduler);
 181     scheduler_process_messages(interp, scheduler);
 182
 183     /* TODO: Sort the task list index */
 184
 185     SCHEDULER_cache_valid_SET(scheduler);
 186     return;
 187 }
 188
 189 /*
 190
 191 =item C<void Parrot_cx_runloop_wake(PARROT_INTERP, PMC *scheduler)>
 192
 193 Wake a sleeping scheduler runloop (generally called when new tasks are added to
 194 the scheduler's task list).
 195
 196 =cut
 197
 198 */
 199
 200 void
 201 Parrot_cx_runloop_wake(PARROT_INTERP, ARGMOD(PMC *scheduler))
 202 {
 203     ASSERT_ARGS(Parrot_cx_runloop_wake)
 204     enable_event_checking(interp);
 205     SCHEDULER_wake_requested_SET(scheduler);
 206 }
 207
 208
 209 /*
 210
 211 =item C<void Parrot_cx_runloop_end(PARROT_INTERP)>
 212
 213 Schedule an event to terminate the scheduler runloop.
 214
 215 =cut
 216
 217 */
 218
 219 PARROT_EXPORT
 220 void
 221 Parrot_cx_runloop_end(PARROT_INTERP)
 222 {
 223     ASSERT_ARGS(Parrot_cx_runloop_end)
 224     SCHEDULER_terminate_requested_SET(interp->scheduler);
 225     Parrot_cx_handle_tasks(interp, interp->scheduler);
 226 }
 227
 228 /*
 229
 230 =item C<void Parrot_cx_schedule_task(PARROT_INTERP, PMC *task)>
 231
 232 Add a task to scheduler's task list. Cannot be called across
 233 interpreters/threads, must be called from within the interpreter's runloop.
 234
 235 =cut
 236
 237 */
 238
 239 PARROT_EXPORT
 240 void
 241 Parrot_cx_schedule_task(PARROT_INTERP, ARGIN(PMC *task))
 242 {
 243     ASSERT_ARGS(Parrot_cx_schedule_task)
 244     if (!interp->scheduler)
 245         Parrot_ex_throw_from_c_args(interp, NULL, EXCEPTION_INVALID_OPERATION,
 246             "Scheduler was not initialized for this interpreter.\n");
 247
 248     VTABLE_push_pmc(interp, interp->scheduler, task);
 249 }
 250
 251 /*
 252
 253 =item C<PMC * Parrot_cx_peek_task(PARROT_INTERP)>
 254
 255 Retrieve the the top task on the scheduler's task list, but don't remove it
 256 from the list.
 257
 258 =cut
 259
 260 */
 261
 262 PARROT_EXPORT
 263 PARROT_CAN_RETURN_NULL
 264 PMC *
 265 Parrot_cx_peek_task(PARROT_INTERP)
 266 {
 267     ASSERT_ARGS(Parrot_cx_peek_task)
 268     if (!interp->scheduler)
 269         Parrot_ex_throw_from_c_args(interp, NULL, EXCEPTION_INVALID_OPERATION,
 270             "Scheduler was not initialized for this interpreter.\n");
 271
 272     return VTABLE_pop_pmc(interp, interp->scheduler);
 273 }
 274
 275 /*
 276
 277 =item C<void Parrot_cx_schedule_timer(PARROT_INTERP, STRING *type, FLOATVAL
 278 duration, FLOATVAL interval, INTVAL repeat, PMC *sub)>
 279
 280 Create a new timer event due at C<diff> from now, repeated at C<interval>
 281 and running the passed C<sub>.
 282
 283 =cut
 284
 285 */
 286
 287 PARROT_EXPORT
 288 void
 289 Parrot_cx_schedule_timer(PARROT_INTERP,
 290         ARGIN_NULLOK(STRING *type), FLOATVAL duration, FLOATVAL interval,
 291         INTVAL repeat, ARGIN_NULLOK(PMC *sub))
 292 {
 293     ASSERT_ARGS(Parrot_cx_schedule_timer)
 294     PMC * const timer = pmc_new(interp, enum_class_Timer);
 295
 296     VTABLE_set_number_keyed_int(interp, timer, PARROT_TIMER_NSEC, duration);
 297     VTABLE_set_number_keyed_int(interp, timer, PARROT_TIMER_INTERVAL, interval);
 298     VTABLE_set_integer_keyed_int(interp, timer, PARROT_TIMER_REPEAT, repeat);
 299
 300     if (!PMC_IS_NULL(sub))
 301         VTABLE_set_pmc_keyed_int(interp, timer, PARROT_TIMER_HANDLER, sub);
 302
 303     if (!STRING_IS_NULL(type))
 304         VTABLE_set_string_native(interp, timer, type);
 305
 306     if (repeat && FLOAT_IS_ZERO(interval))
 307         VTABLE_set_number_keyed_int(interp, timer, PARROT_TIMER_INTERVAL, duration);
 308
 309     Parrot_cx_schedule_task(interp, timer);
 310 }
 311
 312 /*
 313
 314 =item C<void Parrot_cx_schedule_repeat(PARROT_INTERP, PMC *task)>
 315
 316 Add a repeat task to scheduler's task list.
 317
 318 =cut
 319
 320 */
 321
 322 PARROT_EXPORT
 323 void
 324 Parrot_cx_schedule_repeat(PARROT_INTERP, ARGIN(PMC *task))
 325 {
 326     ASSERT_ARGS(Parrot_cx_schedule_repeat)
 327     INTVAL repeat = VTABLE_get_integer_keyed_int(interp, task,
 328             PARROT_TIMER_REPEAT);
 329     FLOATVAL duration = VTABLE_get_number_keyed_int(interp, task,
 330             PARROT_TIMER_INTERVAL);
 331     if (repeat != 0) {
 332         PMC *repeat_task = VTABLE_clone(interp, task);
 333         VTABLE_set_number_keyed_int(interp, repeat_task, PARROT_TIMER_NSEC, duration);
 334
 335         if (repeat > 0)
 336             VTABLE_set_integer_keyed_int(interp, repeat_task,
 337                 PARROT_TIMER_REPEAT, repeat - 1);
 338
 339         Parrot_cx_schedule_task(interp, repeat_task);
 340     }
 341 }
 342
 343 /*
 344
 345 =item C<void Parrot_cx_schedule_callback(PARROT_INTERP, PMC *user_data, char
 346 *ext_data)>
 347
 348 Create a new callback event, with an argument for the call.
 349
 350 =cut
 351
 352 */
 353
 354 PARROT_EXPORT
 355 void
 356 Parrot_cx_schedule_callback(PARROT_INTERP,
 357         ARGIN(PMC *user_data), ARGIN(char *ext_data))
 358 {
 359     ASSERT_ARGS(Parrot_cx_schedule_callback)
 360     PMC *callback = pmc_new(interp, enum_class_Task);
 361     Parrot_Task_attributes * const task_struct = PARROT_TASK(callback);
 362
 363     task_struct->type    = CONST_STRING(interp, "callback");
 364     task_struct->data    = user_data;
 365     task_struct->cb_data = ext_data;
 366
 367     Parrot_cx_schedule_task(interp, callback);
 368 }
 369
 370 /*
 371
 372 =item C<void Parrot_cx_request_suspend_for_gc(PARROT_INTERP)>
 373
 374 Tell the scheduler to suspend for GC at the next safe pause.
 375
 376 =cut
 377
 378 */
 379
 380 PARROT_EXPORT
 381 void
 382 Parrot_cx_request_suspend_for_gc(PARROT_INTERP)
 383 {
 384     ASSERT_ARGS(Parrot_cx_request_suspend_for_gc)
 385 #if CX_DEBUG
 386     fprintf(stderr, "requesting gc suspend [interp=%p]\n", interp);
 387 #endif
 388     Parrot_cx_send_message(interp, CONST_STRING(interp, "suspend_for_gc"), PMCNULL);
 389 }
 390
 391 /*
 392
 393 =item C<void Parrot_cx_delete_task(PARROT_INTERP, PMC *task)>
 394
 395 Remove a task from the scheduler's task list.
 396
 397 =cut
 398
 399 */
 400
 401 PARROT_EXPORT
 402 void
 403 Parrot_cx_delete_task(PARROT_INTERP, ARGIN(PMC *task))
 404 {
 405     ASSERT_ARGS(Parrot_cx_delete_task)
 406     if (interp->scheduler && !PObj_on_free_list_TEST(interp->scheduler)) {
 407         const INTVAL tid = VTABLE_get_integer(interp, task);
 408         VTABLE_delete_keyed_int(interp, interp->scheduler, tid);
 409     }
 410     else if (interp->scheduler)
 411         Parrot_ex_throw_from_c_args(interp, NULL, EXCEPTION_INVALID_OPERATION,
 412             "Scheduler was not initialized for this interpreter.\n");
 413 }
 414
 415
 416 /*
 417
 418 =item C<PMC * Parrot_cx_delete_suspend_for_gc(PARROT_INTERP)>
 419
 420 Remove a message that would suspend GC from the message queue. (Provided for
 421 backward compatibility in the threads implementation.)
 422
 423 =cut
 424
 425 */
 426
 427 PARROT_EXPORT
 428 PARROT_CAN_RETURN_NULL
 429 PMC *
 430 Parrot_cx_delete_suspend_for_gc(PARROT_INTERP)
 431 {
 432     ASSERT_ARGS(Parrot_cx_delete_suspend_for_gc)
 433     if (interp->scheduler) {
 434         STRING *suspend_str = CONST_STRING(interp, "suspend_for_gc");
 435         Parrot_Scheduler_attributes * sched_struct = PARROT_SCHEDULER(interp->scheduler);
 436         INTVAL num_tasks, index;
 437
 438 #if CX_DEBUG
 439     fprintf(stderr, "called delete_suspend_for_gc\n");
 440 #endif
 441
 442 #if CX_DEBUG
 443     fprintf(stderr, "locking msg_lock (delete) [interp=%p]\n", interp);
 444 #endif
 445         LOCK(sched_struct->msg_lock);
 446         /* Search the task index for GC suspend tasks */
 447         num_tasks = VTABLE_elements(interp, sched_struct->messages);
 448         for (index = 0; index < num_tasks; index++) {
 449             PMC *message = VTABLE_get_pmc_keyed_int(interp, sched_struct->messages, index);
 450             if (!PMC_IS_NULL(message)
 451             &&   Parrot_str_equal(interp, VTABLE_get_string(interp, message),
 452                         suspend_str)) {
 453                 VTABLE_delete_keyed_int(interp, sched_struct->messages, index);
 454                 UNLOCK(sched_struct->msg_lock);
 455                 return message;
 456             }
 457         }
 458 #if CX_DEBUG
 459     fprintf(stderr, "unlocking msg_lock (delete) [interp=%p]\n", interp);
 460 #endif
 461         UNLOCK(sched_struct->msg_lock);
 462
 463     }
 464     else
 465         Parrot_ex_throw_from_c_args(interp, NULL, EXCEPTION_INVALID_OPERATION,
 466             "Scheduler was not initialized for this interpreter.\n");
 467
 468     return PMCNULL;
 469 }
 470
 471 /*
 472
 473 =item C<void Parrot_cx_add_handler_local(PARROT_INTERP, PMC *handler)>
 474
 475 Add a handler to the current context's list of handlers.
 476
 477 =cut
 478
 479 */
 480
 481 PARROT_EXPORT
 482 void
 483 Parrot_cx_add_handler_local(PARROT_INTERP, ARGIN(PMC *handler))
 484 {
 485     ASSERT_ARGS(Parrot_cx_add_handler_local)
 486     if (PMC_IS_NULL(Parrot_pcc_get_handlers(interp, interp->ctx)))
 487         Parrot_pcc_set_handlers(interp, interp->ctx, pmc_new(interp, enum_class_ResizablePMCArray));
 488
 489     VTABLE_unshift_pmc(interp, Parrot_pcc_get_handlers(interp, interp->ctx), handler);
 490
 491 }
 492
 493 /*
 494
 495 =item C<void Parrot_cx_delete_handler_local(PARROT_INTERP, STRING
 496 *handler_type)>
 497
 498 Remove the top task handler of a particular type from the context's list of
 499 handlers.
 500
 501 =cut
 502
 503 */
 504
 505 PARROT_EXPORT
 506 void
 507 Parrot_cx_delete_handler_local(PARROT_INTERP, ARGIN(STRING *handler_type))
 508 {
 509     ASSERT_ARGS(Parrot_cx_delete_handler_local)
 510     PMC *handlers  = Parrot_pcc_get_handlers(interp, interp->ctx);
 511
 512     if (PMC_IS_NULL(handlers))
 513         Parrot_ex_throw_from_c_args(interp, NULL, EXCEPTION_INVALID_OPERATION,
 514             "No handler to delete.");
 515
 516     if (STRING_IS_NULL(handler_type) || STRING_IS_EMPTY(handler_type))
 517         VTABLE_shift_pmc(interp, handlers);
 518     else {
 519         /* Loop from newest handler to oldest handler. */
 520         STRING      *exception_str = CONST_STRING(interp, "exception");
 521         STRING      *event_str     = CONST_STRING(interp, "event");
 522         STRING      *handler_str   = CONST_STRING(interp, "ExceptionHandler");
 523         const INTVAL elements      = VTABLE_elements(interp, handlers);
 524         INTVAL       index;
 525         typedef enum { Hunknown,  Hexception, Hevent } Htype;
 526
 527         const Htype htype =
 528             Parrot_str_equal(interp, handler_type, exception_str) ?
 529             Hexception :
 530             Parrot_str_equal(interp, handler_type, event_str) ?
 531                 Hevent :
 532                 Hunknown;
 533         STRING * const handler_name = (htype == Hexception) ?
 534             handler_str : (STRING *) NULL;
 535
 536         for (index = 0; index < elements; ++index) {
 537             PMC *handler = VTABLE_get_pmc_keyed_int(interp, handlers, index);
 538             if (!PMC_IS_NULL(handler)) {
 539                 switch (htype) {
 540                     case Hexception:
 541                         if (VTABLE_isa(interp, handler, handler_name)) {
 542                             VTABLE_set_pmc_keyed_int(interp, handlers, index, PMCNULL);
 543                             return;
 544                         }
 545                         break;
 546                     case Hevent:
 547                         if (handler->vtable->base_type == enum_class_EventHandler) {
 548                             VTABLE_set_pmc_keyed_int(interp, handlers, index, PMCNULL);
 549                             return;
 550                         }
 551                         break;
 552                     default:
 553                         break;
 554                 }
 555             }
 556         }
 557
 558         Parrot_ex_throw_from_c_args(interp, NULL,
 559             EXCEPTION_INVALID_OPERATION, "No handler to delete.");
 560     }
 561 }
 562
 563
 564 /*
 565
 566 =item C<INTVAL Parrot_cx_count_handlers_local(PARROT_INTERP, STRING
 567 *handler_type)>
 568
 569 Count the number of active handlers of a particular type from the
 570 context's list of handlers.
 571
 572 =cut
 573
 574 */
 575
 576 PARROT_EXPORT
 577 INTVAL
 578 Parrot_cx_count_handlers_local(PARROT_INTERP, ARGIN(STRING *handler_type))
 579 {
 580     ASSERT_ARGS(Parrot_cx_count_handlers_local)
 581     PMC *handlers = Parrot_pcc_get_handlers(interp, interp->ctx);
 582     INTVAL elements;
 583
 584     if (PMC_IS_NULL(handlers))
 585         return 0;
 586
 587     elements = VTABLE_elements(interp, handlers);
 588
 589     if (STRING_IS_NULL(handler_type) || STRING_IS_EMPTY(handler_type))
 590         return elements;
 591
 592     /* Loop from newest handler to oldest handler. */
 593     {
 594         STRING      *exception_str = CONST_STRING(interp, "exception");
 595         STRING      *event_str     = CONST_STRING(interp, "event");
 596         STRING      *handler_str   = CONST_STRING(interp, "ExceptionHandler");
 597         INTVAL       count = 0;
 598         INTVAL       index;
 599         typedef enum { Hunknown,  Hexception, Hevent } Htype;
 600
 601         const Htype htype =
 602             (Parrot_str_equal(interp, handler_type, exception_str)) ?
 603             Hexception :
 604             (Parrot_str_equal(interp, handler_type, event_str)) ?
 605                 Hevent :
 606                 Hunknown;
 607         STRING * const handler_name = (htype == Hexception) ?
 608             handler_str : (STRING *) NULL;
 609
 610         for (index = 0; index < elements; ++index) {
 611             PMC *handler = VTABLE_get_pmc_keyed_int(interp, handlers, index);
 612             if (!PMC_IS_NULL(handler)) {
 613                 switch (htype) {
 614                 case Hexception:
 615                         if (VTABLE_isa(interp, handler, handler_name))
 616                             count++;
 617                         break;
 618                     case Hevent:
 619                         if (handler->vtable->base_type == enum_class_EventHandler)
 620                             count++;
 621                         break;
 622                     default:
 623                         break;
 624                 }
 625             }
 626         }
 627         return count;
 628     }
 629 }
 630
 631
 632 /*
 633
 634 =item C<void Parrot_cx_add_handler(PARROT_INTERP, PMC *handler)>
 635
 636 Add a task handler to scheduler's list of handlers.
 637
 638 =cut
 639
 640 */
 641
 642 PARROT_EXPORT
 643 void
 644 Parrot_cx_add_handler(PARROT_INTERP, ARGIN(PMC *handler))
 645 {
 646     ASSERT_ARGS(Parrot_cx_add_handler)
 647     STRING *add_handler = CONST_STRING(interp, "add_handler");
 648     if (!interp->scheduler)
 649         Parrot_ex_throw_from_c_args(interp, NULL, EXCEPTION_INVALID_OPERATION,
 650             "Scheduler was not initialized for this interpreter.\n");
 651
 652     Parrot_pcc_invoke_method_from_c_args(interp, interp->scheduler, add_handler, "P->", handler);
 653 }
 654
 655 /*
 656
 657 =item C<void Parrot_cx_delete_handler_typed(PARROT_INTERP, STRING
 658 *handler_type)>
 659
 660 Remove the top task handler of a particular type from the scheduler's list of
 661 handlers.
 662
 663 =cut
 664
 665 */
 666
 667 PARROT_EXPORT
 668 void
 669 Parrot_cx_delete_handler_typed(PARROT_INTERP, ARGIN(STRING *handler_type))
 670 {
 671     ASSERT_ARGS(Parrot_cx_delete_handler_typed)
 672     if (!interp->scheduler)
 673         Parrot_ex_throw_from_c_args(interp, NULL, EXCEPTION_INVALID_OPERATION,
 674             "Scheduler was not initialized for this interpreter.\n");
 675
 676     Parrot_pcc_invoke_method_from_c_args(interp, interp->scheduler, CONST_STRING(interp, "delete_handler"), "S->", handler_type);
 677 }
 678
 679 /*
 680
 681 =item C<INTVAL Parrot_cx_count_handlers_typed(PARROT_INTERP, STRING
 682 *handler_type)>
 683
 684 Count the number of active handlers of a particular type (event, exception) in
 685 the concurrency scheduler.
 686
 687 =cut
 688
 689 */
 690
 691 PARROT_EXPORT
 692 INTVAL
 693 Parrot_cx_count_handlers_typed(PARROT_INTERP, ARGIN(STRING *handler_type))
 694 {
 695     ASSERT_ARGS(Parrot_cx_count_handlers_typed)
 696     INTVAL count = 0;
 697
 698     if (!interp->scheduler)
 699         Parrot_ex_throw_from_c_args(interp, NULL, EXCEPTION_INVALID_OPERATION,
 700             "Scheduler was not initialized for this interpreter.\n");
 701
 702     Parrot_pcc_invoke_method_from_c_args(interp, interp->scheduler, CONST_STRING(interp, "count_handlers"), "S->I", handler_type, &count);
 703
 704     return count;
 705 }
 706
 707 /*
 708
 709 =back
 710
 711 =head2 Scheduler Message Interface Functions
 712
 713 Functions that are used to interface with the message queue in the concurrency
 714 scheduler.
 715
 716 =over 4
 717
 718 =item C<void Parrot_cx_send_message(PARROT_INTERP, STRING *messagetype, PMC
 719 *payload)>
 720
 721 Send a message to a scheduler in a different interpreter/thread.
 722
 723 =cut
 724
 725 */
 726
 727 PARROT_EXPORT
 728 void
 729 Parrot_cx_send_message(PARROT_INTERP, ARGIN(STRING *messagetype), SHIM(PMC *payload))
 730 {
 731     ASSERT_ARGS(Parrot_cx_send_message)
 732     if (interp->scheduler) {
 733         Parrot_Scheduler_attributes * sched_struct = PARROT_SCHEDULER(interp->scheduler);
 734         PMC *message = pmc_new(interp, enum_class_SchedulerMessage);
 735         VTABLE_set_string_native(interp, message, messagetype);
 736         message = VTABLE_share_ro(interp, message);
 737
 738 #if CX_DEBUG
 739     fprintf(stderr, "sending message[interp=%p]\n", interp);
 740 #endif
 741
 742 #if CX_DEBUG
 743     fprintf(stderr, "locking msg_lock (send) [interp=%p]\n", interp);
 744 #endif
 745         LOCK(sched_struct->msg_lock);
 746         VTABLE_push_pmc(interp, sched_struct->messages, message);
 747 #if CX_DEBUG
 748     fprintf(stderr, "unlocking msg_lock (send) [interp=%p]\n", interp);
 749 #endif
 750         UNLOCK(sched_struct->msg_lock);
 751         Parrot_cx_runloop_wake(interp, interp->scheduler);
 752
 753     }
 754
 755 }
 756
 757 /*
 758
 759 =item C<void Parrot_cx_broadcast_message(PARROT_INTERP, STRING *messagetype, PMC
 760 *data)>
 761
 762 Send a message to the schedulers in all interpreters/threads linked to this
 763 one.
 764
 765 =cut
 766
 767 */
 768
 769 PARROT_EXPORT
 770 void
 771 Parrot_cx_broadcast_message(PARROT_INTERP, ARGIN(STRING *messagetype), ARGIN_NULLOK(PMC *data))
 772 {
 773     ASSERT_ARGS(Parrot_cx_broadcast_message)
 774     UINTVAL i;
 775     LOCK(interpreter_array_mutex);
 776     for (i = 0; i < n_interpreters; ++i) {
 777         Parrot_Interp other_interp = interpreter_array[i];
 778         if (interp == other_interp)
 779             continue;
 780         Parrot_cx_send_message(other_interp, messagetype, data);
 781     }
 782     UNLOCK(interpreter_array_mutex);
 783
 784 }
 785
 786 /*
 787
 788 =back
 789
 790 =head2 Task Interface Functions
 791
 792 Functions that are used to interface with a specific task in the concurrency scheduler.
 793
 794 =over 4
 795
 796 =item C<PMC * Parrot_cx_find_handler_for_task(PARROT_INTERP, PMC *task)>
 797
 798 Retrieve a handler appropriate to a given task. If the scheduler has no
 799 appropriate handler, returns PMCNULL.
 800
 801 =cut
 802
 803 */
 804
 805 PARROT_EXPORT
 806 PARROT_CAN_RETURN_NULL
 807 PMC *
 808 Parrot_cx_find_handler_for_task(PARROT_INTERP, ARGIN(PMC *task))
 809 {
 810     ASSERT_ARGS(Parrot_cx_find_handler_for_task)
 811     PMC *handler = PMCNULL;
 812 #if CX_DEBUG
 813     fprintf(stderr, "searching for handler\n");
 814 #endif
 815
 816     if (!interp->scheduler)
 817         Parrot_ex_throw_from_c_args(interp, NULL, EXCEPTION_INVALID_OPERATION,
 818             "Scheduler was not initialized for this interpreter.\n");
 819
 820     Parrot_pcc_invoke_method_from_c_args(interp, interp->scheduler, CONST_STRING(interp, "find_handler"), "P->P", task, &handler);
 821
 822 #if CX_DEBUG
 823     fprintf(stderr, "done searching for handler\n");
 824 #endif
 825
 826     return handler;
 827 }
 828
 829 /*
 830
 831 =item C<PMC * Parrot_cx_find_handler_local(PARROT_INTERP, PMC *task)>
 832
 833 Retrieve a handler appropriate to a given task from the local context. If the
 834 context has no appropriate handler, returns PMCNULL.
 835
 836 =cut
 837
 838 */
 839
 840 PARROT_EXPORT
 841 PARROT_CAN_RETURN_NULL
 842 PMC *
 843 Parrot_cx_find_handler_local(PARROT_INTERP, ARGIN(PMC *task))
 844 {
 845     ASSERT_ARGS(Parrot_cx_find_handler_local)
 846
 847     /*
 848      * Quick&dirty way to avoid infinite recursion
 849      * when an exception is thrown while looking
 850      * for a handler
 851      */
 852     static int already_doing = 0;
 853     static PMC * keep_context = NULL;
 854
 855     PMC            *context;
 856     PMC            *iter        = PMCNULL;
 857     STRING * const  handled_str = CONST_STRING(interp, "handled");
 858     STRING * const  iter_str    = CONST_STRING(interp, "handler_iter");
 859
 860     if (already_doing) {
 861         Parrot_io_eprintf(interp,
 862             "** Exception caught while looking for a handler, trying next **\n");
 863         if (! keep_context)
 864             return NULL;
 865         /*
 866          * Note that we are now trying to handle the new exception,
 867          * not the initial task argument (exception or whatever).
 868          */
 869         context = Parrot_pcc_get_caller_ctx(interp, keep_context);
 870         keep_context = NULL;
 871         if (context && !PMC_IS_NULL(Parrot_pcc_get_handlers(interp, context)))
 872             iter = VTABLE_get_iter(interp, Parrot_pcc_get_handlers(interp, context));
 873         else
 874             iter = PMCNULL;
 875     }
 876     else {
 877         ++already_doing;
 878
 879         /* Exceptions store the handler iterator for rethrow, other kinds of
 880          * tasks don't (though they could). */
 881         if (task->vtable->base_type == enum_class_Exception
 882         && VTABLE_get_integer_keyed_str(interp, task, handled_str) == -1) {
 883             iter    = VTABLE_get_attr_str(interp, task, iter_str);
 884             context = (PMC *)VTABLE_get_pointer(interp, task);
 885         }
 886         else {
 887             context = CURRENT_CONTEXT(interp);
 888             if (!PMC_IS_NULL(Parrot_pcc_get_handlers(interp, context)))
 889                 iter = VTABLE_get_iter(interp, Parrot_pcc_get_handlers(interp, context));
 890         }
 891     }
 892
 893     while (context) {
 894         keep_context = context;
 895         /* Loop from newest handler to oldest handler. */
 896         while (!PMC_IS_NULL(iter) && VTABLE_get_bool(interp, iter)) {
 897             PMC *handler = VTABLE_shift_pmc(interp, iter);
 898
 899             if (!PMC_IS_NULL(handler)) {
 900                 INTVAL valid_handler = 0;
 901                 if (handler->vtable->base_type == enum_class_Object)
 902                     Parrot_pcc_invoke_method_from_c_args(interp, handler, CONST_STRING(interp, "can_handle"),
 903                         "P->I", task, &valid_handler);
 904                 else
 905                     Parrot_pcc_invoke_method_from_c_args(interp, handler, CONST_STRING(interp, "can_handle"),
 906                         "P->I", task, &valid_handler);
 907
 908                 if (valid_handler) {
 909                     if (task->vtable->base_type == enum_class_Exception) {
 910                         /* Store iterator and context for a later rethrow. */
 911                         VTABLE_set_attr_str(interp, task, CONST_STRING(interp, "handler_iter"), iter);
 912                         VTABLE_set_pointer(interp, task, context);
 913                     }
 914                     --already_doing;
 915                     keep_context = NULL;
 916                     return handler;
 917                 }
 918             }
 919         }
 920
 921         /* Continue the search in the next context up the chain. */
 922         context = Parrot_pcc_get_caller_ctx(interp, context);
 923         if (context && !PMC_IS_NULL(Parrot_pcc_get_handlers(interp, context)))
 924             iter = VTABLE_get_iter(interp, Parrot_pcc_get_handlers(interp, context));
 925         else
 926             iter = PMCNULL;
 927     }
 928
 929     /* Reached the end of the context chain without finding a handler. */
 930
 931     --already_doing;
 932     return PMCNULL;
 933 }
 934
 935 /*
 936
 937 =item C<void Parrot_cx_timer_invoke(PARROT_INTERP, PMC *timer)>
 938
 939 Run the associated code block for a timer event, when the timer fires.
 940
 941 =cut
 942
 943 */
 944
 945 void
 946 Parrot_cx_timer_invoke(PARROT_INTERP, ARGIN(PMC *timer))
 947 {
 948     ASSERT_ARGS(Parrot_cx_timer_invoke)
 949     Parrot_Timer_attributes * const timer_struct = PARROT_TIMER(timer);
 950 #if CX_DEBUG
 951     fprintf(stderr, "current timer time: %f, %f\n",
 952                     timer_struct->birthtime + timer_struct->duration,
 953                     Parrot_floatval_time());
 954 #endif
 955     if (!PMC_IS_NULL(timer_struct->codeblock)) {
 956         Parrot_pcc_invoke_sub_from_c_args(interp,
 957                 timer_struct->codeblock, "->");
 958     }
 959 }
 960
 961 /*
 962
 963 =item C<void Parrot_cx_invoke_callback(PARROT_INTERP, PMC *callback)>
 964
 965 Run the associated code block for a callback event.
 966
 967 =cut
 968
 969 */
 970
 971 void
 972 Parrot_cx_invoke_callback(PARROT_INTERP, ARGIN(PMC *callback))
 973 {
 974     ASSERT_ARGS(Parrot_cx_invoke_callback)
 975     Parrot_Task_attributes * const task_struct = PARROT_TASK(callback);
 976     if (!PMC_IS_NULL(task_struct->data)) {
 977         Parrot_run_callback(interp, task_struct->data,
 978                 task_struct->cb_data);
 979     }
 980 }
 981
 982 /*
 983
 984 =back
 985
 986 =head2 Opcode Functions
 987
 988 Functions that are called from within opcodes, that take and return an
 989 opcode_t* to allow for changing the code flow.
 990
 991 =over 4
 992
 993
 994 =item C<opcode_t * Parrot_cx_schedule_sleep(PARROT_INTERP, FLOATVAL time,
 995 opcode_t *next)>
 996
 997 Add a sleep timer to the scheduler. This function is called by the C<sleep>
 998 opcode.
 999
1000 =cut
1001
1002 */
1003
1004 PARROT_EXPORT
1005 PARROT_WARN_UNUSED_RESULT
1006 PARROT_CAN_RETURN_NULL
1007 opcode_t *
1008 Parrot_cx_schedule_sleep(PARROT_INTERP, FLOATVAL time, ARGIN_NULLOK(opcode_t *next))
1009 {
1010     ASSERT_ARGS(Parrot_cx_schedule_sleep)
1011 #ifdef PARROT_HAS_THREADS
1012     Parrot_cond condition;
1013     Parrot_mutex lock;
1014     const FLOATVAL timer_end = time + Parrot_floatval_time();
1015     struct timespec time_struct;
1016
1017     /* Tell the scheduler runloop to wake, this is a good time to process
1018      * pending tasks. */
1019     Parrot_cx_runloop_wake(interp, interp->scheduler);
1020
1021     /* Tell this thread to sleep for the requested time. */
1022     COND_INIT(condition);
1023     MUTEX_INIT(lock);
1024     LOCK(lock);
1025     time_struct.tv_sec = (time_t) timer_end;
1026     time_struct.tv_nsec = (long)((timer_end - time_struct.tv_sec)*1000.0f) *1000L*1000L;
1027     COND_TIMED_WAIT(condition, lock, &time_struct);
1028     UNLOCK(lock);
1029     COND_DESTROY(condition);
1030     MUTEX_DESTROY(lock);
1031 #else
1032     /* A more primitive, platform-specific, non-threaded form of sleep. */
1033     if (time > 1000) {
1034         /* prevent integer overflow when converting to microseconds */
1035         const int seconds = floor(time);
1036         Parrot_sleep(seconds);
1037         time -= seconds;
1038     }
1039     Parrot_usleep((UINTVAL) time*1000000);
1040 #endif
1041     return next;
1042 }
1043
1044
1045 /*
1046
1047 =back
1048
1049 =head2 Internal Functions
1050
1051 Functions that are only used within the scheduler.
1052
1053 =over 4
1054
1055 =item C<static void scheduler_process_wait_list(PARROT_INTERP, PMC *scheduler)>
1056
1057 Scheduler maintenance, scan the list of waiting tasks to see if any are ready
1058 to become active tasks.
1059
1060 =cut
1061
1062 */
1063
1064 static void
1065 scheduler_process_wait_list(PARROT_INTERP, ARGMOD(PMC *scheduler))
1066 {
1067     ASSERT_ARGS(scheduler_process_wait_list)
1068     Parrot_Scheduler_attributes * sched_struct = PARROT_SCHEDULER(scheduler);
1069     INTVAL num_tasks, index;
1070
1071     /* Sweep the wait list for completed timers */
1072     num_tasks = VTABLE_elements(interp, sched_struct->wait_index);
1073     for (index = 0; index < num_tasks; index++) {
1074         INTVAL tid = VTABLE_get_integer_keyed_int(interp, sched_struct->wait_index, index);
1075         if (tid > 0) {
1076             PMC *task = VTABLE_get_pmc_keyed_int(interp, sched_struct->task_list, tid);
1077             if (PMC_IS_NULL(task)) {
1078                 /* Cleanup expired tasks. */
1079                 VTABLE_set_integer_keyed_int(interp, sched_struct->wait_index, index, 0);
1080             }
1081             else {
1082                 /* Move the timer to the active task list if the timer has
1083                  * completed. */
1084                 FLOATVAL timer_end_time = VTABLE_get_number_keyed_int(interp,
1085                         task, PARROT_TIMER_NSEC);
1086                 if (timer_end_time <= Parrot_floatval_time()) {
1087                     VTABLE_push_integer(interp, sched_struct->task_index, tid);
1088                     VTABLE_set_integer_keyed_int(interp, sched_struct->wait_index, index, 0);
1089                     Parrot_cx_schedule_repeat(interp, task);
1090                     SCHEDULER_cache_valid_CLEAR(scheduler);
1091                 }
1092             }
1093         }
1094     }
1095 }
1096
1097 /*
1098
1099 =over 4
1100
1101 =item C<static void scheduler_process_messages(PARROT_INTERP, PMC *scheduler)>
1102
1103 Scheduler maintenance, scan the list of messages sent from other schedulers and
1104 take appropriate action on any received.
1105
1106 =cut
1107
1108 */
1109
1110 static void
1111 scheduler_process_messages(PARROT_INTERP, ARGMOD(PMC *scheduler))
1112 {
1113     ASSERT_ARGS(scheduler_process_messages)
1114     Parrot_Scheduler_attributes * sched_struct = PARROT_SCHEDULER(scheduler);
1115
1116     PMC    *message;
1117     STRING *suspend_str = CONST_STRING(interp, "suspend_for_gc");
1118
1119 #if CX_DEBUG
1120     fprintf(stderr, "processing messages [interp=%p]\n", interp);
1121 #endif
1122
1123     while (VTABLE_elements(interp, sched_struct->messages) > 0) {
1124 #if CX_DEBUG
1125     fprintf(stderr, "locking msg_lock (process) [interp=%p]\n", interp);
1126 #endif
1127         LOCK(sched_struct->msg_lock);
1128         message = VTABLE_pop_pmc(interp, sched_struct->messages);
1129 #if CX_DEBUG
1130     fprintf(stderr, "unlocking msg_lock (process) [interp=%p]\n", interp);
1131 #endif
1132         UNLOCK(sched_struct->msg_lock);
1133         if (!PMC_IS_NULL(message)
1134          && Parrot_str_equal(interp, VTABLE_get_string(interp, message),
1135                 suspend_str)) {
1136 #if CX_DEBUG
1137     fprintf(stderr, "found a suspend, suspending [interp=%p]\n", interp);
1138 #endif
1139             pt_suspend_self_for_gc(interp);
1140         }
1141     }
1142
1143 }
1144
1145 /*
1146
1147 =back
1148
1149 =cut
1150
1151 */
1152
1153
1154 /*
1155  * Local variables:
1156  *   c-file-style: "parrot"
1157  * End:
1158  * vim: expandtab shiftwidth=4:
1159  */