gst/cothreads.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *                    2000 Wim Taymans <wtay@chello.be>
   4  *
   5  * cothreads.c: Cothreading routines
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  20  * Boston, MA 02111-1307, USA.
  21  */
  22
  23 #include <pthread.h>
  24 #include <stdio.h>
  25 #include <stdlib.h>
  26 #include <signal.h>
  27 #include <setjmp.h>
  28 #include <unistd.h>
  29 #include <errno.h>
  30 #include <sys/mman.h>
  31
  32 #include "gst_private.h"
  33
  34 #include "cothreads.h"
  35 #include "gstarch.h"
  36 #include "gstlog.h"
  37 #include "gstutils.h"
  38
  39
  40 #define STACK_SIZE 0x200000
  41
  42 #define COTHREAD_MAGIC_NUMBER 0xabcdef
  43
  44 #define COTHREAD_MAXTHREADS 16
  45 #define COTHREAD_STACKSIZE (STACK_SIZE/COTHREAD_MAXTHREADS)
  46
  47 static void     cothread_destroy        (cothread_state *thread);
  48
  49 struct _cothread_context
  50 {
  51   cothread_state *threads[COTHREAD_MAXTHREADS];
  52   int nthreads;
  53   int current;
  54   GHashTable *data;
  55 };
  56
  57
  58 static pthread_key_t _cothread_key = -1;
  59
  60 /* Disabling this define allows you to shut off a few checks in
  61  * cothread_switch.  This likely will speed things up fractionally */
  62 #define COTHREAD_PARANOID
  63
  64 /**
  65  * cothread_context_init:
  66  *
  67  * Create and initialize a new cothread context
  68  *
  69  * Returns: the new cothread context
  70  */
  71 cothread_context *
  72 cothread_context_init (void)
  73 {
  74   cothread_context *ctx = (cothread_context *) g_malloc (sizeof (cothread_context));
  75
  76   /* we consider the initiating process to be cothread 0 */
  77   ctx->nthreads = 1;
  78   ctx->current = 0;
  79   ctx->data = g_hash_table_new (g_str_hash, g_str_equal);
  80
  81   GST_INFO (GST_CAT_COTHREADS, "initializing cothreads");
  82
  83   if (_cothread_key == (pthread_key_t)-1) {
  84     if (pthread_key_create (&_cothread_key, NULL) != 0) {
  85       perror ("pthread_key_create");
  86       return NULL;
  87     }
  88   }
  89   pthread_setspecific (_cothread_key, ctx);
  90
  91   memset (ctx->threads, 0, sizeof (ctx->threads));
  92
  93   ctx->threads[0] = (cothread_state *) g_malloc0 (sizeof (cothread_state));
  94   ctx->threads[0]->ctx = ctx;
  95   ctx->threads[0]->threadnum = 0;
  96   ctx->threads[0]->func = NULL;
  97   ctx->threads[0]->argc = 0;
  98   ctx->threads[0]->argv = NULL;
  99   ctx->threads[0]->priv = NULL;
 100   ctx->threads[0]->flags = COTHREAD_STARTED;
 101   ctx->threads[0]->sp = (void *) CURRENT_STACK_FRAME;
 102   ctx->threads[0]->pc = 0;
 103
 104   /* initialize the lock */
 105 #ifdef COTHREAD_ATOMIC
 106   atomic_set (&ctx->threads[0]->lock, 0);
 107 #else
 108   ctx->threads[0]->lock = g_mutex_new ();
 109 #endif
 110
 111   GST_INFO (GST_CAT_COTHREADS, "0th thread is %p at sp:%p", ctx->threads[0], ctx->threads[0]->sp);
 112
 113   return ctx;
 114 }
 115
 116 /**
 117  * cothread_context_free:
 118  * @ctx: the cothread context to free
 119  *
 120  * Free the cothread context.
 121  */
 122 void
 123 cothread_context_free (cothread_context *ctx)
 124 {
 125   gint i;
 126
 127   g_return_if_fail (ctx != NULL);
 128
 129   GST_INFO (GST_CAT_COTHREADS, "free cothread context");
 130
 131   for (i = 0; i < COTHREAD_MAXTHREADS; i++) {
 132     if (ctx->threads[i]) {
 133       cothread_destroy (ctx->threads[i]);
 134     }
 135   }
 136   g_hash_table_destroy (ctx->data);
 137   g_free (ctx);
 138 }
 139
 140 /**
 141  * cothread_create:
 142  * @ctx: the cothread context
 143  *
 144  * Create a new cothread state in the given context
 145  *
 146  * Returns: the new cothread state or NULL on error
 147  */
 148 cothread_state*
 149 cothread_create (cothread_context *ctx)
 150 {
 151   cothread_state *thread;
 152   void *sp;
 153   void *mmaped = 0;
 154   guchar *stack_end;
 155   gint slot = 0;
 156
 157   g_return_val_if_fail (ctx != NULL, NULL);
 158
 159   if (ctx->nthreads == COTHREAD_MAXTHREADS) {
 160     /* this is pretty fatal */
 161     g_warning ("cothread_create: attempt to create > COTHREAD_MAXTHREADS\n");
 162     return NULL;
 163   }
 164   /* find a free spot in the stack, note slot 0 has the main thread */
 165   for (slot = 1; slot < ctx->nthreads; slot++) {
 166     if (ctx->threads[slot] == NULL)
 167       break;
 168     else if (ctx->threads[slot]->flags & COTHREAD_DESTROYED &&
 169                     slot != ctx->current) {
 170       cothread_destroy (ctx->threads[slot]);
 171       break;
 172     }
 173   }
 174
 175   GST_DEBUG(GST_CAT_COTHREADS, "Found free cothread slot %d", slot);
 176
 177   sp = CURRENT_STACK_FRAME;
 178   /* FIXME this may not be 64bit clean
 179    *       could use casts to uintptr_t from inttypes.h
 180    *       if only all platforms had inttypes.h
 181    */
 182   stack_end = (guchar *) ((gulong) sp & ~(STACK_SIZE - 1));
 183
 184   thread = (cothread_state *) (stack_end + ((slot - 1) * COTHREAD_STACKSIZE));
 185   GST_DEBUG (GST_CAT_COTHREADS,
 186              "mmap   cothread slot stack from %p to %p (size 0x%lx)",
 187              thread, thread + COTHREAD_STACKSIZE,
 188              (long) COTHREAD_STACKSIZE);
 189
 190   GST_DEBUG (GST_CAT_COTHREADS, "going into mmap");
 191   /* the mmap is used to reserve part of the stack
 192    * ie. we state explicitly that we are going to use it */
 193   mmaped = mmap ((void *) thread, COTHREAD_STACKSIZE,
 194                   PROT_READ | PROT_WRITE | PROT_EXEC,
 195                   MAP_FIXED | MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
 196   GST_DEBUG (GST_CAT_COTHREADS, "coming out of mmap");
 197   if (mmaped == MAP_FAILED) {
 198     perror ("mmap'ing cothread stack space");
 199     return NULL;
 200   }
 201   if (mmaped != thread) {
 202     g_warning ("could not mmap requested memory");
 203     return NULL;
 204   }
 205
 206   thread->magic_number = COTHREAD_MAGIC_NUMBER;
 207   GST_DEBUG (GST_CAT_COTHREADS, "create  cothread %d with magic number 0x%x",
 208              slot, thread->magic_number);
 209   thread->ctx = ctx;
 210   thread->threadnum = slot;
 211   thread->flags = 0;
 212   thread->priv = NULL;
 213   thread->sp = ((guchar *) thread + COTHREAD_STACKSIZE);
 214   thread->top_sp = thread->sp; /* for debugging purposes to detect stack overruns */
 215
 216   /* initialize the lock */
 217 #ifdef COTHREAD_ATOMIC
 218   atomic_set (thread->lock, 0);
 219 #else
 220   thread->lock = g_mutex_new ();
 221 #endif
 222
 223   GST_INFO (GST_CAT_COTHREADS,
 224             "created cothread #%d in slot %d: %p at sp:%p lock:%p",
 225             ctx->nthreads, slot, thread, thread->sp, thread->lock);
 226
 227   ctx->threads[slot] = thread;
 228   ctx->nthreads++;
 229
 230   return thread;
 231 }
 232
 233 /**
 234  * cothread_free:
 235  * @thread: the cothread state
 236  *
 237  * Free the given cothread state
 238  */
 239 void
 240 cothread_free (cothread_state *thread)
 241 {
 242   g_return_if_fail (thread != NULL);
 243
 244   GST_INFO (GST_CAT_COTHREADS, "flag cothread %d for destruction", thread->threadnum);
 245
 246   /* we simply flag the cothread for destruction here */
 247   thread->flags |= COTHREAD_DESTROYED;
 248 }
 249
 250 static void
 251 cothread_destroy (cothread_state *thread)
 252 {
 253   cothread_context *ctx;
 254   gint threadnum;
 255
 256   g_return_if_fail (thread != NULL);
 257
 258   threadnum = thread->threadnum;
 259   ctx = thread->ctx;
 260
 261   GST_INFO (GST_CAT_COTHREADS, "destroy cothread %d %p %d", threadnum, thread, ctx->current);
 262
 263   /* we have to unlock here because we might be switched out with the lock held */
 264   cothread_unlock (thread);
 265
 266 #ifndef COTHREAD_ATOMIC
 267   g_mutex_free (thread->lock);
 268 #endif
 269
 270   if (threadnum == 0)
 271   {
 272     GST_INFO (GST_CAT_COTHREADS,
 273               "trying to destroy cothread 0 with %d cothreads left",
 274              ctx->nthreads);
 275     if (ctx->nthreads > 1)
 276     {
 277       /* we're trying to destroy cothread 0 when there are still cothreads
 278        * active, so kill those first */
 279       int i;
 280
 281       for (i = 1; i < COTHREAD_MAXTHREADS; ++i)
 282       {
 283         if (ctx->threads[i] != NULL)
 284         {
 285           cothread_destroy (ctx->threads[i]);
 286           GST_INFO (GST_CAT_COTHREADS,
 287                     "destroyed cothread %d, %d cothreads left\n",
 288                     i, ctx->nthreads);
 289         }
 290       }
 291     }
 292     g_assert (ctx->nthreads == 1);
 293     g_free (thread);
 294   }
 295   else {
 296     int res;
 297
 298     /* doing cleanups of the cothread create */
 299     GST_DEBUG (GST_CAT_COTHREADS, "destroy cothread %d with magic number 0x%x",
 300              threadnum, thread->magic_number);
 301     g_assert (thread->magic_number == COTHREAD_MAGIC_NUMBER);
 302
 303     g_assert (thread->priv == NULL);
 304     g_assert (thread->lock != NULL);
 305
 306     /* FIXME: I'm pretty sure we have to do something to the lock, no ? */
 307 #ifdef COTHREAD_ATOMIC
 308     /* FIXME: I don't think we need to do anything to an atomic lock */
 309 #else
 310     //g_mutex_free (thread->lock);
 311     //thread->lock = NULL;
 312 #endif
 313
 314     GST_DEBUG (GST_CAT_COTHREADS,
 315                "munmap cothread slot stack from %p to %p (size 0x%lx)",
 316                thread, thread + COTHREAD_STACKSIZE,
 317                (long) COTHREAD_STACKSIZE);
 318     res = munmap (thread, COTHREAD_STACKSIZE);
 319     if (res != 0)
 320     {
 321       switch (res)
 322       {
 323         case EINVAL:
 324           g_warning ("munmap doesn't like start %p or length %d\n",
 325                      thread, COTHREAD_STACKSIZE);
 326           break;
 327         default:
 328           g_warning ("Thomas was too lazy to check for all errors, so I can't tell you what is wrong.\n");
 329           break;
 330       }
 331     }
 332   }
 333
 334   ctx->threads[threadnum] = NULL;
 335   ctx->nthreads--;
 336 }
 337
 338 /**
 339  * cothread_setfunc:
 340  * @thread: the cothread state
 341  * @func: the function to call
 342  * @argc: argument count for the cothread function
 343  * @argv: arguments for the cothread function
 344  *
 345  * Set the cothread function
 346  */
 347 void
 348 cothread_setfunc (cothread_state * thread, cothread_func func, int argc, char **argv)
 349 {
 350   thread->func = func;
 351   thread->argc = argc;
 352   thread->argv = argv;
 353   thread->pc = (void *) func;
 354 }
 355
 356 /**
 357  * cothread_stop:
 358  * @thread: the cothread to stop
 359  *
 360  * Stop the cothread and reset the stack and program counter.
 361  */
 362 void
 363 cothread_stop (cothread_state * thread)
 364 {
 365   thread->flags &= ~COTHREAD_STARTED;
 366   thread->pc = 0;
 367   thread->sp = thread->top_sp;
 368 }
 369
 370 /**
 371  * cothread_main:
 372  * @ctx: cothread context to find main thread of
 373  *
 374  * Get the main thread.
 375  *
 376  * Returns: the #cothread_state of the main (0th) thread
 377  */
 378 cothread_state *
 379 cothread_main (cothread_context * ctx)
 380 {
 381   GST_DEBUG (GST_CAT_COTHREADS, "returning %p, the 0th cothread", ctx->threads[0]);
 382   return ctx->threads[0];
 383 }
 384
 385 /**
 386  * cothread_current_main:
 387  *
 388  * Get the main thread in the current pthread.
 389  *
 390  * Returns: the #cothread_state of the main (0th) thread in the current pthread
 391  */
 392 cothread_state *
 393 cothread_current_main (void)
 394 {
 395   cothread_context *ctx = pthread_getspecific (_cothread_key);
 396
 397   return ctx->threads[0];
 398 }
 399
 400 /**
 401  * cothread_current:
 402  *
 403  * Get the currenttly executing cothread
 404  *
 405  * Returns: the #cothread_state of the current cothread
 406  */
 407 cothread_state *
 408 cothread_current (void)
 409 {
 410   cothread_context *ctx = pthread_getspecific (_cothread_key);
 411
 412   return ctx->threads[ctx->current];
 413 }
 414
 415 static void
 416 cothread_stub (void)
 417 {
 418   cothread_context *ctx = pthread_getspecific (_cothread_key);
 419   register cothread_state *thread = ctx->threads[ctx->current];
 420
 421   GST_DEBUG_ENTER ("");
 422
 423   thread->flags |= COTHREAD_STARTED;
 424 /*
 425  * ifdef COTHREAD_ATOMIC
 426  *   do something here to lock
 427  * else
 428  *  g_mutex_lock(thread->lock);
 429  * endif
 430  */
 431
 432   while (TRUE) {
 433     thread->func (thread->argc, thread->argv);
 434     /* we do this to avoid ever returning, we just switch to 0th thread */
 435     cothread_switch (cothread_main (ctx));
 436   }
 437   GST_DEBUG_LEAVE ("");
 438 }
 439
 440 /**
 441  * cothread_getcurrent:
 442  *
 443  * Get the current cothread id
 444  *
 445  * Returns: the current cothread id
 446  */
 447 int cothread_getcurrent (void) __attribute__ ((no_instrument_function));
 448 int
 449 cothread_getcurrent (void)
 450 {
 451   cothread_context *ctx = pthread_getspecific (_cothread_key);
 452
 453   if (!ctx)
 454     return -1;
 455   return ctx->current;
 456 }
 457
 458 /**
 459  * cothread_set_private:
 460  * @thread: the cothread state
 461  * @data: the data
 462  *
 463  * set private data for the cothread.
 464  */
 465 void
 466 cothread_set_private (cothread_state *thread, gpointer data)
 467 {
 468   thread->priv = data;
 469 }
 470
 471 /**
 472  * cothread_context_set_data:
 473  * @thread: the cothread state
 474  * @key: a key for the data
 475  * @data: the data
 476  *
 477  * adds data to a cothread
 478  */
 479 void
 480 cothread_context_set_data (cothread_state *thread, gchar *key, gpointer data)
 481 {
 482   cothread_context *ctx = pthread_getspecific (_cothread_key);
 483
 484   g_hash_table_insert (ctx->data, key, data);
 485 }
 486
 487 /**
 488  * cothread_get_private:
 489  * @thread: the cothread state
 490  *
 491  * get the private data from the cothread
 492  *
 493  * Returns: the private data of the cothread
 494  */
 495 gpointer
 496 cothread_get_private (cothread_state *thread)
 497 {
 498   return thread->priv;
 499 }
 500
 501 /**
 502  * cothread_context_get_data:
 503  * @thread: the cothread state
 504  * @key: a key for the data
 505  *
 506  * get data from the cothread
 507  *
 508  * Returns: the data associated with the key
 509  */
 510 gpointer
 511 cothread_context_get_data (cothread_state * thread, gchar * key)
 512 {
 513   cothread_context *ctx = pthread_getspecific (_cothread_key);
 514
 515   return g_hash_table_lookup (ctx->data, key);
 516 }
 517
 518 /**
 519  * cothreads_stackquery:
 520  * @stack: Will be set to point to the allocated stack location
 521  * @stacksize: Will be set to the size of the allocated stack
 522  *
 523  *  Returns: #TRUE on success, #FALSE otherwise.
 524  */
 525 gboolean
 526 cothread_stackquery (void **stack, glong* stacksize)
 527 {
 528   /* wingo says: use posix_memalign to allocate a 2M-aligned, 2M stack */
 529
 530   int retval = 0;
 531
 532   retval = posix_memalign (stack, STACK_SIZE, STACK_SIZE);
 533   if (retval != 0)
 534   {
 535     g_warning ("Could not posix_memalign stack !\n");
 536     if (retval == EINVAL)
 537       g_warning ("The alignment parameter %d was not a power of two !\n",
 538                  STACK_SIZE);
 539     if (retval == ENOMEM)
 540       g_warning ("Insufficient memory to allocate the request of %d !\n",
 541                  STACK_SIZE);
 542     *stacksize = 0;
 543     return FALSE;
 544   }
 545   GST_DEBUG (GST_CAT_THREAD, "have  posix_memalign at %p of size %d",
 546            (void *) *stack, STACK_SIZE);
 547   GST_DEBUG (GST_CAT_COTHREADS,
 548              "Got new cothread stack from %p to %p (size %ld)",
 549              *stack, *stack + STACK_SIZE - 1, (long) STACK_SIZE);
 550   *stacksize = STACK_SIZE;
 551   return TRUE;
 552 }
 553
 554 /**
 555  * cothread_switch:
 556  * @thread: cothread state to switch to
 557  *
 558  * Switches to the given cothread state
 559  */
 560 void
 561 cothread_switch (cothread_state * thread)
 562 {
 563   cothread_context *ctx;
 564   cothread_state *current;
 565   int enter;
 566
 567 #ifdef COTHREAD_PARANOID
 568   if (thread == NULL)
 569     goto nothread;
 570 #endif
 571   ctx = thread->ctx;
 572 #ifdef COTHREAD_PARANOID
 573   if (ctx == NULL)
 574     goto nocontext;
 575 #endif
 576
 577   current = ctx->threads[ctx->current];
 578 #ifdef COTHREAD_PARANOID
 579   if (current == NULL)
 580     goto nocurrent;
 581 #endif
 582   if (current == thread)
 583     goto selfswitch;
 584
 585   /* unlock the current thread, we're out of that context now */
 586 #ifdef COTHREAD_ATOMIC
 587   /* do something to unlock the cothread */
 588 #else
 589   g_mutex_unlock (current->lock);
 590 #endif
 591
 592   /* lock the next cothread before we even switch to it */
 593 #ifdef COTHREAD_ATOMIC
 594   /* do something to lock the cothread */
 595 #else
 596   g_mutex_lock (thread->lock);
 597 #endif
 598
 599   /* find the number of the thread to switch to */
 600   GST_INFO (GST_CAT_COTHREAD_SWITCH, "switching from cothread #%d to cothread #%d",
 601             ctx->current, thread->threadnum);
 602   ctx->current = thread->threadnum;
 603
 604   /* save the current stack pointer, frame pointer, and pc */
 605 #ifdef GST_ARCH_PRESETJMP
 606   GST_ARCH_PRESETJMP ();
 607 #endif
 608   enter = setjmp (current->jmp);
 609   if (enter != 0) {
 610     GST_DEBUG (GST_CAT_COTHREADS, "enter thread #%d %d %p<->%p (%d) %p", current->threadnum, enter,
 611                current->sp, current->top_sp, (char*)current->top_sp - (char*)current->sp, current->jmp);
 612     return;
 613   }
 614   GST_DEBUG (GST_CAT_COTHREADS, "exit thread #%d %d %p<->%p (%d) %p", current->threadnum, enter,
 615                current->sp, current->top_sp, (char*)current->top_sp - (char*)current->sp, current->jmp);
 616   enter = 1;
 617
 618   if (current->flags & COTHREAD_DESTROYED) {
 619     cothread_destroy (current);
 620   }
 621
 622   GST_DEBUG (GST_CAT_COTHREADS, "set stack to %p", thread->sp);
 623   /* restore stack pointer and other stuff of new cothread */
 624   if (thread->flags & COTHREAD_STARTED) {
 625     GST_DEBUG (GST_CAT_COTHREADS, "in thread %p", thread->jmp);
 626     /* switch to it */
 627     longjmp (thread->jmp, 1);
 628   }
 629   else {
 630     GST_ARCH_SETUP_STACK ((char*)thread->sp);
 631     GST_ARCH_SET_SP (thread->sp);
 632     /* start it */
 633     GST_ARCH_CALL (cothread_stub);
 634     GST_DEBUG (GST_CAT_COTHREADS, "exit thread ");
 635     ctx->current = 0;
 636   }
 637
 638   return;
 639
 640 #ifdef COTHREAD_PARANOID
 641 nothread:
 642   g_print ("cothread: can't switch to NULL cothread!\n");
 643   return;
 644 nocontext:
 645   g_print ("cothread: there's no context, help!\n");
 646   exit (2);
 647 nocurrent:
 648   g_print ("cothread: there's no current thread, help!\n");
 649   exit (2);
 650 #endif /* COTHREAD_PARANOID */
 651 selfswitch:
 652   g_print ("cothread: trying to switch to same thread, legal but not necessary\n");
 653   return;
 654 }
 655
 656 /**
 657  * cothread_lock:
 658  * @thread: cothread state to lock
 659  *
 660  * Locks the cothread state.
 661  */
 662 void
 663 cothread_lock (cothread_state * thread)
 664 {
 665 #ifdef COTHREAD_ATOMIC
 666   /* do something to lock the cothread */
 667 #else
 668   if (thread->lock)
 669     g_mutex_lock (thread->lock);
 670 #endif
 671 }
 672
 673 /**
 674  * cothread_trylock:
 675  * @thread: cothread state to try to lock
 676  *
 677  * Try to lock the cothread state
 678  *
 679  * Returns: TRUE if the cothread could be locked.
 680  */
 681 gboolean
 682 cothread_trylock (cothread_state * thread)
 683 {
 684 #ifdef COTHREAD_ATOMIC
 685   /* do something to try to lock the cothread */
 686 #else
 687   if (thread->lock)
 688     return g_mutex_trylock (thread->lock);
 689   else
 690     return FALSE;
 691 #endif
 692 }
 693
 694 /**
 695  * cothread_unlock:
 696  * @thread: cothread state to unlock
 697  *
 698  * Unlock the cothread state.
 699  */
 700 void
 701 cothread_unlock (cothread_state * thread)
 702 {
 703 #ifdef COTHREAD_ATOMIC
 704   /* do something to unlock the cothread */
 705 #else
 706   if (thread->lock)
 707     g_mutex_unlock (thread->lock);
 708 #endif
 709 }