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 <sys/mman.h>
  30
  31 #include "gst_private.h"
  32
  33 #include "cothreads.h"
  34 #include "gstarch.h"
  35 #include "gstlog.h"
  36
  37
  38 #define STACK_SIZE 0x200000
  39
  40 #define COTHREAD_MAXTHREADS 16
  41 #define COTHREAD_STACKSIZE (STACK_SIZE/COTHREAD_MAXTHREADS)
  42
  43 static void     cothread_destroy        (cothread_state *thread);
  44
  45 struct _cothread_context
  46 {
  47   cothread_state *threads[COTHREAD_MAXTHREADS];
  48   int nthreads;
  49   int current;
  50   GHashTable *data;
  51 };
  52
  53
  54 static pthread_key_t _cothread_key = -1;
  55
  56 /* Disablig this define allows you to shut off a few checks in
  57  * cothread_switch.  This likely will speed things up fractionally */
  58 #define COTHREAD_PARANOID
  59
  60 /**
  61  * cothread_context_init:
  62  *
  63  * Create and initialize a new cothread context
  64  *
  65  * Returns: the new cothread context
  66  */
  67 cothread_context *
  68 cothread_context_init (void)
  69 {
  70   cothread_context *ctx = (cothread_context *) g_malloc (sizeof (cothread_context));
  71
  72   /* we consider the initiating process to be cothread 0 */
  73   ctx->nthreads = 1;
  74   ctx->current = 0;
  75   ctx->data = g_hash_table_new (g_str_hash, g_str_equal);
  76
  77   GST_INFO (GST_CAT_COTHREADS, "initializing cothreads");
  78
  79   if (_cothread_key == (pthread_key_t)-1) {
  80     if (pthread_key_create (&_cothread_key, NULL) != 0) {
  81       perror ("pthread_key_create");
  82       return NULL;
  83     }
  84   }
  85   pthread_setspecific (_cothread_key, ctx);
  86
  87   memset (ctx->threads, 0, sizeof (ctx->threads));
  88
  89   ctx->threads[0] = (cothread_state *) g_malloc (sizeof (cothread_state));
  90   ctx->threads[0]->ctx = ctx;
  91   ctx->threads[0]->threadnum = 0;
  92   ctx->threads[0]->func = NULL;
  93   ctx->threads[0]->argc = 0;
  94   ctx->threads[0]->argv = NULL;
  95   ctx->threads[0]->priv = NULL;
  96   ctx->threads[0]->flags = COTHREAD_STARTED;
  97   ctx->threads[0]->sp = (void *) CURRENT_STACK_FRAME;
  98   ctx->threads[0]->pc = 0;
  99
 100   /* initialize the lock */
 101 #ifdef COTHREAD_ATOMIC
 102   atomic_set (&ctx->threads[0]->lock, 0);
 103 #else
 104   ctx->threads[0]->lock = g_mutex_new ();
 105 #endif
 106
 107   GST_INFO (GST_CAT_COTHREADS, "0th thread is %p at sp:%p", ctx->threads[0], ctx->threads[0]->sp);
 108
 109   return ctx;
 110 }
 111
 112 /**
 113  * cothread_context_free:
 114  * @ctx: the cothread context to free
 115  *
 116  * Free the cothread context.
 117  */
 118 void
 119 cothread_context_free (cothread_context *ctx)
 120 {
 121   gint i;
 122
 123   g_return_if_fail (ctx != NULL);
 124
 125   GST_INFO (GST_CAT_COTHREADS, "free cothread context");
 126
 127   for (i = 0; i < COTHREAD_MAXTHREADS; i++) {
 128     if (ctx->threads[i]) {
 129       cothread_destroy (ctx->threads[i]);
 130     }
 131   }
 132   g_hash_table_destroy (ctx->data);
 133   g_free (ctx);
 134 }
 135
 136 /**
 137  * cothread_create:
 138  * @ctx: the cothread context
 139  *
 140  * Create a new cothread state in the given context
 141  *
 142  * Returns: the new cothread state or NULL on error
 143  */
 144 cothread_state*
 145 cothread_create (cothread_context *ctx)
 146 {
 147   cothread_state *thread;
 148   void *sp;
 149   guchar *stack_end;
 150   gint slot = 0;
 151
 152   g_return_val_if_fail (ctx != NULL, NULL);
 153
 154   if (ctx->nthreads == COTHREAD_MAXTHREADS) {
 155     /* this is pretty fatal */
 156     g_warning ("cothread_create: attempt to create > COTHREAD_MAXTHREADS\n");
 157     return NULL;
 158   }
 159   /* find a free spot in the stack, note slot 0 has the main thread */
 160   for (slot = 1; slot < ctx->nthreads; slot++) {
 161     if (ctx->threads[slot] == NULL)
 162       break;
 163     else if (ctx->threads[slot]->flags & COTHREAD_DESTROYED &&
 164                     slot != ctx->current) {
 165       cothread_destroy (ctx->threads[slot]);
 166       break;
 167     }
 168   }
 169
 170   sp = CURRENT_STACK_FRAME;
 171   /* FIXME this may not be 64bit clean
 172    *       could use casts to uintptr_t from inttypes.h
 173    *       if only all platforms had inttypes.h
 174    */
 175   stack_end = (guchar *) ((gulong) sp & ~(STACK_SIZE - 1));
 176
 177   thread = (cothread_state *) (stack_end + ((slot - 1) * COTHREAD_STACKSIZE));
 178   GST_DEBUG (0, "new stack at %p", thread);
 179
 180   if (mmap ((void *) thread, COTHREAD_STACKSIZE,
 181             PROT_READ | PROT_WRITE | PROT_EXEC, MAP_FIXED | MAP_PRIVATE | MAP_ANON, -1, 0) == MAP_FAILED) {
 182     perror ("mmap'ing cothread stack space");
 183     return NULL;
 184   }
 185
 186   thread->ctx = ctx;
 187   thread->threadnum = slot;
 188   thread->flags = 0;
 189   thread->priv = NULL;
 190   thread->sp = ((guchar *) thread + COTHREAD_STACKSIZE);
 191   thread->top_sp = thread->sp; /* for debugging purposes to detect stack overruns */
 192
 193   /* initialize the lock */
 194 #ifdef COTHREAD_ATOMIC
 195   atomic_set (thread->lock, 0);
 196 #else
 197   thread->lock = g_mutex_new ();
 198 #endif
 199
 200   GST_INFO (GST_CAT_COTHREADS, "created cothread #%d in slot %d: %p at sp:%p lock:%p",
 201                   ctx->nthreads, slot, thread, thread->sp, thread->lock);
 202
 203   ctx->threads[slot] = thread;
 204   ctx->nthreads++;
 205
 206   return thread;
 207 }
 208
 209 /**
 210  * cothread_free:
 211  * @thread: the cothread state
 212  *
 213  * Free the given cothread state
 214  */
 215 void
 216 cothread_free (cothread_state *thread)
 217 {
 218   g_return_if_fail (thread != NULL);
 219
 220   GST_INFO (GST_CAT_COTHREADS, "flag cothread %d for destruction", thread->threadnum);
 221
 222   /* we simply flag the cothread for destruction here */
 223   thread->flags |= COTHREAD_DESTROYED;
 224 }
 225
 226 static void
 227 cothread_destroy (cothread_state *thread)
 228 {
 229   cothread_context *ctx;
 230   gint threadnum;
 231
 232   g_return_if_fail (thread != NULL);
 233
 234   threadnum = thread->threadnum;
 235   ctx = thread->ctx;
 236
 237   GST_INFO (GST_CAT_COTHREADS, "destroy cothread %d %p %d", threadnum, thread, ctx->current);
 238
 239   /* we have to unlock here because we might be switched out with the lock held */
 240   cothread_unlock (thread);
 241
 242 #ifndef COTHREAD_ATOMIC
 243   g_mutex_free (thread->lock);
 244 #endif
 245
 246   if (threadnum == 0) {
 247     g_free (thread);
 248   }
 249   else {
 250     /* this doesn't seem to work very well */
 251     /* munmap ((void *) thread, COTHREAD_STACKSIZE); */
 252   }
 253
 254   ctx->threads[threadnum] = NULL;
 255   ctx->nthreads--;
 256 }
 257
 258 /**
 259  * cothread_setfunc:
 260  * @thread: the cothread state
 261  * @func: the function to call
 262  * @argc: argument count for the cothread function
 263  * @argv: arguments for the cothread function
 264  *
 265  * Set the cothread function
 266  */
 267 void
 268 cothread_setfunc (cothread_state * thread, cothread_func func, int argc, char **argv)
 269 {
 270   thread->func = func;
 271   thread->argc = argc;
 272   thread->argv = argv;
 273   thread->pc = (void *) func;
 274 }
 275
 276 /**
 277  * cothread_stop:
 278  * @thread: the cothread to stop
 279  *
 280  * Stop the cothread and reset the stack and program counter.
 281  */
 282 void
 283 cothread_stop (cothread_state * thread)
 284 {
 285   thread->flags &= ~COTHREAD_STARTED;
 286   thread->pc = 0;
 287   thread->sp = thread->top_sp;
 288 }
 289
 290 /**
 291  * cothread_main:
 292  * @ctx: cothread context to find main thread of
 293  *
 294  * Get the main thread.
 295  *
 296  * Returns: the #cothread_state of the main (0th) thread
 297  */
 298 cothread_state *
 299 cothread_main (cothread_context * ctx)
 300 {
 301   GST_DEBUG (0, "returning %p, the 0th cothread", ctx->threads[0]);
 302   return ctx->threads[0];
 303 }
 304
 305 /**
 306  * cothread_current_main:
 307  *
 308  * Get the main thread in the current pthread.
 309  *
 310  * Returns: the #cothread_state of the main (0th) thread in the current pthread
 311  */
 312 cothread_state *
 313 cothread_current_main (void)
 314 {
 315   cothread_context *ctx = pthread_getspecific (_cothread_key);
 316
 317   return ctx->threads[0];
 318 }
 319
 320 /**
 321  * cothread_current:
 322  *
 323  * Get the currenttly executing cothread
 324  *
 325  * Returns: the #cothread_state of the current cothread
 326  */
 327 cothread_state *
 328 cothread_current (void)
 329 {
 330   cothread_context *ctx = pthread_getspecific (_cothread_key);
 331
 332   return ctx->threads[ctx->current];
 333 }
 334
 335 static void
 336 cothread_stub (void)
 337 {
 338   cothread_context *ctx = pthread_getspecific (_cothread_key);
 339   register cothread_state *thread = ctx->threads[ctx->current];
 340
 341   GST_DEBUG_ENTER ("");
 342
 343   thread->flags |= COTHREAD_STARTED;
 344 /*
 345  * ifdef COTHREAD_ATOMIC
 346  *   do something here to lock
 347  * else
 348  *  g_mutex_lock(thread->lock);
 349  * endif
 350  */
 351
 352   while (TRUE) {
 353     thread->func (thread->argc, thread->argv);
 354     /* we do this to avoid ever returning, we just switch to 0th thread */
 355     cothread_switch (cothread_main (ctx));
 356   }
 357   GST_DEBUG_LEAVE ("");
 358 }
 359
 360 /**
 361  * cothread_getcurrent:
 362  *
 363  * Get the current cothread id
 364  *
 365  * Returns: the current cothread id
 366  */
 367 int cothread_getcurrent (void) __attribute__ ((no_instrument_function));
 368 int
 369 cothread_getcurrent (void)
 370 {
 371   cothread_context *ctx = pthread_getspecific (_cothread_key);
 372
 373   if (!ctx)
 374     return -1;
 375   return ctx->current;
 376 }
 377
 378 /**
 379  * cothread_set_private:
 380  * @thread: the cothread state
 381  * @data: the data
 382  *
 383  * set private data for the cothread.
 384  */
 385 void
 386 cothread_set_private (cothread_state *thread, gpointer data)
 387 {
 388   thread->priv = data;
 389 }
 390
 391 /**
 392  * cothread_context_set_data:
 393  * @thread: the cothread state
 394  * @key: a key for the data
 395  * @data: the data
 396  *
 397  * adds data to a cothread
 398  */
 399 void
 400 cothread_context_set_data (cothread_state *thread, gchar *key, gpointer data)
 401 {
 402   cothread_context *ctx = pthread_getspecific (_cothread_key);
 403
 404   g_hash_table_insert (ctx->data, key, data);
 405 }
 406
 407 /**
 408  * cothread_get_private:
 409  * @thread: the cothread state
 410  *
 411  * get the private data from the cothread
 412  *
 413  * Returns: the private data of the cothread
 414  */
 415 gpointer
 416 cothread_get_private (cothread_state *thread)
 417 {
 418   return thread->priv;
 419 }
 420
 421 /**
 422  * cothread_context_get_data:
 423  * @thread: the cothread state
 424  * @key: a key for the data
 425  *
 426  * get data from the cothread
 427  *
 428  * Returns: the data associated with the key
 429  */
 430 gpointer
 431 cothread_context_get_data (cothread_state * thread, gchar * key)
 432 {
 433   cothread_context *ctx = pthread_getspecific (_cothread_key);
 434
 435   return g_hash_table_lookup (ctx->data, key);
 436 }
 437
 438 /**
 439  * cothread_switch:
 440  * @thread: cothread state to switch to
 441  *
 442  * Switches to the given cothread state
 443  */
 444 void
 445 cothread_switch (cothread_state * thread)
 446 {
 447   cothread_context *ctx;
 448   cothread_state *current;
 449   int enter;
 450
 451 #ifdef COTHREAD_PARANOID
 452   if (thread == NULL)
 453     goto nothread;
 454 #endif
 455   ctx = thread->ctx;
 456 #ifdef COTHREAD_PARANOID
 457   if (ctx == NULL)
 458     goto nocontext;
 459 #endif
 460
 461   current = ctx->threads[ctx->current];
 462 #ifdef COTHREAD_PARANOID
 463   if (current == NULL)
 464     goto nocurrent;
 465 #endif
 466   if (current == thread)
 467     goto selfswitch;
 468
 469   /* unlock the current thread, we're out of that context now */
 470 #ifdef COTHREAD_ATOMIC
 471   /* do something to unlock the cothread */
 472 #else
 473   g_mutex_unlock (current->lock);
 474 #endif
 475
 476   /* lock the next cothread before we even switch to it */
 477 #ifdef COTHREAD_ATOMIC
 478   /* do something to lock the cothread */
 479 #else
 480   g_mutex_lock (thread->lock);
 481 #endif
 482
 483   /* find the number of the thread to switch to */
 484   GST_INFO (GST_CAT_COTHREAD_SWITCH, "switching from cothread #%d to cothread #%d",
 485             ctx->current, thread->threadnum);
 486   ctx->current = thread->threadnum;
 487
 488   /* save the current stack pointer, frame pointer, and pc */
 489 #ifdef GST_ARCH_PRESETJMP
 490   GST_ARCH_PRESETJMP ();
 491 #endif
 492   enter = setjmp (current->jmp);
 493   if (enter != 0) {
 494     GST_DEBUG (0, "enter thread #%d %d %p<->%p (%d)", current->threadnum, enter,
 495                current->sp, current->top_sp, (char*)current->top_sp - (char*)current->sp);
 496     return;
 497   }
 498   GST_DEBUG (0, "exit thread #%d %d %p<->%p (%d)", current->threadnum, enter,
 499                current->sp, current->top_sp, (char*)current->top_sp - (char*)current->sp);
 500   enter = 1;
 501
 502   if (current->flags & COTHREAD_DESTROYED)
 503     cothread_destroy (current);
 504
 505   GST_DEBUG (0, "set stack to %p", thread->sp);
 506   /* restore stack pointer and other stuff of new cothread */
 507   if (thread->flags & COTHREAD_STARTED) {
 508     GST_DEBUG (0, "in thread ");
 509     /* switch to it */
 510     longjmp (thread->jmp, 1);
 511   }
 512   else {
 513     GST_ARCH_SETUP_STACK ((char*)thread->sp);
 514     GST_ARCH_SET_SP (thread->sp);
 515     /* start it */
 516     GST_ARCH_CALL (cothread_stub);
 517     GST_DEBUG (0, "exit thread ");
 518     ctx->current = 0;
 519   }
 520
 521   return;
 522
 523 #ifdef COTHREAD_PARANOID
 524 nothread:
 525   g_print ("cothread: can't switch to NULL cothread!\n");
 526   return;
 527 nocontext:
 528   g_print ("cothread: there's no context, help!\n");
 529   exit (2);
 530 nocurrent:
 531   g_print ("cothread: there's no current thread, help!\n");
 532   exit (2);
 533 #endif /* COTHREAD_PARANOID */
 534 selfswitch:
 535   g_print ("cothread: trying to switch to same thread, legal but not necessary\n");
 536   return;
 537 }
 538
 539 /**
 540  * cothread_lock:
 541  * @thread: cothread state to lock
 542  *
 543  * Locks the cothread state.
 544  */
 545 void
 546 cothread_lock (cothread_state * thread)
 547 {
 548 #ifdef COTHREAD_ATOMIC
 549   /* do something to lock the cothread */
 550 #else
 551   if (thread->lock)
 552     g_mutex_lock (thread->lock);
 553 #endif
 554 }
 555
 556 /**
 557  * cothread_trylock:
 558  * @thread: cothread state to try to lock
 559  *
 560  * Try to lock the cothread state
 561  *
 562  * Returns: TRUE if the cothread could be locked.
 563  */
 564 gboolean
 565 cothread_trylock (cothread_state * thread)
 566 {
 567 #ifdef COTHREAD_ATOMIC
 568   /* do something to try to lock the cothread */
 569 #else
 570   if (thread->lock)
 571     return g_mutex_trylock (thread->lock);
 572   else
 573     return FALSE;
 574 #endif
 575 }
 576
 577 /**
 578  * cothread_unlock:
 579  * @thread: cothread state to unlock
 580  *
 581  * Unlock the cothread state.
 582  */
 583 void
 584 cothread_unlock (cothread_state * thread)
 585 {
 586 #ifdef COTHREAD_ATOMIC
 587   /* do something to unlock the cothread */
 588 #else
 589   if (thread->lock)
 590     g_mutex_unlock (thread->lock);
 591 #endif
 592 }