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 #if 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 /* 1K room for the cond wait routine */
 515 #if SCM_STACK_GROWS_UP
 516 # define COOP_STACK_ROOM (256)
 517 #else
 518 # define COOP_STACK_ROOM (-256)
 519 #endif
 520
 521 static void *
 522 dummy_start (void *coop_thread)
 523 {
 524   coop_t *t = (coop_t *) coop_thread;
 525   int res;
 526   t->sp = (qt_t *) (&t + COOP_STACK_ROOM);
 527   pthread_mutex_init (&t->dummy_mutex, NULL);
 528   pthread_mutex_lock (&t->dummy_mutex);
 529   coop_child = 0;
 530   do
 531     res = pthread_cond_wait (&coop_cond_quit, &t->dummy_mutex);
 532   while (res == EINTR);
 533   return 0;
 534 }
 535
 536 static void *
 537 mother (void *dummy)
 538 {
 539   pthread_mutex_lock (&coop_mutex_create);
 540   while (!coop_quitting_p)
 541     {
 542       int res;
 543       pthread_create (&coop_child->dummy_thread,
 544                       NULL,
 545                       dummy_start,
 546                       coop_child);
 547       mother_awake_p = 0;
 548       do
 549         res = pthread_cond_wait (&coop_cond_create, &coop_mutex_create);
 550       while (res == EINTR);
 551     }
 552   return 0;
 553 }
 554
 555 #endif
 556
 557 coop_t *
 558 coop_create (coop_userf_t *f, void *pu)
 559 {
 560   coop_t *t;
 561 #ifndef GUILE_PTHREAD_COMPAT
 562   void *sto;
 563 #endif
 564
 565 #ifdef GUILE_PTHREAD_COMPAT
 566   t = coop_qget (&coop_deadq);
 567   if (t)
 568     {
 569       t->sp = t->base;
 570       t->specific = 0;
 571       t->n_keys = 0;
 572     }
 573   else
 574 #endif
 575     {
 576       t = scm_malloc (sizeof (coop_t));
 577       t->specific = NULL;
 578       t->n_keys = 0;
 579 #ifdef GUILE_PTHREAD_COMPAT
 580       coop_child = t;
 581       mother_awake_p = 1;
 582       if (coop_quitting_p < 0)
 583         {
 584           coop_quitting_p = 0;
 585           /* We can't create threads ourselves since the pthread
 586            * corresponding to this stack might be sleeping.
 587            */
 588           pthread_create (&coop_mother, NULL, mother, NULL);
 589         }
 590       else
 591         {
 592           pthread_cond_signal (&coop_cond_create);
 593         }
 594       /* We can't use a pthreads condition variable since "this"
 595        * pthread could already be asleep.  We can't use a COOP
 596        * condition variable because they are not safe against
 597        * pre-emptive switching.
 598        */
 599       while (coop_child || mother_awake_p)
 600         usleep (0);
 601 #else
 602       t->sto = scm_malloc (COOP_STKSIZE);
 603       sto = COOP_STKALIGN (t->sto, QT_STKALIGN);
 604       t->sp = QT_SP (sto, COOP_STKSIZE - QT_STKALIGN);
 605 #endif
 606       t->base = t->sp;
 607     }
 608   t->sp = QT_ARGS (t->sp, pu, t, (qt_userf_t *)f, coop_only);
 609   t->joining = NULL;
 610   coop_qput (&coop_global_runq, t);
 611   coop_all_qput (&coop_global_allq, t);
 612
 613   return t;
 614 }
 615
 616
 617 static void
 618 coop_only (void *pu, void *pt, qt_userf_t *f)
 619 {
 620   coop_global_curr = (coop_t *)pt;
 621   (*(coop_userf_t *)f)(pu);
 622   coop_abort();
 623   /* NOTREACHED */
 624 }
 625
 626
 627 void
 628 coop_abort ()
 629 {
 630   coop_t *old, *newthread;
 631
 632   /* Wake up any threads that are waiting to join this one */
 633   if (coop_global_curr->joining)
 634     {
 635       while ((newthread = coop_qget ((coop_q_t *)(coop_global_curr->joining)))
 636              != NULL)
 637         {
 638           coop_qput (&coop_global_runq, newthread);
 639         }
 640       free (coop_global_curr->joining);
 641     }
 642
 643   scm_I_am_dead = 1;
 644   do {
 645     newthread = coop_wait_for_runnable_thread();
 646   } while (newthread == coop_global_curr);
 647   scm_I_am_dead = 0;
 648   coop_all_qremove (&coop_global_allq, coop_global_curr);
 649   old = coop_global_curr;
 650   coop_global_curr = newthread;
 651   QT_ABORT (coop_aborthelp, old, (void *) NULL, newthread->sp);
 652 }
 653
 654
 655 static void *
 656 coop_aborthelp (qt_t *sp, void *old, void *null)
 657 {
 658   coop_t *oldthread = (coop_t *) old;
 659
 660   if (oldthread->specific)
 661     free (oldthread->specific);
 662 #ifndef GUILE_PTHREAD_COMPAT
 663   free (oldthread->sto);
 664   free (oldthread);
 665 #else
 666   coop_qput (&coop_deadq, oldthread);
 667 #endif
 668
 669   return NULL;
 670 }
 671
 672
 673 void
 674 coop_join(coop_t *t)
 675 {
 676   coop_t *old, *newthread;
 677
 678   /* Create a join list if necessary */
 679   if (t->joining == NULL)
 680     {
 681       t->joining = scm_malloc(sizeof(coop_q_t));
 682       coop_qinit((coop_q_t *) t->joining);
 683     }
 684
 685   newthread = coop_wait_for_runnable_thread();
 686   if (newthread == coop_global_curr)
 687     return;
 688   old = coop_global_curr;
 689   coop_global_curr = newthread;
 690   QT_BLOCK (coop_yieldhelp, old, (coop_q_t *) t->joining, newthread->sp);
 691 }
 692
 693 void
 694 coop_yield()
 695 {
 696   coop_t *old = NULL;
 697   coop_t *newthread;
 698
 699   newthread = coop_next_runnable_thread();
 700
 701   /* There may be no other runnable threads. Return if this is the
 702      case. */
 703   if (newthread == coop_global_curr)
 704       return;
 705
 706   old = coop_global_curr;
 707
 708   coop_global_curr = newthread;
 709   QT_BLOCK (coop_yieldhelp, old, &coop_global_runq, newthread->sp);
 710 }
 711
 712
 713 static void *
 714 coop_yieldhelp (qt_t *sp, void *old, void *blockq)
 715 {
 716   ((coop_t *)old)->sp = sp;
 717   coop_qput ((coop_q_t *)blockq, (coop_t *)old);
 718   return NULL;
 719 }
 720
 721 /* Replacement for the system's sleep() function. Does the right thing
 722    for the process - but not for the system (it busy-waits) */
 723
 724 void *
 725 coop_sleephelp (qt_t *sp, void *old, void *blockq)
 726 {
 727   ((coop_t *)old)->sp = sp;
 728   /* old is already on the sleep queue - so there's no need to
 729      do anything extra here */
 730   return NULL;
 731 }
 732
 733 unsigned long
 734 scm_thread_usleep (unsigned long usec)
 735 {
 736   struct timeval timeout;
 737   timeout.tv_sec = 0;
 738   timeout.tv_usec = usec;
 739   scm_internal_select (0, NULL, NULL, NULL, &timeout);
 740   return 0;  /* Maybe we should calculate actual time slept,
 741                 but this is faster... :) */
 742 }
 743
 744 unsigned long
 745 scm_thread_sleep (unsigned long sec)
 746 {
 747   time_t now = time (NULL);
 748   struct timeval timeout;
 749   unsigned long slept;
 750   timeout.tv_sec = sec;
 751   timeout.tv_usec = 0;
 752   scm_internal_select (0, NULL, NULL, NULL, &timeout);
 753   slept = time (NULL) - now;
 754   return slept > sec ? 0 : sec - slept;
 755 }
 756
 757 /*
 758   Local Variables:
 759   c-file-style: "gnu"
 760   End:
 761 */