libguile/coop.c

   1 /*      Copyright (C) 1995, 1996, 1997, 1998, 1999, 2000, 2001, 2002, 2006 Free Software Foundation, Inc.
   2  *
   3  * This library is free software; you can redistribute it and/or
   4  * modify it under the terms of the GNU Lesser General Public
   5  * License as published by the Free Software Foundation; either
   6  * version 2.1 of the License, or (at your option) any later version.
   7  *
   8  * This library is distributed in the hope that it will be useful,
   9  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  10  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  11  * Lesser General Public License for more details.
  12  *
  13  * You should have received a copy of the GNU Lesser General Public
  14  * License along with this library; if not, write to the Free Software
  15  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  16  */
  17 \f
  18
  19 /* $Id: coop.c,v 1.39 2006-04-17 00:05:38 kryde Exp $ */
  20
  21 /* Cooperative thread library, based on QuickThreads */
  22
  23 #ifdef HAVE_CONFIG_H
  24 #  include <config.h>
  25 #endif
  26
  27 #include <stdio.h>
  28
  29 #ifdef HAVE_UNISTD_H
  30 #include <unistd.h>
  31 #endif
  32
  33 #include <errno.h>
  34
  35 #include "qt/qt.h"
  36 #include "libguile/eval.h"
  37
  38 \f/* #define COOP_STKSIZE (0x10000) */
  39 #define COOP_STKSIZE (scm_eval_stack)
  40
  41 /* `alignment' must be a power of 2. */
  42 #define COOP_STKALIGN(sp, alignment) \
  43 ((void *)((((qt_word_t)(sp)) + (alignment) - 1) & ~((alignment)-1)))
  44
  45 \f
  46
  47 /* Queue access functions. */
  48
  49 static void
  50 coop_qinit (coop_q_t *q)
  51 {
  52   q->t.next = q->tail = &q->t;
  53
  54   q->t.all_prev = NULL;
  55   q->t.all_next = NULL;
  56   q->t.nfds = 0;
  57   q->t.readfds = NULL;
  58   q->t.writefds = NULL;
  59   q->t.exceptfds = NULL;
  60   q->t.timeoutp = 0;
  61 }
  62
  63
  64 coop_t *
  65 coop_qget (coop_q_t *q)
  66 {
  67   coop_t *t;
  68
  69   t = q->t.next;
  70   q->t.next = t->next;
  71   if (t->next == &q->t)
  72     {
  73       if (t == &q->t)
  74         {                       /* If it was already empty .. */
  75           return NULL;          /* .. say so. */
  76         }
  77       q->tail = &q->t;          /* Else now it is empty. */
  78     }
  79   return (t);
  80 }
  81
  82
  83 void
  84 coop_qput (coop_q_t *q, coop_t *t)
  85 {
  86   q->tail->next = t;
  87   t->next = &q->t;
  88   q->tail = t;
  89 }
  90
  91 static void
  92 coop_all_qput (coop_q_t *q, coop_t *t)
  93 {
  94   if (q->t.all_next)
  95     q->t.all_next->all_prev = t;
  96   t->all_prev = NULL;
  97   t->all_next = q->t.all_next;
  98   q->t.all_next = t;
  99 }
 100
 101 static void
 102 coop_all_qremove (coop_q_t *q, coop_t *t)
 103 {
 104   if (t->all_prev)
 105     t->all_prev->all_next = t->all_next;
 106   else
 107     q->t.all_next = t->all_next;
 108   if (t->all_next)
 109       t->all_next->all_prev = t->all_prev;
 110 }
 111
 112 /* Insert thread t into the ordered queue q.
 113    q is ordered after wakeup_time.  Threads which aren't sleeping but
 114    waiting for I/O go last into the queue. */
 115 void
 116 coop_timeout_qinsert (coop_q_t *q, coop_t *t)
 117 {
 118   coop_t *pred = &q->t;
 119   int sec = t->wakeup_time.tv_sec;
 120   int usec = t->wakeup_time.tv_usec;
 121   while (pred->next != &q->t
 122          && pred->next->timeoutp
 123          && (pred->next->wakeup_time.tv_sec < sec
 124              || (pred->next->wakeup_time.tv_sec == sec
 125                  && pred->next->wakeup_time.tv_usec < usec)))
 126     pred = pred->next;
 127   t->next = pred->next;
 128   pred->next = t;
 129   if (t->next == &q->t)
 130     q->tail = t;
 131 }
 132
 133 \f
 134
 135 /* Thread routines. */
 136
 137 coop_q_t coop_global_runq;      /* A queue of runable threads. */
 138 coop_q_t coop_global_sleepq;    /* A queue of sleeping threads. */
 139 coop_q_t coop_tmp_queue;        /* A temp working queue */
 140 coop_q_t coop_global_allq;      /* A queue of all threads. */
 141 static coop_t coop_global_main; /* Thread for the process. */
 142 coop_t *coop_global_curr;       /* Currently-executing thread. */
 143
 144 #ifdef GUILE_PTHREAD_COMPAT
 145 static coop_q_t coop_deadq;
 146 static int coop_quitting_p = -1;
 147 static pthread_cond_t coop_cond_quit;
 148 static pthread_cond_t coop_cond_create;
 149 static pthread_mutex_t coop_mutex_create;
 150 static pthread_t coop_mother;
 151 static int mother_awake_p = 0;
 152 static coop_t *coop_child;
 153 #endif
 154
 155 static void *coop_starthelp (qt_t *old, void *ignore0, void *ignore1);
 156 static void coop_only (void *pu, void *pt, qt_userf_t *f);
 157 static void *coop_aborthelp (qt_t *sp, void *old, void *null);
 158 static void *coop_yieldhelp (qt_t *sp, void *old, void *blockq);
 159
 160
 161 /* called on process termination.  */
 162 #ifdef HAVE_ATEXIT
 163 static void
 164 coop_finish (void)
 165 #else
 166 #ifdef HAVE_ON_EXIT
 167 extern int on_exit (void (*procp) (), int arg);
 168
 169 static void
 170 coop_finish (int status, void *arg)
 171 #else
 172 #error Dont know how to setup a cleanup handler on your system.
 173 #endif
 174 #endif
 175 {
 176 #ifdef GUILE_PTHREAD_COMPAT
 177   coop_quitting_p = 1;
 178   pthread_cond_signal (&coop_cond_create);
 179   pthread_cond_broadcast (&coop_cond_quit);
 180 #endif
 181 }
 182
 183 void
 184 coop_init ()
 185 {
 186   coop_qinit (&coop_global_runq);
 187   coop_qinit (&coop_global_sleepq);
 188   coop_qinit (&coop_tmp_queue);
 189   coop_qinit (&coop_global_allq);
 190   coop_global_curr = &coop_global_main;
 191 #ifdef GUILE_PTHREAD_COMPAT
 192   coop_qinit (&coop_deadq);
 193   pthread_cond_init (&coop_cond_quit, NULL);
 194   pthread_cond_init (&coop_cond_create, NULL);
 195   pthread_mutex_init (&coop_mutex_create, NULL);
 196 #endif
 197 #ifdef HAVE_ATEXIT
 198   atexit (coop_finish);
 199 #else
 200 #ifdef HAVE_ON_EXIT
 201   on_exit (coop_finish, 0);
 202 #endif
 203 #endif
 204 }
 205
 206 void
 207 coop_start()
 208 {
 209   coop_t *next;
 210
 211   while ((next = coop_qget (&coop_global_runq)) != NULL) {
 212     coop_global_curr = next;
 213     QT_BLOCK (coop_starthelp, 0, 0, next->sp);
 214   }
 215 }
 216
 217
 218 static void *
 219 coop_starthelp (qt_t *old, void *ignore0, void *ignore1)
 220 {
 221   coop_global_main.sp = old;
 222   coop_global_main.joining = NULL;
 223   coop_qput (&coop_global_runq, &coop_global_main);
 224   return NULL; /* not used, but keeps compiler happy */
 225 }
 226
 227 int
 228 coop_mutex_init (coop_m *m)
 229 {
 230   return coop_new_mutex_init (m, NULL);
 231 }
 232
 233 int
 234 coop_new_mutex_init (coop_m *m, coop_mattr *attr)
 235 {
 236   m->owner = NULL;
 237   m->level = 0;
 238   coop_qinit(&(m->waiting));
 239   return 0;
 240 }
 241
 242 int
 243 coop_mutex_trylock (coop_m *m)
 244 {
 245   if (m->owner == NULL)
 246     {
 247       m->owner = coop_global_curr;
 248       return 0;
 249     }
 250   else if (m->owner == coop_global_curr)
 251     {
 252       m->level++;
 253       return 0;
 254     }
 255   else
 256     return EBUSY;
 257 }
 258
 259 int
 260 coop_mutex_lock (coop_m *m)
 261 {
 262   if (m->owner == NULL)
 263     {
 264       m->owner = coop_global_curr;
 265     }
 266   else if (m->owner == coop_global_curr)
 267     {
 268       m->level++;
 269     }
 270   else
 271     {
 272       coop_t *old, *newthread;
 273
 274       /* Record the current top-of-stack before going to sleep */
 275       coop_global_curr->top = &old;
 276
 277       newthread = coop_wait_for_runnable_thread();
 278       if (newthread == coop_global_curr)
 279         coop_abort ();
 280       old = coop_global_curr;
 281       coop_global_curr = newthread;
 282       QT_BLOCK (coop_yieldhelp, old, &(m->waiting), newthread->sp);
 283     }
 284   return 0;
 285 }
 286
 287
 288 int
 289 coop_mutex_unlock (coop_m *m)
 290 {
 291   coop_t *old, *newthread;
 292
 293   if (m->level == 0)
 294     {
 295       newthread = coop_qget (&(m->waiting));
 296       if (newthread != NULL)
 297         {
 298           /* Record the current top-of-stack before going to sleep */
 299           coop_global_curr->top = &old;
 300
 301           old = coop_global_curr;
 302           coop_global_curr = newthread;
 303           /* The new thread came into m->waiting through a lock operation.
 304              It now owns this mutex. */
 305           m->owner = coop_global_curr;
 306           QT_BLOCK (coop_yieldhelp, old, &coop_global_runq, newthread->sp);
 307         }
 308       else
 309         {
 310           m->owner = NULL;
 311         }
 312     }
 313   else if (m->level > 0)
 314     m->level--;
 315   else
 316     abort (); /* XXX */
 317
 318   return 0;
 319 }
 320
 321
 322 int
 323 coop_mutex_destroy (coop_m *m)
 324 {
 325   return 0;
 326 }
 327
 328
 329 int
 330 coop_condition_variable_init (coop_c *c)
 331 {
 332   return coop_new_condition_variable_init (c, NULL);
 333 }
 334
 335 int
 336 coop_new_condition_variable_init (coop_c *c, coop_cattr *a)
 337 {
 338   coop_qinit(&(c->waiting));
 339   return 0;
 340 }
 341
 342 int
 343 coop_condition_variable_wait_mutex (coop_c *c, coop_m *m)
 344 {
 345   coop_t *old, *newthread;
 346
 347   /* coop_mutex_unlock (m); */
 348   newthread = coop_qget (&(m->waiting));
 349   if (newthread != NULL)
 350     {
 351       m->owner = newthread;
 352     }
 353   else
 354     {
 355       m->owner = NULL;
 356       /*fixme* Should we really wait here?  Isn't it OK just to proceed? */
 357       newthread = coop_wait_for_runnable_thread();
 358       if (newthread == coop_global_curr)
 359         coop_abort ();
 360     }
 361   coop_global_curr->top = &old;
 362   old = coop_global_curr;
 363   coop_global_curr = newthread;
 364   QT_BLOCK (coop_yieldhelp, old, &(c->waiting), newthread->sp);
 365
 366   coop_mutex_lock (m);
 367   return 0;
 368 }
 369
 370 int
 371 coop_condition_variable_timed_wait_mutex (coop_c *c,
 372                                           coop_m *m,
 373                                           const scm_t_timespec *abstime)
 374 {
 375   coop_t *old, *t;
 376 #ifdef ETIMEDOUT
 377   int res = ETIMEDOUT;
 378 #elif defined (WSAETIMEDOUT)
 379   int res = WSAETIMEDOUT;
 380 #else
 381   int res = 0;
 382 #endif
 383
 384   /* coop_mutex_unlock (m); */
 385   t = coop_qget (&(m->waiting));
 386   if (t != NULL)
 387     {
 388       m->owner = t;
 389     }
 390   else
 391     {
 392       m->owner = NULL;
 393       coop_global_curr->timeoutp = 1;
 394       coop_global_curr->wakeup_time.tv_sec = abstime->tv_sec;
 395       coop_global_curr->wakeup_time.tv_usec = abstime->tv_nsec / 1000;
 396       coop_timeout_qinsert (&coop_global_sleepq, coop_global_curr);
 397       t = coop_wait_for_runnable_thread();
 398     }
 399   if (t != coop_global_curr)
 400     {
 401       coop_global_curr->top = &old;
 402       old = coop_global_curr;
 403       coop_global_curr = t;
 404       QT_BLOCK (coop_yieldhelp, old, &(c->waiting), t->sp);
 405
 406       /* Are we still in the sleep queue? */
 407       old = &coop_global_sleepq.t;
 408       for (t = old->next; t != &coop_global_sleepq.t; old = t, t = t->next)
 409         if (t == coop_global_curr)
 410           {
 411             old->next = t->next; /* unlink */
 412             res = 0;
 413             break;
 414           }
 415     }
 416   coop_mutex_lock (m);
 417   return res;
 418 }
 419
 420 int
 421 coop_condition_variable_broadcast (coop_c *c)
 422 {
 423   coop_t *newthread;
 424
 425   while ((newthread = coop_qget (&(c->waiting))) != NULL)
 426     {
 427       coop_qput (&coop_global_runq, newthread);
 428     }
 429   return 0;
 430 }
 431
 432 int
 433 coop_condition_variable_signal (coop_c *c)
 434 {
 435   return coop_condition_variable_broadcast (c);
 436 }
 437
 438
 439 /* {Keys}
 440  */
 441
 442 static int n_keys = 0;
 443 static int max_keys = 0;
 444 static void (**destructors) (void *) = 0;
 445
 446 int
 447 coop_key_create (coop_k *keyp, void (*destructor) (void *value))
 448 {
 449   if (n_keys >= max_keys)
 450     {
 451       int i;
 452       max_keys = max_keys ? max_keys * 3 / 2 : 10;
 453       destructors = realloc (destructors, sizeof (void *) * max_keys);
 454       if (destructors == 0)
 455         {
 456           fprintf (stderr, "Virtual memory exceeded in coop_key_create\n");
 457           exit (1);
 458         }
 459       for (i = n_keys; i < max_keys; ++i)
 460         destructors[i] = NULL;
 461     }
 462   destructors[n_keys] = destructor;
 463   *keyp = n_keys++;
 464   return 0;
 465 }
 466
 467 int
 468 coop_setspecific (coop_k key, const void *value)
 469 {
 470   int n_keys = coop_global_curr->n_keys;
 471   if (key >= n_keys)
 472     {
 473       int i;
 474       coop_global_curr->n_keys = max_keys;
 475       coop_global_curr->specific = realloc (n_keys
 476                                             ? coop_global_curr->specific
 477                                             : NULL,
 478                                             sizeof (void *) * max_keys);
 479       if (coop_global_curr->specific == 0)
 480         {
 481           fprintf (stderr, "Virtual memory exceeded in coop_setspecific\n");
 482           exit (1);
 483         }
 484       for (i = n_keys; i < max_keys; ++i)
 485         coop_global_curr->specific[i] = NULL;
 486     }
 487   coop_global_curr->specific[key] = (void *) value;
 488   return 0;
 489 }
 490
 491 void *
 492 coop_getspecific (coop_k key)
 493 {
 494   return (key < coop_global_curr->n_keys
 495           ? coop_global_curr->specific[key]
 496           : NULL);
 497 }
 498
 499 int
 500 coop_key_delete (coop_k key)
 501 {
 502   return 0;
 503 }
 504
 505
 506 int
 507 coop_condition_variable_destroy (coop_c *c)
 508 {
 509   return 0;
 510 }
 511
 512 #ifdef GUILE_PTHREAD_COMPAT
 513
 514 #include "libguile/boehm-gc.h"
 515
 516 /* 1K room for the cond wait routine */
 517 #if SCM_STACK_GROWS_UP
 518 # define COOP_STACK_ROOM (256)
 519 #else
 520 # define COOP_STACK_ROOM (-256)
 521 #endif
 522
 523 static void *
 524 dummy_start (void *coop_thread)
 525 {
 526   coop_t *t = (coop_t *) coop_thread;
 527   int res;
 528   t->sp = (qt_t *) (&t + COOP_STACK_ROOM);
 529   pthread_mutex_init (&t->dummy_mutex, NULL);
 530   pthread_mutex_lock (&t->dummy_mutex);
 531   coop_child = 0;
 532   do
 533     res = pthread_cond_wait (&coop_cond_quit, &t->dummy_mutex);
 534   while (res == EINTR);
 535   return 0;
 536 }
 537
 538 static void *
 539 mother (void *dummy)
 540 {
 541   pthread_mutex_lock (&coop_mutex_create);
 542   while (!coop_quitting_p)
 543     {
 544       int res;
 545       pthread_create (&coop_child->dummy_thread,
 546                       NULL,
 547                       dummy_start,
 548                       coop_child);
 549       mother_awake_p = 0;
 550       do
 551         res = pthread_cond_wait (&coop_cond_create, &coop_mutex_create);
 552       while (res == EINTR);
 553     }
 554   return 0;
 555 }
 556
 557 #endif
 558
 559 coop_t *
 560 coop_create (coop_userf_t *f, void *pu)
 561 {
 562   coop_t *t;
 563 #ifndef GUILE_PTHREAD_COMPAT
 564   void *sto;
 565 #endif
 566
 567 #ifdef GUILE_PTHREAD_COMPAT
 568   t = coop_qget (&coop_deadq);
 569   if (t)
 570     {
 571       t->sp = t->base;
 572       t->specific = 0;
 573       t->n_keys = 0;
 574     }
 575   else
 576 #endif
 577     {
 578       t = scm_malloc (sizeof (coop_t));
 579       t->specific = NULL;
 580       t->n_keys = 0;
 581 #ifdef GUILE_PTHREAD_COMPAT
 582       coop_child = t;
 583       mother_awake_p = 1;
 584       if (coop_quitting_p < 0)
 585         {
 586           coop_quitting_p = 0;
 587           /* We can't create threads ourselves since the pthread
 588            * corresponding to this stack might be sleeping.
 589            */
 590           pthread_create (&coop_mother, NULL, mother, NULL);
 591         }
 592       else
 593         {
 594           pthread_cond_signal (&coop_cond_create);
 595         }
 596       /* We can't use a pthreads condition variable since "this"
 597        * pthread could already be asleep.  We can't use a COOP
 598        * condition variable because they are not safe against
 599        * pre-emptive switching.
 600        */
 601       while (coop_child || mother_awake_p)
 602         usleep (0);
 603 #else
 604       t->sto = scm_malloc (COOP_STKSIZE);
 605       sto = COOP_STKALIGN (t->sto, QT_STKALIGN);
 606       t->sp = QT_SP (sto, COOP_STKSIZE - QT_STKALIGN);
 607 #endif
 608       t->base = t->sp;
 609     }
 610   t->sp = QT_ARGS (t->sp, pu, t, (qt_userf_t *)f, coop_only);
 611   t->joining = NULL;
 612   coop_qput (&coop_global_runq, t);
 613   coop_all_qput (&coop_global_allq, t);
 614
 615   return t;
 616 }
 617
 618
 619 static void
 620 coop_only (void *pu, void *pt, qt_userf_t *f)
 621 {
 622   coop_global_curr = (coop_t *)pt;
 623   (*(coop_userf_t *)f)(pu);
 624   coop_abort();
 625   /* NOTREACHED */
 626 }
 627
 628
 629 void
 630 coop_abort ()
 631 {
 632   coop_t *old, *newthread;
 633
 634   /* Wake up any threads that are waiting to join this one */
 635   if (coop_global_curr->joining)
 636     {
 637       while ((newthread = coop_qget ((coop_q_t *)(coop_global_curr->joining)))
 638              != NULL)
 639         {
 640           coop_qput (&coop_global_runq, newthread);
 641         }
 642       free (coop_global_curr->joining);
 643     }
 644
 645   scm_I_am_dead = 1;
 646   do {
 647     newthread = coop_wait_for_runnable_thread();
 648   } while (newthread == coop_global_curr);
 649   scm_I_am_dead = 0;
 650   coop_all_qremove (&coop_global_allq, coop_global_curr);
 651   old = coop_global_curr;
 652   coop_global_curr = newthread;
 653   QT_ABORT (coop_aborthelp, old, (void *) NULL, newthread->sp);
 654 }
 655
 656
 657 static void *
 658 coop_aborthelp (qt_t *sp, void *old, void *null)
 659 {
 660   coop_t *oldthread = (coop_t *) old;
 661
 662   if (oldthread->specific)
 663     free (oldthread->specific);
 664 #ifndef GUILE_PTHREAD_COMPAT
 665   free (oldthread->sto);
 666   free (oldthread);
 667 #else
 668   coop_qput (&coop_deadq, oldthread);
 669 #endif
 670
 671   return NULL;
 672 }
 673
 674
 675 void
 676 coop_join(coop_t *t)
 677 {
 678   coop_t *old, *newthread;
 679
 680   /* Create a join list if necessary */
 681   if (t->joining == NULL)
 682     {
 683       t->joining = scm_malloc(sizeof(coop_q_t));
 684       coop_qinit((coop_q_t *) t->joining);
 685     }
 686
 687   newthread = coop_wait_for_runnable_thread();
 688   if (newthread == coop_global_curr)
 689     return;
 690   old = coop_global_curr;
 691   coop_global_curr = newthread;
 692   QT_BLOCK (coop_yieldhelp, old, (coop_q_t *) t->joining, newthread->sp);
 693 }
 694
 695 void
 696 coop_yield()
 697 {
 698   coop_t *old = NULL;
 699   coop_t *newthread;
 700
 701   newthread = coop_next_runnable_thread();
 702
 703   /* There may be no other runnable threads. Return if this is the
 704      case. */
 705   if (newthread == coop_global_curr)
 706       return;
 707
 708   old = coop_global_curr;
 709
 710   coop_global_curr = newthread;
 711   QT_BLOCK (coop_yieldhelp, old, &coop_global_runq, newthread->sp);
 712 }
 713
 714
 715 static void *
 716 coop_yieldhelp (qt_t *sp, void *old, void *blockq)
 717 {
 718   ((coop_t *)old)->sp = sp;
 719   coop_qput ((coop_q_t *)blockq, (coop_t *)old);
 720   return NULL;
 721 }
 722
 723 /* Replacement for the system's sleep() function. Does the right thing
 724    for the process - but not for the system (it busy-waits) */
 725
 726 void *
 727 coop_sleephelp (qt_t *sp, void *old, void *blockq)
 728 {
 729   ((coop_t *)old)->sp = sp;
 730   /* old is already on the sleep queue - so there's no need to
 731      do anything extra here */
 732   return NULL;
 733 }
 734
 735 unsigned long
 736 scm_thread_usleep (unsigned long usec)
 737 {
 738   struct timeval timeout;
 739   timeout.tv_sec = 0;
 740   timeout.tv_usec = usec;
 741   scm_internal_select (0, NULL, NULL, NULL, &timeout);
 742   return 0;  /* Maybe we should calculate actual time slept,
 743                 but this is faster... :) */
 744 }
 745
 746 unsigned long
 747 scm_thread_sleep (unsigned long sec)
 748 {
 749   time_t now = time (NULL);
 750   struct timeval timeout;
 751   unsigned long slept;
 752   timeout.tv_sec = sec;
 753   timeout.tv_usec = 0;
 754   scm_internal_select (0, NULL, NULL, NULL, &timeout);
 755   slept = time (NULL) - now;
 756   return slept > sec ? 0 : sec - slept;
 757 }
 758
 759 /*
 760   Local Variables:
 761   c-file-style: "gnu"
 762   End:
 763 */