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 /* we make too much noise for normal debugging... */
  32 /* #define GST_DEBUG_FORCE_DISABLE */
  33 #include "gst_private.h"
  34
  35 #include "cothreads.h"
  36 #include "gstarch.h"
  37
  38
  39 #define STACK_SIZE 0x200000
  40
  41 #define COTHREAD_MAXTHREADS 16
  42 #define COTHREAD_STACKSIZE (STACK_SIZE/COTHREAD_MAXTHREADS)
  43
  44 struct _cothread_context {
  45   cothread_state *threads[COTHREAD_MAXTHREADS];
  46   int nthreads;
  47   int current;
  48   GHashTable *data;
  49 };
  50
  51
  52 pthread_key_t _cothread_key = -1;
  53
  54 /* Disablig this define allows you to shut off a few checks in
  55  * cothread_switch.  This likely will speed things up fractionally */
  56 #define COTHREAD_PARANOID
  57
  58 /**
  59  * cothread_init:
  60  *
  61  * Create and initialize a new cothread context
  62  *
  63  * Returns: the new cothread context
  64  */
  65 cothread_context*
  66 cothread_init (void)
  67 {
  68   cothread_context *ctx = (cothread_context *)malloc(sizeof(cothread_context));
  69
  70   /* we consider the initiating process to be cothread 0 */
  71   ctx->nthreads = 1;
  72   ctx->current = 0;
  73   ctx->data = g_hash_table_new(g_str_hash, g_str_equal);
  74
  75   GST_INFO (GST_CAT_COTHREADS,"initializing cothreads");
  76
  77   if (_cothread_key == -1) {
  78     if (pthread_key_create (&_cothread_key,NULL) != 0) {
  79       perror ("pthread_key_create");
  80       return NULL;
  81     }
  82   }
  83   pthread_setspecific (_cothread_key,ctx);
  84
  85   memset (ctx->threads,0,sizeof(ctx->threads));
  86
  87   ctx->threads[0] = (cothread_state *)malloc(sizeof(cothread_state));
  88   ctx->threads[0]->ctx = ctx;
  89   ctx->threads[0]->threadnum = 0;
  90   ctx->threads[0]->func = NULL;
  91   ctx->threads[0]->argc = 0;
  92   ctx->threads[0]->argv = NULL;
  93   ctx->threads[0]->flags = COTHREAD_STARTED;
  94   ctx->threads[0]->sp = (void *)CURRENT_STACK_FRAME;
  95   ctx->threads[0]->pc = 0;
  96
  97   /* initialize the lock */
  98 #ifdef COTHREAD_ATOMIC
  99   atomic_set (&ctx->threads[0]->lock, 0);
 100 #else
 101   ctx->threads[0]->lock = g_mutex_new();
 102 #endif
 103
 104   GST_INFO (GST_CAT_COTHREADS,"0th thread is %p at sp:%p",ctx->threads[0], ctx->threads[0]->sp);
 105
 106   return ctx;
 107 }
 108
 109 void
 110 cothread_free (cothread_context *ctx)
 111 {
 112   gint i;
 113
 114   for (i=0; i<ctx->nthreads; i++) {
 115 #ifndef COTHREAD_ATOMIC
 116     if (ctx->threads[i]->lock) {
 117       g_mutex_unlock(ctx->threads[i]->lock);
 118       g_mutex_free (ctx->threads[i]->lock);
 119       ctx->threads[i]->lock = NULL;
 120     }
 121 #endif
 122     if (i == 0) {
 123       g_free (ctx->threads[i]);
 124       ctx->threads[i] = NULL;
 125     }
 126   }
 127   g_hash_table_destroy (ctx->data);
 128   g_free (ctx);
 129 }
 130
 131 /**
 132  * cothread_create:
 133  * @ctx: the cothread context
 134  *
 135  * Create a new cothread state in the given context
 136  *
 137  * Returns: the new cothread state or NULL on error
 138  */
 139 cothread_state*
 140 cothread_create (cothread_context *ctx)
 141 {
 142   cothread_state *s;
 143
 144   if (ctx->nthreads == COTHREAD_MAXTHREADS) {
 145     GST_DEBUG (0, "attempt to create > COTHREAD_MAXTHREADS\n");
 146     return NULL;
 147   }
 148   GST_DEBUG (0,"pthread_self() %ld\n",pthread_self());
 149   /* if (0) { */
 150   if (pthread_self() == 0) {    /* FIXME uh, what does this test really do? */
 151     s = (cothread_state *)malloc(COTHREAD_STACKSIZE);
 152     GST_DEBUG (0,"new stack (case 1) at %p\n",s);
 153   } else {
 154     void *sp = CURRENT_STACK_FRAME;
 155     /* FIXME this may not be 64bit clean
 156      *       could use casts to uintptr_t from inttypes.h
 157      *       if only all platforms had inttypes.h
 158      */
 159     guchar *stack_end = (guchar *)((unsigned long)sp & ~(STACK_SIZE - 1));
 160     s = (cothread_state *)(stack_end + ((ctx->nthreads - 1) *
 161                            COTHREAD_STACKSIZE));
 162     GST_DEBUG (0,"new stack (case 2) at %p\n",s);
 163     if (mmap((void *)s,COTHREAD_STACKSIZE,
 164              PROT_READ|PROT_WRITE|PROT_EXEC,MAP_FIXED|MAP_PRIVATE|MAP_ANON,
 165              -1,0) < 0) {
 166       perror("mmap'ing cothread stack space");
 167       return NULL;
 168     }
 169   }
 170
 171   s->ctx = ctx;
 172   s->threadnum = ctx->nthreads;
 173   s->flags = 0;
 174   s->sp = ((guchar *)s + COTHREAD_STACKSIZE);
 175   /* is this needed anymore? */
 176   s->top_sp = s->sp;
 177
 178   /* initialize the lock */
 179 #ifdef COTHREAD_ATOMIC
 180   atomic_set (s->lock, 0);
 181 #else
 182   s->lock = g_mutex_new();
 183 #endif
 184
 185   GST_INFO (GST_CAT_COTHREADS,"created cothread #%d: %p at sp:%p lock:%p", ctx->nthreads,
 186                   s, s->sp, s->lock);
 187
 188   ctx->threads[ctx->nthreads++] = s;
 189
 190   return s;
 191 }
 192
 193 /**
 194  * cothread_setfunc:
 195  * @thread: the cothread state
 196  * @func: the function to call
 197  * @argc: argument count for the cothread function
 198  * @argv: arguments for the cothread function
 199  *
 200  * Set the cothread function
 201  */
 202 void
 203 cothread_setfunc (cothread_state *thread,
 204                   cothread_func func,
 205                   int argc,
 206                   char **argv)
 207 {
 208   thread->func = func;
 209   thread->argc = argc;
 210   thread->argv = argv;
 211   thread->pc = (void *)func;
 212 }
 213
 214 /**
 215  * cothread_main:
 216  * @ctx: cothread context to find main thread of
 217  *
 218  * Get the main thread.
 219  *
 220  * Returns: the #cothread_state of the main (0th) thread
 221  */
 222 cothread_state*
 223 cothread_main (cothread_context *ctx)
 224 {
 225   GST_DEBUG (0,"returning %p, the 0th cothread\n",ctx->threads[0]);
 226   return ctx->threads[0];
 227 }
 228
 229 /**
 230  * cothread_current_main:
 231  *
 232  * Get the main thread in the current pthread.
 233  *
 234  * Returns: the #cothread_state of the main (0th) thread in the current pthread
 235  */
 236 cothread_state*
 237 cothread_current_main (void)
 238 {
 239   cothread_context *ctx = pthread_getspecific(_cothread_key);
 240   return ctx->threads[0];
 241 }
 242
 243 static void
 244 cothread_stub (void)
 245 {
 246   cothread_context *ctx = pthread_getspecific(_cothread_key);
 247   register cothread_state *thread = ctx->threads[ctx->current];
 248
 249   GST_DEBUG_ENTER("");
 250
 251   thread->flags |= COTHREAD_STARTED;
 252 /* #ifdef COTHREAD_ATOMIC
 253  *   do something here to lock
 254  * #else
 255  *  g_mutex_lock(thread->lock);
 256  * #endif
 257  */
 258   while (1) {
 259     thread->func(thread->argc,thread->argv);
 260     /* we do this to avoid ever returning, we just switch to 0th thread */
 261     cothread_switch(cothread_main(ctx));
 262   }
 263   thread->flags &= ~COTHREAD_STARTED;
 264   thread->pc = 0;
 265   thread->sp = thread->top_sp;
 266   fprintf(stderr,"uh, yeah, we shouldn't be here, but we should deal anyway\n");
 267   GST_DEBUG_LEAVE("");
 268 }
 269
 270 /**
 271  * cothread_getcurrent:
 272  *
 273  * Get the current cothread id
 274  *
 275  * Returns: the current cothread id
 276  */
 277 int cothread_getcurrent (void) __attribute__ ((no_instrument_function));
 278 int
 279 cothread_getcurrent (void)
 280 {
 281   cothread_context *ctx = pthread_getspecific(_cothread_key);
 282   if (!ctx) return -1;
 283   return ctx->current;
 284 }
 285
 286 /**
 287  * cothread_set_data:
 288  * @thread: the cothread state
 289  * @key: a key for the data
 290  * @data: the data
 291  *
 292  * adds data to a cothread
 293  */
 294 void
 295 cothread_set_data (cothread_state *thread,
 296                    gchar *key,
 297                    gpointer data)
 298 {
 299   cothread_context *ctx = pthread_getspecific(_cothread_key);
 300
 301   g_hash_table_insert(ctx->data, key, data);
 302 }
 303
 304 /**
 305  * cothread_get_data:
 306  * @thread: the cothread state
 307  * @key: a key for the data
 308  *
 309  * get data from the cothread
 310  *
 311  * Returns: the data assiciated with the key
 312  */
 313 gpointer
 314 cothread_get_data (cothread_state *thread,
 315                    gchar *key)
 316 {
 317   cothread_context *ctx = pthread_getspecific(_cothread_key);
 318
 319   return g_hash_table_lookup(ctx->data, key);
 320 }
 321
 322 /**
 323  * cothread_switch:
 324  * @thread: cothread state to switch to
 325  *
 326  * Switches to the given cothread state
 327  */
 328 void
 329 cothread_switch (cothread_state *thread)
 330 {
 331   cothread_context *ctx;
 332   cothread_state *current;
 333   int enter;
 334
 335 #ifdef COTHREAD_PARANOID
 336   if (thread == NULL) goto nothread;
 337 #endif
 338   ctx = thread->ctx;
 339 #ifdef COTHREAD_PARANOID
 340   if (ctx == NULL) goto nocontext;
 341 #endif
 342
 343   current = ctx->threads[ctx->current];
 344 #ifdef COTHREAD_PARANOID
 345   if (current == NULL) goto nocurrent;
 346 #endif
 347   if (current == thread) goto selfswitch;
 348
 349   /* unlock the current thread, we're out of that context now */
 350 #ifdef COTHREAD_ATOMIC
 351   /* do something to unlock the cothread */
 352 #else
 353   g_mutex_unlock(current->lock);
 354 #endif
 355
 356   /* lock the next cothread before we even switch to it */
 357 #ifdef COTHREAD_ATOMIC
 358   /* do something to lock the cothread */
 359 #else
 360   g_mutex_lock(thread->lock);
 361 #endif
 362
 363   /* find the number of the thread to switch to */
 364   GST_INFO (GST_CAT_COTHREAD_SWITCH,"switching from cothread #%d to cothread #%d",
 365             ctx->current,thread->threadnum);
 366   ctx->current = thread->threadnum;
 367
 368   /* save the current stack pointer, frame pointer, and pc */
 369 #ifdef GST_ARCH_PRESETJMP
 370   GST_ARCH_PRESETJMP();
 371 #endif
 372   enter = sigsetjmp(current->jmp, 1);
 373   if (enter != 0) {
 374     GST_DEBUG (0,"enter thread #%d %d %p<->%p (%d)\n",current->threadnum, enter,
 375                     current->sp, current->top_sp, current->top_sp-current->sp);
 376     return;
 377   }
 378   GST_DEBUG (0,"exit thread #%d %d %p<->%p (%d)\n",current->threadnum, enter,
 379                     current->sp, current->top_sp, current->top_sp-current->sp);
 380   enter = 1;
 381
 382   GST_DEBUG (0,"set stack to %p\n", thread->sp);
 383   /* restore stack pointer and other stuff of new cothread */
 384   if (thread->flags & COTHREAD_STARTED) {
 385     GST_DEBUG (0,"in thread \n");
 386     /* switch to it */
 387     siglongjmp(thread->jmp,1);
 388   } else {
 389     GST_ARCH_SETUP_STACK(thread->sp);
 390     GST_ARCH_SET_SP(thread->sp);
 391     /* start it */
 392     GST_ARCH_CALL(cothread_stub);
 393     GST_DEBUG (0,"exit thread \n");
 394     ctx->current = 0;
 395   }
 396
 397   return;
 398
 399 #ifdef COTHREAD_PARANOID
 400 nothread:
 401   g_print("cothread: can't switch to NULL cothread!\n");
 402   return;
 403 nocontext:
 404   g_print("cothread: there's no context, help!\n");
 405   exit(2);
 406 nocurrent:
 407   g_print("cothread: there's no current thread, help!\n");
 408   exit(2);
 409 #endif /* COTHREAD_PARANOID */
 410 selfswitch:
 411   g_print("cothread: trying to switch to same thread, legal but not necessary\n");
 412   return;
 413 }
 414
 415 /**
 416  * cothread_lock:
 417  * @thread: cothread state to lock
 418  *
 419  * Locks the cothread state.
 420  */
 421 void
 422 cothread_lock (cothread_state *thread)
 423 {
 424 #ifdef COTHREAD_ATOMIC
 425   /* do something to lock the cothread */
 426 #else
 427   if (thread->lock)
 428     g_mutex_lock(thread->lock);
 429 #endif
 430 }
 431
 432 /**
 433  * cothread_trylock:
 434  * @thread: cothread state to try to lock
 435  *
 436  * Try to lock the cothread state
 437  *
 438  * Returns: TRUE if the cothread could be locked.
 439  */
 440 gboolean
 441 cothread_trylock (cothread_state *thread)
 442 {
 443 #ifdef COTHREAD_ATOMIC
 444   /* do something to try to lock the cothread */
 445 #else
 446   if (thread->lock)
 447     return g_mutex_trylock(thread->lock);
 448   else
 449     return FALSE;
 450 #endif
 451 }
 452
 453 /**
 454  * cothread_unlock:
 455  * @thread: cothread state to unlock
 456  *
 457  * Unlock the cothread state.
 458  */
 459 void
 460 cothread_unlock (cothread_state *thread)
 461 {
 462 #ifdef COTHREAD_ATOMIC
 463   /* do something to unlock the cothread */
 464 #else
 465   if (thread->lock)
 466     g_mutex_unlock(thread->lock);
 467 #endif
 468 }
 469