src/scheduler.c

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