/* GStreamer
- * Copyright (C) <1999> Erik Walthinsen <omega@cse.ogi.edu>
+ * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
+ * 2000 Wim Taymans <wtay@chello.be>
+ *
+ * cothreads.c: Cothreading routines
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
*/
#include <pthread.h>
-#include <sys/time.h>
-#include <linux/linkage.h>
#include <stdio.h>
#include <stdlib.h>
#include <signal.h>
#include <sys/mman.h>
/* we make too much noise for normal debugging... */
-#define GST_DEBUG_FORCE_DISABLE
+/* #define GST_DEBUG_FORCE_DISABLE */
+#include "gst_private.h"
+
+#include "cothreads.h"
+#include "gstarch.h"
+
+
+#define STACK_SIZE 0x200000
+
+#define COTHREAD_MAXTHREADS 16
+#define COTHREAD_STACKSIZE (STACK_SIZE/COTHREAD_MAXTHREADS)
+
+struct _cothread_context {
+ cothread_state *threads[COTHREAD_MAXTHREADS];
+ int nthreads;
+ int current;
+ GHashTable *data;
+};
-#include <gst/gst.h>
-#include <gst/cothreads.h>
-#include <gst/gstarch.h>
pthread_key_t _cothread_key = -1;
#define COTHREAD_PARANOID
/**
+ * cothread_init:
+ *
+ * Create and initialize a new cothread context
+ *
+ * Returns: the new cothread context
+ */
+cothread_context*
+cothread_init (void)
+{
+ cothread_context *ctx = (cothread_context *)malloc(sizeof(cothread_context));
+
+ /* we consider the initiating process to be cothread 0 */
+ ctx->nthreads = 1;
+ ctx->current = 0;
+ ctx->data = g_hash_table_new(g_str_hash, g_str_equal);
+
+ GST_INFO (GST_CAT_COTHREADS,"initializing cothreads");
+
+ if (_cothread_key == -1) {
+ if (pthread_key_create (&_cothread_key,NULL) != 0) {
+ perror ("pthread_key_create");
+ return NULL;
+ }
+ }
+ pthread_setspecific (_cothread_key,ctx);
+
+ memset (ctx->threads,0,sizeof(ctx->threads));
+
+ ctx->threads[0] = (cothread_state *)malloc(sizeof(cothread_state));
+ ctx->threads[0]->ctx = ctx;
+ ctx->threads[0]->threadnum = 0;
+ ctx->threads[0]->func = NULL;
+ ctx->threads[0]->argc = 0;
+ ctx->threads[0]->argv = NULL;
+ ctx->threads[0]->flags = COTHREAD_STARTED;
+ ctx->threads[0]->sp = (void *)CURRENT_STACK_FRAME;
+ ctx->threads[0]->pc = 0;
+
+ /* initialize the lock */
+#ifdef COTHREAD_ATOMIC
+ atomic_set (&ctx->threads[0]->lock, 0);
+#else
+ ctx->threads[0]->lock = g_mutex_new();
+#endif
+
+ GST_INFO (GST_CAT_COTHREADS,"0th thread is %p at sp:%p",ctx->threads[0], ctx->threads[0]->sp);
+
+ return ctx;
+}
+
+void
+cothread_free (cothread_context *ctx)
+{
+ gint i;
+
+ for (i=0; i<ctx->nthreads; i++) {
+#ifndef COTHREAD_ATOMIC
+ if (ctx->threads[i]->lock) {
+ g_mutex_unlock(ctx->threads[i]->lock);
+ g_mutex_free (ctx->threads[i]->lock);
+ ctx->threads[i]->lock = NULL;
+ }
+#endif
+ if (i == 0) {
+ g_free (ctx->threads[i]);
+ ctx->threads[i] = NULL;
+ }
+ }
+ g_hash_table_destroy (ctx->data);
+ g_free (ctx);
+}
+
+/**
* cothread_create:
* @ctx: the cothread context
*
- * create a new cotread state in the given context
+ * Create a new cothread state in the given context
*
- * Returns: the new cothread state
+ * Returns: the new cothread state or NULL on error
*/
cothread_state*
cothread_create (cothread_context *ctx)
{
cothread_state *s;
- DEBUG("pthread_self() %ld\n",pthread_self());
- //if (0) {
- if (pthread_self() == 0) {
- s = (cothread_state *)malloc(sizeof(int) * COTHREAD_STACKSIZE);
- DEBUG("new stack at %p\n",s);
+ if (ctx->nthreads == COTHREAD_MAXTHREADS) {
+ GST_DEBUG (0, "attempt to create > COTHREAD_MAXTHREADS\n");
+ return NULL;
+ }
+ GST_DEBUG (0,"pthread_self() %ld\n",pthread_self());
+ /* if (0) { */
+ if (pthread_self() == 0) { /* FIXME uh, what does this test really do? */
+ s = (cothread_state *)malloc(COTHREAD_STACKSIZE);
+ GST_DEBUG (0,"new stack (case 1) at %p\n",s);
} else {
- char *sp = CURRENT_STACK_FRAME;
- unsigned long *stack_end = (unsigned long *)((unsigned long)sp &
- ~(STACK_SIZE - 1));
+ void *sp = CURRENT_STACK_FRAME;
+ /* FIXME this may not be 64bit clean
+ * could use casts to uintptr_t from inttypes.h
+ * if only all platforms had inttypes.h
+ */
+ guchar *stack_end = (guchar *)((unsigned long)sp & ~(STACK_SIZE - 1));
s = (cothread_state *)(stack_end + ((ctx->nthreads - 1) *
COTHREAD_STACKSIZE));
- if (mmap((char *)s,COTHREAD_STACKSIZE*(sizeof(int)),
- PROT_READ|PROT_WRITE|PROT_EXEC,MAP_PRIVATE|MAP_ANONYMOUS,
+ GST_DEBUG (0,"new stack (case 2) at %p\n",s);
+ if (mmap((void *)s,COTHREAD_STACKSIZE,
+ PROT_READ|PROT_WRITE|PROT_EXEC,MAP_FIXED|MAP_PRIVATE|MAP_ANON,
-1,0) < 0) {
perror("mmap'ing cothread stack space");
return NULL;
s->ctx = ctx;
s->threadnum = ctx->nthreads;
s->flags = 0;
- s->sp = ((int *)s + COTHREAD_STACKSIZE);
+ s->sp = ((guchar *)s + COTHREAD_STACKSIZE);
+ /* is this needed anymore? */
s->top_sp = s->sp;
- ctx->threads[ctx->nthreads++] = s;
+ /* initialize the lock */
+#ifdef COTHREAD_ATOMIC
+ atomic_set (s->lock, 0);
+#else
+ s->lock = g_mutex_new();
+#endif
- DEBUG("created cothread at %p %p\n",s, s->sp);
+ GST_INFO (GST_CAT_COTHREADS,"created cothread #%d: %p at sp:%p lock:%p", ctx->nthreads,
+ s, s->sp, s->lock);
+
+ ctx->threads[ctx->nthreads++] = s;
return s;
}
* cothread_setfunc:
* @thread: the cothread state
* @func: the function to call
- * @argc: the argument count for the cothread function
- * @argv: the arguments for the cothread function
+ * @argc: argument count for the cothread function
+ * @argv: arguments for the cothread function
*
* Set the cothread function
*/
thread->func = func;
thread->argc = argc;
thread->argv = argv;
- thread->pc = (int *)func;
+ thread->pc = (void *)func;
}
/**
- * cothread_init:
+ * cothread_main:
+ * @ctx: cothread context to find main thread of
*
- * create and initialize a new cotread context
+ * Get the main thread.
*
- * Returns: the new cothread context
+ * Returns: the #cothread_state of the main (0th) thread
*/
-cothread_context*
-cothread_init (void)
+cothread_state*
+cothread_main (cothread_context *ctx)
{
- cothread_context *ctx = (cothread_context *)malloc(sizeof(cothread_context));
-
- if (_cothread_key == -1) {
- if (pthread_key_create (&_cothread_key,NULL) != 0) {
- perror ("pthread_key_create");
- return NULL;
- }
- }
- pthread_setspecific (_cothread_key,ctx);
-
- memset (ctx->threads,0,sizeof(ctx->threads));
-
- ctx->threads[0] = (cothread_state *)malloc(sizeof(cothread_state));
- ctx->threads[0]->ctx = ctx;
- ctx->threads[0]->threadnum = 0;
- ctx->threads[0]->func = NULL;
- ctx->threads[0]->argc = 0;
- ctx->threads[0]->argv = NULL;
- ctx->threads[0]->flags = COTHREAD_STARTED;
- ctx->threads[0]->sp = (int *)CURRENT_STACK_FRAME;
- ctx->threads[0]->pc = 0;
-
- DEBUG("0th thread is at %p %p\n",ctx->threads[0], ctx->threads[0]->sp);
-
- // we consider the initiating process to be cothread 0
- ctx->nthreads = 1;
- ctx->current = 0;
- ctx->data = g_hash_table_new(g_str_hash, g_str_equal);
-
- return ctx;
+ GST_DEBUG (0,"returning %p, the 0th cothread\n",ctx->threads[0]);
+ return ctx->threads[0];
}
/**
- * cothread_main:
- * @ctx: the cothread context
+ * cothread_current_main:
*
- * Returns: the new cothread state
+ * Get the main thread in the current pthread.
+ *
+ * Returns: the #cothread_state of the main (0th) thread in the current pthread
*/
cothread_state*
-cothread_main(cothread_context *ctx)
+cothread_current_main (void)
{
- DEBUG("returning %p, the 0th cothread\n",ctx->threads[0]);
+ cothread_context *ctx = pthread_getspecific(_cothread_key);
return ctx->threads[0];
}
-void
+static void
cothread_stub (void)
{
cothread_context *ctx = pthread_getspecific(_cothread_key);
register cothread_state *thread = ctx->threads[ctx->current];
- DEBUG_ENTER("");
+ GST_DEBUG_ENTER("");
+
thread->flags |= COTHREAD_STARTED;
- if (thread->func)
+/* #ifdef COTHREAD_ATOMIC
+ * do something here to lock
+ * #else
+ * g_mutex_lock(thread->lock);
+ * #endif
+ */
+ while (1) {
thread->func(thread->argc,thread->argv);
+ /* we do this to avoid ever returning, we just switch to 0th thread */
+ cothread_switch(cothread_main(ctx));
+ }
thread->flags &= ~COTHREAD_STARTED;
thread->pc = 0;
thread->sp = thread->top_sp;
- DEBUG_LEAVE("");
-// printf("uh, yeah, we shouldn't be here, but we should deal anyway\n");
+ fprintf(stderr,"uh, yeah, we shouldn't be here, but we should deal anyway\n");
+ GST_DEBUG_LEAVE("");
}
/**
* cothread_getcurrent:
*
+ * Get the current cothread id
+ *
* Returns: the current cothread id
*/
-int cothread_getcurrent(void) {
+int cothread_getcurrent (void) __attribute__ ((no_instrument_function));
+int
+cothread_getcurrent (void)
+{
cothread_context *ctx = pthread_getspecific(_cothread_key);
if (!ctx) return -1;
return ctx->current;
/**
* cothread_switch:
- * @thread: the cothread state
+ * @thread: cothread state to switch to
*
- * switches to the given cothread state
+ * Switches to the given cothread state
*/
void
cothread_switch (cothread_state *thread)
#endif
if (current == thread) goto selfswitch;
- // find the number of the thread to switch to
+ /* unlock the current thread, we're out of that context now */
+#ifdef COTHREAD_ATOMIC
+ /* do something to unlock the cothread */
+#else
+ g_mutex_unlock(current->lock);
+#endif
+
+ /* lock the next cothread before we even switch to it */
+#ifdef COTHREAD_ATOMIC
+ /* do something to lock the cothread */
+#else
+ g_mutex_lock(thread->lock);
+#endif
+
+ /* find the number of the thread to switch to */
+ GST_INFO (GST_CAT_COTHREAD_SWITCH,"switching from cothread #%d to cothread #%d",
+ ctx->current,thread->threadnum);
ctx->current = thread->threadnum;
- DEBUG("about to switch to thread #%d\n",ctx->current);
/* save the current stack pointer, frame pointer, and pc */
- GET_SP(current->sp);
- enter = setjmp(current->jmp);
+#ifdef GST_ARCH_PRESETJMP
+ GST_ARCH_PRESETJMP();
+#endif
+ enter = sigsetjmp(current->jmp, 1);
if (enter != 0) {
- DEBUG("enter thread #%d %d %p<->%p (%d)\n",current->threadnum, enter,
+ GST_DEBUG (0,"enter thread #%d %d %p<->%p (%d)\n",current->threadnum, enter,
current->sp, current->top_sp, current->top_sp-current->sp);
return;
}
- DEBUG("exit thread #%d %d %p<->%p (%d)\n",current->threadnum, enter,
+ GST_DEBUG (0,"exit thread #%d %d %p<->%p (%d)\n",current->threadnum, enter,
current->sp, current->top_sp, current->top_sp-current->sp);
enter = 1;
- DEBUG("set stack to %p\n", thread->sp);
+ GST_DEBUG (0,"set stack to %p\n", thread->sp);
/* restore stack pointer and other stuff of new cothread */
if (thread->flags & COTHREAD_STARTED) {
- DEBUG("in thread \n");
- SET_SP(thread->sp);
- // switch to it
- longjmp(thread->jmp,1);
+ GST_DEBUG (0,"in thread \n");
+ /* switch to it */
+ siglongjmp(thread->jmp,1);
} else {
- SETUP_STACK(thread->sp);
- SET_SP(thread->sp);
- // start it
- cothread_stub();
- DEBUG("exit thread \n");
+ GST_ARCH_SETUP_STACK(thread->sp);
+ GST_ARCH_SET_SP(thread->sp);
+ /* start it */
+ GST_ARCH_CALL(cothread_stub);
+ GST_DEBUG (0,"exit thread \n");
ctx->current = 0;
}
#ifdef COTHREAD_PARANOID
nothread:
- g_print("cothread: there's no thread, strange...\n");
+ g_print("cothread: can't switch to NULL cothread!\n");
return;
nocontext:
g_print("cothread: there's no context, help!\n");
g_print("cothread: trying to switch to same thread, legal but not necessary\n");
return;
}
+
+/**
+ * cothread_lock:
+ * @thread: cothread state to lock
+ *
+ * Locks the cothread state.
+ */
+void
+cothread_lock (cothread_state *thread)
+{
+#ifdef COTHREAD_ATOMIC
+ /* do something to lock the cothread */
+#else
+ if (thread->lock)
+ g_mutex_lock(thread->lock);
+#endif
+}
+
+/**
+ * cothread_trylock:
+ * @thread: cothread state to try to lock
+ *
+ * Try to lock the cothread state
+ *
+ * Returns: TRUE if the cothread could be locked.
+ */
+gboolean
+cothread_trylock (cothread_state *thread)
+{
+#ifdef COTHREAD_ATOMIC
+ /* do something to try to lock the cothread */
+#else
+ if (thread->lock)
+ return g_mutex_trylock(thread->lock);
+ else
+ return FALSE;
+#endif
+}
+
+/**
+ * cothread_unlock:
+ * @thread: cothread state to unlock
+ *
+ * Unlock the cothread state.
+ */
+void
+cothread_unlock (cothread_state *thread)
+{
+#ifdef COTHREAD_ATOMIC
+ /* do something to unlock the cothread */
+#else
+ if (thread->lock)
+ g_mutex_unlock(thread->lock);
+#endif
+}
+
projects / platform / upstream / gstreamer.git / blobdiff