mirror/guile
mirror/guile
1
Fork 0
mirror of https://git.savannah.gnu.org/git/guile.git synced 2025-06-03 18:50:19 +02:00
Code Activity

Remove useless cooperative multi-threading source files.

Browse source
This commit is contained in:
Ludovic Courtès 2009-01-18 12:53:01 +01:00
parent 0208ec4013
commit 6bf4c3f9c1
3 changed files with 0 additions and 1789 deletions
Download patch file Download diff file Expand all files Collapse all files

943
libguile/coop-pthreads.c
View file

@ -1,943 +0,0 @@
/* Copyright (C) 1995,1996,1997,1998,2000,2001, 2002, 2006 Free Software Foundation, Inc.
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include "libguile/_scm.h" /* config.h, _scm.h, __scm.h should be first */
#include <unistd.h>
#include <stdio.h>
#include <assert.h>
#include <sys/time.h>
#include "libguile/validate.h"
#include "libguile/coop-pthreads.h"
#include "libguile/root.h"
#include "libguile/eval.h"
#include "libguile/async.h"
#include "libguile/ports.h"
#include "libguile/smob.h"
#undef DEBUG
/*** Queues */
static SCM
make_queue ()
{
return scm_cons (SCM_EOL, SCM_EOL);
}
static void
enqueue (SCM q, SCM t)
{
SCM c = scm_cons (t, SCM_EOL);
if (scm_is_null (SCM_CAR (q)))
SCM_SETCAR (q, c);
else
SCM_SETCDR (SCM_CDR (q), c);
SCM_SETCDR (q, c);
}
static SCM
dequeue (SCM q)
{
SCM c = SCM_CAR (q);
if (scm_is_null (c))
return SCM_BOOL_F;
else
{
SCM_SETCAR (q, SCM_CDR (c));
if (scm_is_null (SCM_CAR (q)))
SCM_SETCDR (q, SCM_EOL);
return SCM_CAR (c);
}
}
/*** Threads */
typedef struct scm_copt_thread {
/* A condition variable for sleeping on.
*/
pthread_cond_t sleep_cond;
/* A link for waiting queues.
*/
struct scm_copt_thread *next_waiting;
scm_root_state *root;
SCM handle;
pthread_t pthread;
SCM result;
SCM joining_threads;
/* For keeping track of the stack and registers. */
SCM_STACKITEM *base;
SCM_STACKITEM *top;
jmp_buf regs;
} scm_copt_thread;
static SCM
make_thread (SCM creation_protects)
{
SCM z;
scm_copt_thread *t = scm_gc_malloc (sizeof(*t), "thread");
SCM_NEWSMOB (z, t);
t->handle = z;
t->result = creation_protects;
t->base = NULL;
t->joining_threads = make_queue ();
pthread_cond_init (&t->sleep_cond, NULL);
return z;
}
static void
init_thread_creator (SCM thread, pthread_t th, scm_root_state *r)
{
scm_copt_thread *t = SCM_THREAD_DATA(thread);
t->root = r;
t->pthread = th;
#ifdef DEBUG
// fprintf (stderr, "%ld created %ld\n", pthread_self (), th);
#endif
}
static void
init_thread_creatant (SCM thread, SCM_STACKITEM *base)
{
scm_copt_thread *t = SCM_THREAD_DATA(thread);
t->base = base;
t->top = NULL;
}
static int
thread_print (SCM exp, SCM port, scm_print_state *pstate SCM_UNUSED)
{
scm_copt_thread *t = SCM_THREAD_DATA (exp);
scm_puts ("#<thread ", port);
scm_uintprint ((scm_t_bits)t, 16, port);
if (t->pthread != -1)
{
scm_putc (' ', port);
scm_intprint (t->pthread, 10, port);
}
else
scm_puts (" (exited)", port);
scm_putc ('>', port);
return 1;
}
static size_t
thread_free (SCM obj)
{
scm_copt_thread *t = SCM_THREAD_DATA (obj);
if (t->pthread != -1)
abort ();
scm_gc_free (t, sizeof (*t), "thread");
return 0;
}
/*** Fair mutexes */
/* POSIX mutexes are not necessarily fair but since we'd like to use a
mutex for scheduling, we build a fair one on top of POSIX.
*/
typedef struct fair_mutex {
pthread_mutex_t lock;
scm_copt_thread *owner;
scm_copt_thread *next_waiting, *last_waiting;
} fair_mutex;
static void
fair_mutex_init (fair_mutex *m)
{
pthread_mutex_init (&m->lock, NULL);
m->owner = NULL;
m->next_waiting = NULL;
m->last_waiting = NULL;
}
static void
fair_mutex_lock_1 (fair_mutex *m, scm_copt_thread *t)
{
if (m->owner == NULL)
m->owner = t;
else
{
t->next_waiting = NULL;
if (m->last_waiting)
m->last_waiting->next_waiting = t;
else
m->next_waiting = t;
m->last_waiting = t;
do
{
pthread_cond_wait (&t->sleep_cond, &m->lock);
}
while (m->owner != t);
assert (m->next_waiting == t);
m->next_waiting = t->next_waiting;
if (m->next_waiting == NULL)
m->last_waiting = NULL;
}
pthread_mutex_unlock (&m->lock);
}
static void
fair_mutex_lock (fair_mutex *m, scm_copt_thread *t)
{
pthread_mutex_lock (&m->lock);
fair_mutex_lock_1 (m, t);
}
static void
fair_mutex_unlock_1 (fair_mutex *m)
{
scm_copt_thread *t;
pthread_mutex_lock (&m->lock);
// fprintf (stderr, "%ld unlocking\n", m->owner->pthread);
if ((t = m->next_waiting) != NULL)
{
m->owner = t;
pthread_cond_signal (&t->sleep_cond);
}
else
m->owner = NULL;
// fprintf (stderr, "%ld unlocked\n", pthread_self ());
}
static void
fair_mutex_unlock (fair_mutex *m)
{
fair_mutex_unlock_1 (m);
pthread_mutex_unlock (&m->lock);
}
/* Temporarily give up the mutex. This function makes sure that we
are on the wait queue before starting the next thread. Otherwise
the next thread might preempt us and we will have a hard time
getting on the wait queue.
*/
#if 0
static void
fair_mutex_yield (fair_mutex *m)
{
scm_copt_thread *self, *next;
pthread_mutex_lock (&m->lock);
/* get next thread
*/
if ((next = m->next_waiting) == NULL)
{
/* No use giving it up. */
pthread_mutex_unlock (&m->lock);
return;
}
/* put us on queue
*/
self = m->owner;
self->next_waiting = NULL;
if (m->last_waiting)
m->last_waiting->next_waiting = self;
else
m->next_waiting = self;
m->last_waiting = self;
/* wake up next thread
*/
m->owner = next;
pthread_cond_signal (&next->sleep_cond);
/* wait for mutex
*/
do
{
pthread_cond_wait (&self->sleep_cond, &m->lock);
}
while (m->owner != self);
assert (m->next_waiting == self);
m->next_waiting = self->next_waiting;
if (m->next_waiting == NULL)
m->last_waiting = NULL;
pthread_mutex_unlock (&m->lock);
}
#else
static void
fair_mutex_yield (fair_mutex *m)
{
scm_copt_thread *self = m->owner;
fair_mutex_unlock_1 (m);
fair_mutex_lock_1 (m, self);
}
#endif
static void
fair_cond_wait (pthread_cond_t *c, fair_mutex *m)
{
scm_copt_thread *t = m->owner;
fair_mutex_unlock_1 (m);
pthread_cond_wait (c, &m->lock);
fair_mutex_lock_1 (m, t);
}
/* Return 1 when the mutex was signalled and 0 when not. */
static int
fair_cond_timedwait (pthread_cond_t *c, fair_mutex *m, scm_t_timespec *at)
{
int res;
scm_copt_thread *t = m->owner;
fair_mutex_unlock_1 (m);
res = pthread_cond_timedwait (c, &m->lock, at); /* XXX - signals? */
fair_mutex_lock_1 (m, t);
return res == 0;
}
/*** Scheduling */
/* When a thread wants to execute Guile functions, it locks the
guile_mutex.
*/
static fair_mutex guile_mutex;
static SCM cur_thread;
void *scm_i_copt_thread_data;
void
scm_i_copt_set_thread_data (void *data)
{
scm_copt_thread *t = SCM_THREAD_DATA (cur_thread);
scm_i_copt_thread_data = data;
t->root = (scm_root_state *)data;
}
static void
resume (scm_copt_thread *t)
{
cur_thread = t->handle;
scm_i_copt_thread_data = t->root;
t->top = NULL;
}
static void
enter_guile (scm_copt_thread *t)
{
fair_mutex_lock (&guile_mutex, t);
resume (t);
}
static scm_copt_thread *
suspend ()
{
SCM cur = cur_thread;
scm_copt_thread *c = SCM_THREAD_DATA (cur);
/* record top of stack for the GC */
c->top = (SCM_STACKITEM *)&c;
/* save registers. */
SCM_FLUSH_REGISTER_WINDOWS;
setjmp (c->regs);
return c;
}
static scm_copt_thread *
leave_guile ()
{
scm_copt_thread *c = suspend ();
fair_mutex_unlock (&guile_mutex);
return c;
}
int scm_i_switch_counter;
SCM
scm_yield ()
{
/* Testing guile_mutex.next_waiting without locking guile_mutex.lock
is OK since the outcome is not critical. Even when it changes
after the test, we do the right thing.
*/
if (guile_mutex.next_waiting)
{
scm_copt_thread *t = suspend ();
fair_mutex_yield (&guile_mutex);
resume (t);
}
return SCM_BOOL_T;
}
/* Put the current thread to sleep until it is explicitely unblocked.
*/
static void
block ()
{
scm_copt_thread *t = suspend ();
fair_cond_wait (&t->sleep_cond, &guile_mutex);
resume (t);
}
/* Put the current thread to sleep until it is explicitely unblocked
or until a signal arrives or until time AT (absolute time) is
reached. Return 1 when it has been unblocked; 0 otherwise.
*/
static int
timed_block (scm_t_timespec *at)
{
int res;
scm_copt_thread *t = suspend ();
res = fair_cond_timedwait (&t->sleep_cond, &guile_mutex, at);
resume (t);
return res;
}
/* Unblock a sleeping thread.
*/
static void
unblock (scm_copt_thread *t)
{
pthread_cond_signal (&t->sleep_cond);
}
/*** Thread creation */
static SCM all_threads;
static int thread_count;
typedef struct launch_data {
SCM thread;
SCM rootcont;
scm_t_catch_body body;
void *body_data;
scm_t_catch_handler handler;
void *handler_data;
} launch_data;
static SCM
body_bootstrip (launch_data* data)
{
/* First save the new root continuation */
data->rootcont = scm_root->rootcont;
return (data->body) (data->body_data);
// return scm_call_0 (data->body);
}
static SCM
handler_bootstrip (launch_data* data, SCM tag, SCM throw_args)
{
scm_root->rootcont = data->rootcont;
return (data->handler) (data->handler_data, tag, throw_args);
// return scm_apply_1 (data->handler, tag, throw_args);
}
static void
really_launch (SCM_STACKITEM *base, launch_data *data)
{
SCM thread = data->thread;
scm_copt_thread *t = SCM_THREAD_DATA (thread);
init_thread_creatant (thread, base);
enter_guile (t);
data->rootcont = SCM_BOOL_F;
t->result =
scm_internal_cwdr ((scm_t_catch_body) body_bootstrip,
data,
(scm_t_catch_handler) handler_bootstrip,
data, base);
free (data);
pthread_detach (t->pthread);
all_threads = scm_delq (thread, all_threads);
t->pthread = -1;
thread_count--;
leave_guile ();
}
static void *
launch_thread (void *p)
{
really_launch ((SCM_STACKITEM *)&p, (launch_data *)p);
return NULL;
}
static SCM
create_thread (scm_t_catch_body body, void *body_data,
scm_t_catch_handler handler, void *handler_data,
SCM protects)
{
SCM thread;
/* Make new thread. The first thing the new thread will do is to
lock guile_mutex. Thus, we can safely complete its
initialization after creating it. While the new thread starts,
all its data is protected via all_threads.
*/
{
pthread_t th;
SCM root, old_winds;
launch_data *data;
/* Unwind wind chain. */
old_winds = scm_dynwinds;
scm_dowinds (SCM_EOL, scm_ilength (scm_root->dynwinds));
/* Allocate thread locals. */
root = scm_make_root (scm_root->handle);
data = scm_gc_malloc (sizeof (launch_data));
/* Make thread. */
thread = make_thread (protects);
data->thread = thread;
data->body = body;
data->body_data = body_data;
data->handler = handler;
data->handler_data = handler_data;
pthread_create (&th, NULL, launch_thread, (void *) data);
init_thread_creator (thread, th, SCM_ROOT_STATE (root));
all_threads = scm_cons (thread, all_threads);
thread_count++;
/* Return to old dynamic context. */
scm_dowinds (old_winds, - scm_ilength (old_winds));
}
return thread;
}
SCM
scm_call_with_new_thread (SCM argl)
#define FUNC_NAME s_call_with_new_thread
{
SCM thunk, handler;
/* Check arguments. */
{
register SCM args = argl;
if (!scm_is_pair (args))
SCM_WRONG_NUM_ARGS ();
thunk = SCM_CAR (args);
SCM_ASSERT (scm_is_true (scm_thunk_p (thunk)),
thunk,
SCM_ARG1,
s_call_with_new_thread);
args = SCM_CDR (args);
if (!scm_is_pair (args))
SCM_WRONG_NUM_ARGS ();
handler = SCM_CAR (args);
SCM_ASSERT (scm_is_true (scm_procedure_p (handler)),
handler,
SCM_ARG2,
s_call_with_new_thread);
if (!scm_is_null (SCM_CDR (args)))
SCM_WRONG_NUM_ARGS ();
}
return create_thread ((scm_t_catch_body) scm_call_0, thunk,
(scm_t_catch_handler) scm_apply_1, handler,
argl);
}
#undef FUNC_NAME
SCM
scm_spawn_thread (scm_t_catch_body body, void *body_data,
scm_t_catch_handler handler, void *handler_data)
{
return create_thread (body, body_data, handler, handler_data, SCM_BOOL_F);
}
/*** Mutexes */
/* We implement our own mutex type since we want them to be 'fair', we
want to do fancy things while waiting for them (like running
asyncs) and we want to support waiting on many things at once.
Also, we might add things that are nice for debugging.
*/
typedef struct scm_copt_mutex {
/* the thread currently owning the mutex, or SCM_BOOL_F. */
SCM owner;
/* how much the owner owns us. */
int level;
/* the threads waiting for this mutex. */
SCM waiting;
} scm_copt_mutex;
SCM
scm_make_mutex ()
{
SCM mx = scm_make_smob (scm_tc16_mutex);
scm_copt_mutex *m = SCM_MUTEX_DATA (mx);
m->owner = SCM_BOOL_F;
m->level = 0;
m->waiting = make_queue ();
return mx;
}
SCM
scm_lock_mutex (SCM mx)
#define FUNC_NAME s_lock_mutex
{
scm_copt_mutex *m;
SCM_ASSERT (SCM_MUTEXP (mx), mx, SCM_ARG1, FUNC_NAME);
m = SCM_MUTEX_DATA (mx);
if (m->owner == SCM_BOOL_F)
m->owner = cur_thread;
else if (m->owner == cur_thread)
m->level++;
else
{
while (m->owner != cur_thread)
{
enqueue (m->waiting, cur_thread);
block ();
SCM_ASYNC_TICK;
}
}
return SCM_BOOL_T;
}
#undef FUNC_NAME
SCM
scm_try_mutex (SCM mx)
#define FUNC_NAME s_try_mutex
{
scm_copt_mutex *m;
SCM_ASSERT (SCM_MUTEXP (mx), mx, SCM_ARG1, FUNC_NAME);
m = SCM_MUTEX_DATA (mx);
if (m->owner == SCM_BOOL_F)
m->owner = cur_thread;
else if (m->owner == cur_thread)
m->level++;
else
return SCM_BOOL_F;
return SCM_BOOL_T;
}
#undef FUNC_NAME
SCM
scm_unlock_mutex (SCM mx)
#define FUNC_NAME s_unlock_mutex
{
scm_copt_mutex *m;
SCM_ASSERT (SCM_MUTEXP (mx), mx, SCM_ARG1, FUNC_NAME);
m = SCM_MUTEX_DATA (mx);
if (m->owner != cur_thread)
{
if (m->owner == SCM_BOOL_F)
SCM_MISC_ERROR ("mutex not locked", SCM_EOL);
else
SCM_MISC_ERROR ("mutex not locked by this thread", SCM_EOL);
}
else if (m->level > 0)
m->level--;
else
{
SCM next = dequeue (m->waiting);
if (scm_is_true (next))
{
m->owner = next;
unblock (SCM_THREAD_DATA (next));
scm_yield ();
}
else
m->owner = SCM_BOOL_F;
}
return SCM_BOOL_T;
}
#undef FUNC_NAME
/*** Condition variables */
/* Like mutexes, we implement our own condition variables using the
primitives above.
*/
/* yeah, we don't need a structure for this, but more things (like a
name) will likely follow... */
typedef struct scm_copt_cond {
/* the threads waiting for this condition. */
SCM waiting;
} scm_copt_cond;
static SCM
cond_mark (SCM cv)
{
scm_copt_cond *c = SCM_CONDVAR_DATA (cv);
return c->waiting;
}
SCM
scm_make_condition_variable (void)
{
SCM cv = scm_make_smob (scm_tc16_condvar);
scm_copt_cond *c = SCM_CONDVAR_DATA (cv);
c->waiting = make_queue ();
return cv;
}
SCM
scm_timed_wait_condition_variable (SCM cv, SCM mx, SCM t)
#define FUNC_NAME s_wait_condition_variable
{
scm_copt_cond *c;
scm_t_timespec waittime;
int res;
SCM_ASSERT (SCM_CONDVARP (cv),
cv,
SCM_ARG1,
s_wait_condition_variable);
SCM_ASSERT (SCM_MUTEXP (mx),
mx,
SCM_ARG2,
s_wait_condition_variable);
if (!SCM_UNBNDP (t))
{
if (scm_is_pair (t))
{
SCM_VALIDATE_UINT_COPY (3, SCM_CAR(t), waittime.tv_sec);
SCM_VALIDATE_UINT_COPY (3, SCM_CDR(t), waittime.tv_nsec);
waittime.tv_nsec *= 1000;
}
else
{
SCM_VALIDATE_UINT_COPY (3, t, waittime.tv_sec);
waittime.tv_nsec = 0;
}
}
c = SCM_CONDVAR_DATA (cv);
enqueue (c->waiting, cur_thread);
scm_unlock_mutex (mx);
if (SCM_UNBNDP (t))
{
block ();
res = 1;
}
else
res = timed_block (&waittime);
scm_lock_mutex (mx);
return scm_from_bool (res);
}
#undef FUNC_NAME
SCM
scm_signal_condition_variable (SCM cv)
#define FUNC_NAME s_signal_condition_variable
{
SCM th;
scm_copt_cond *c;
SCM_ASSERT (SCM_CONDVARP (cv),
cv,
SCM_ARG1,
s_signal_condition_variable);
c = SCM_CONDVAR_DATA (cv);
if (scm_is_true (th = dequeue (c->waiting)))
unblock (SCM_THREAD_DATA (th));
return SCM_BOOL_T;
}
#undef FUNC_NAME
SCM
scm_broadcast_condition_variable (SCM cv)
#define FUNC_NAME s_broadcast_condition_variable
{
SCM th;
scm_copt_cond *c;
SCM_ASSERT (SCM_CONDVARP (cv),
cv,
SCM_ARG1,
s_signal_condition_variable);
c = SCM_CONDVAR_DATA (cv);
while (scm_is_true (th = dequeue (c->waiting)))
unblock (SCM_THREAD_DATA (th));
return SCM_BOOL_T;
}
#undef FUNC_NAME
/*** Initialization */
void
scm_threads_init (SCM_STACKITEM *base)
{
scm_tc16_thread = scm_make_smob_type ("thread", 0);
scm_tc16_mutex = scm_make_smob_type ("mutex", sizeof (scm_copt_mutex));
scm_tc16_condvar = scm_make_smob_type ("condition-variable",
sizeof (scm_copt_cond));
scm_i_switch_counter = SCM_I_THREAD_SWITCH_COUNT;
fair_mutex_init (&guile_mutex);
cur_thread = make_thread (SCM_BOOL_F);
enter_guile (SCM_THREAD_DATA (cur_thread));
/* root is set later from init.c */
init_thread_creator (cur_thread, pthread_self(), NULL);
init_thread_creatant (cur_thread, base);
thread_count = 1;
scm_gc_register_root (&all_threads);
all_threads = scm_cons (cur_thread, SCM_EOL);
scm_set_smob_print (scm_tc16_thread, thread_print);
}
/*** Marking stacks */
/* XXX - what to do with this? Do we need to handle this for blocked
threads as well?
*/
#ifdef __ia64__
# define SCM_MARK_BACKING_STORE() do { \
ucontext_t ctx; \
SCM_STACKITEM * top, * bot; \
getcontext (&ctx); \
scm_mark_locations ((SCM_STACKITEM *) &ctx.uc_mcontext, \
((size_t) (sizeof (SCM_STACKITEM) - 1 + sizeof ctx.uc_mcontext) \
/ sizeof (SCM_STACKITEM))); \
bot = (SCM_STACKITEM *) scm_ia64_register_backing_store_base (); \
top = (SCM_STACKITEM *) scm_ia64_ar_bsp (&ctx); \
scm_mark_locations (bot, top - bot); } while (0)
#else
# define SCM_MARK_BACKING_STORE()
#endif
/*** Select */
int
scm_internal_select (int nfds,
SELECT_TYPE *readfds,
SELECT_TYPE *writefds,
SELECT_TYPE *exceptfds,
struct timeval *timeout)
{
int res, eno;
scm_copt_thread *c = leave_guile ();
res = select (nfds, readfds, writefds, exceptfds, timeout);
eno = errno;
enter_guile (c);
SCM_ASYNC_TICK;
errno = eno;
return res;
}
void
scm_init_iselect ()
{
}
unsigned long
scm_thread_usleep (unsigned long usec)
{
scm_copt_thread *c = leave_guile ();
usleep (usec);
enter_guile (c);
return 0;
}
unsigned long
scm_thread_sleep (unsigned long sec)
{
unsigned long res;
scm_copt_thread *c = leave_guile ();
res = sleep (sec);
enter_guile (c);
return res;
}
/*** Misc */
SCM
scm_current_thread (void)
{
return cur_thread;
}
SCM
scm_all_threads (void)
{
return all_threads;
}
scm_root_state *
scm_i_thread_root (SCM thread)
{
if (thread == cur_thread)
return scm_i_copt_thread_data;
else
return ((scm_copt_thread *)SCM_THREAD_DATA (thread))->root;
}
SCM
scm_join_thread (SCM thread)
#define FUNC_NAME s_join_thread
{
scm_copt_thread *t;
SCM res;
SCM_VALIDATE_THREAD (1, thread);
t = SCM_THREAD_DATA (thread);
if (t->pthread != -1)
{
scm_copt_thread *c = leave_guile ();
pthread_join (t->pthread, NULL);
enter_guile (c);
}
res = t->result;
t->result = SCM_BOOL_F;
return res;
}
#undef FUNC_NAME
int
scm_c_thread_exited_p (SCM thread)
#define FUNC_NAME s_scm_thread_exited_p
{
scm_copt_thread *t;
SCM_VALIDATE_THREAD (1, thread);
t = SCM_THREAD_DATA (thread);
return t->pthread == -1;
}
#undef FUNC_NAME
/*
Local Variables:
c-file-style: "gnu"
End:
*/

83
libguile/coop-pthreads.h
View file

@ -1,83 +0,0 @@
/* classes: h_files */
#ifndef SCM_COOP_PTHREADS_H
#define SCM_COOP_PTHREADS_H
/* Copyright (C) 2002, 2006, 2008 Free Software Foundation, Inc.
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
/* The coop-pthreads implementation. We use pthreads for the basic
multi threading stuff, but rig it so that only one thread is ever
active inside Guile.
*/
#include <pthread.h>
#include "libguile/boehm-gc.h"
#include "libguile/iselect.h"
#if (SCM_ENABLE_DEPRECATED == 1)
/* Thread local data support --- generic C API */
typedef pthread_key_t scm_t_key;
#define scm_key_create pthread_key_create
#define scm_setspecific pthread_setspecific
#define scm_getspecific pthread_getspecific
#define scm_key_delete pthread_key_delete
#endif /* SCM_ENABLE_DEPRECATED == 1 */
/* Since only one thread can be active anyway, we don't need to do
anything special around critical sections. In fact, that's the
reason we do only support cooperative threading: Guile's critical
regions have not been completely identified yet. (I think.) */
#define SCM_CRITICAL_SECTION_START
#define SCM_CRITICAL_SECTION_END
#define SCM_I_THREAD_SWITCH_COUNT 50
#define SCM_THREAD_SWITCHING_CODE \
do { \
scm_i_switch_counter--; \
if (scm_i_switch_counter == 0) \
{ \
scm_i_switch_counter = SCM_I_THREAD_SWITCH_COUNT; \
scm_yield(); \
} \
} while (0)
SCM_API int scm_i_switch_counter;
#define SCM_THREAD_LOCAL_DATA (scm_i_copt_thread_data)
#define SCM_SET_THREAD_LOCAL_DATA(ptr) (scm_i_copt_set_thread_data (ptr))
SCM_API void *scm_i_copt_thread_data;
SCM_INTERNAL void scm_i_copt_set_thread_data (void *data);
#endif /* SCM_COOP_PTHREAD_H */
/*
Local Variables:
c-file-style: "gnu"
End:
*/

763
libguile/coop.c
View file

@ -1,763 +0,0 @@
/* Copyright (C) 1995, 1996, 1997, 1998, 1999, 2000, 2001, 2002, 2006 Free Software Foundation, Inc.
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
/* $Id: coop.c,v 1.39 2006-04-17 00:05:38 kryde Exp $ */
/* Cooperative thread library, based on QuickThreads */
#ifdef HAVE_CONFIG_H
# include <config.h>
#endif
#include <stdio.h>
#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif
#include <errno.h>
#include "qt/qt.h"
#include "libguile/eval.h"
/* #define COOP_STKSIZE (0x10000) */
#define COOP_STKSIZE (scm_eval_stack)
/* `alignment' must be a power of 2. */
#define COOP_STKALIGN(sp, alignment) \
((void *)((((qt_word_t)(sp)) + (alignment) - 1) & ~((alignment)-1)))
/* Queue access functions. */
static void
coop_qinit (coop_q_t *q)
{
q->t.next = q->tail = &q->t;
q->t.all_prev = NULL;
q->t.all_next = NULL;
q->t.nfds = 0;
q->t.readfds = NULL;
q->t.writefds = NULL;
q->t.exceptfds = NULL;
q->t.timeoutp = 0;
}
coop_t *
coop_qget (coop_q_t *q)
{
coop_t *t;
t = q->t.next;
q->t.next = t->next;
if (t->next == &q->t)
{
if (t == &q->t)
{ /* If it was already empty .. */
return NULL; /* .. say so. */
}
q->tail = &q->t; /* Else now it is empty. */
}
return (t);
}
void
coop_qput (coop_q_t *q, coop_t *t)
{
q->tail->next = t;
t->next = &q->t;
q->tail = t;
}
static void
coop_all_qput (coop_q_t *q, coop_t *t)
{
if (q->t.all_next)
q->t.all_next->all_prev = t;
t->all_prev = NULL;
t->all_next = q->t.all_next;
q->t.all_next = t;
}
static void
coop_all_qremove (coop_q_t *q, coop_t *t)
{
if (t->all_prev)
t->all_prev->all_next = t->all_next;
else
q->t.all_next = t->all_next;
if (t->all_next)
t->all_next->all_prev = t->all_prev;
}
/* Insert thread t into the ordered queue q.
q is ordered after wakeup_time. Threads which aren't sleeping but
waiting for I/O go last into the queue. */
void
coop_timeout_qinsert (coop_q_t *q, coop_t *t)
{
coop_t *pred = &q->t;
int sec = t->wakeup_time.tv_sec;
int usec = t->wakeup_time.tv_usec;
while (pred->next != &q->t
&& pred->next->timeoutp
&& (pred->next->wakeup_time.tv_sec < sec
|| (pred->next->wakeup_time.tv_sec == sec
&& pred->next->wakeup_time.tv_usec < usec)))
pred = pred->next;
t->next = pred->next;
pred->next = t;
if (t->next == &q->t)
q->tail = t;
}
/* Thread routines. */
coop_q_t coop_global_runq; /* A queue of runable threads. */
coop_q_t coop_global_sleepq; /* A queue of sleeping threads. */
coop_q_t coop_tmp_queue; /* A temp working queue */
coop_q_t coop_global_allq; /* A queue of all threads. */
static coop_t coop_global_main; /* Thread for the process. */
coop_t *coop_global_curr; /* Currently-executing thread. */
#ifdef GUILE_PTHREAD_COMPAT
static coop_q_t coop_deadq;
static int coop_quitting_p = -1;
static pthread_cond_t coop_cond_quit;
static pthread_cond_t coop_cond_create;
static pthread_mutex_t coop_mutex_create;
static pthread_t coop_mother;
static int mother_awake_p = 0;
static coop_t *coop_child;
#endif
static void *coop_starthelp (qt_t *old, void *ignore0, void *ignore1);
static void coop_only (void *pu, void *pt, qt_userf_t *f);
static void *coop_aborthelp (qt_t *sp, void *old, void *null);
static void *coop_yieldhelp (qt_t *sp, void *old, void *blockq);
/* called on process termination. */
#ifdef HAVE_ATEXIT
static void
coop_finish (void)
#else
#ifdef HAVE_ON_EXIT
extern int on_exit (void (*procp) (), int arg);
static void
coop_finish (int status, void *arg)
#else
#error Dont know how to setup a cleanup handler on your system.
#endif
#endif
{
#ifdef GUILE_PTHREAD_COMPAT
coop_quitting_p = 1;
pthread_cond_signal (&coop_cond_create);
pthread_cond_broadcast (&coop_cond_quit);
#endif
}
void
coop_init ()
{
coop_qinit (&coop_global_runq);
coop_qinit (&coop_global_sleepq);
coop_qinit (&coop_tmp_queue);
coop_qinit (&coop_global_allq);
coop_global_curr = &coop_global_main;
#ifdef GUILE_PTHREAD_COMPAT
coop_qinit (&coop_deadq);
pthread_cond_init (&coop_cond_quit, NULL);
pthread_cond_init (&coop_cond_create, NULL);
pthread_mutex_init (&coop_mutex_create, NULL);
#endif
#ifdef HAVE_ATEXIT
atexit (coop_finish);
#else
#ifdef HAVE_ON_EXIT
on_exit (coop_finish, 0);
#endif
#endif
}
void
coop_start()
{
coop_t *next;
while ((next = coop_qget (&coop_global_runq)) != NULL) {
coop_global_curr = next;
QT_BLOCK (coop_starthelp, 0, 0, next->sp);
}
}
static void *
coop_starthelp (qt_t *old, void *ignore0, void *ignore1)
{
coop_global_main.sp = old;
coop_global_main.joining = NULL;
coop_qput (&coop_global_runq, &coop_global_main);
return NULL; /* not used, but keeps compiler happy */
}
int
coop_mutex_init (coop_m *m)
{
return coop_new_mutex_init (m, NULL);
}
int
coop_new_mutex_init (coop_m *m, coop_mattr *attr)
{
m->owner = NULL;
m->level = 0;
coop_qinit(&(m->waiting));
return 0;
}
int
coop_mutex_trylock (coop_m *m)
{
if (m->owner == NULL)
{
m->owner = coop_global_curr;
return 0;
}
else if (m->owner == coop_global_curr)
{
m->level++;
return 0;
}
else
return EBUSY;
}
int
coop_mutex_lock (coop_m *m)
{
if (m->owner == NULL)
{
m->owner = coop_global_curr;
}
else if (m->owner == coop_global_curr)
{
m->level++;
}
else
{
coop_t *old, *newthread;
/* Record the current top-of-stack before going to sleep */
coop_global_curr->top = &old;
newthread = coop_wait_for_runnable_thread();
if (newthread == coop_global_curr)
coop_abort ();
old = coop_global_curr;
coop_global_curr = newthread;
QT_BLOCK (coop_yieldhelp, old, &(m->waiting), newthread->sp);
}
return 0;
}
int
coop_mutex_unlock (coop_m *m)
{
coop_t *old, *newthread;
if (m->level == 0)
{
newthread = coop_qget (&(m->waiting));
if (newthread != NULL)
{
/* Record the current top-of-stack before going to sleep */
coop_global_curr->top = &old;
old = coop_global_curr;
coop_global_curr = newthread;
/* The new thread came into m->waiting through a lock operation.
It now owns this mutex. */
m->owner = coop_global_curr;
QT_BLOCK (coop_yieldhelp, old, &coop_global_runq, newthread->sp);
}
else
{
m->owner = NULL;
}
}
else if (m->level > 0)
m->level--;
else
abort (); /* XXX */
return 0;
}
int
coop_mutex_destroy (coop_m *m)
{
return 0;
}
int
coop_condition_variable_init (coop_c *c)
{
return coop_new_condition_variable_init (c, NULL);
}
int
coop_new_condition_variable_init (coop_c *c, coop_cattr *a)
{
coop_qinit(&(c->waiting));
return 0;
}
int
coop_condition_variable_wait_mutex (coop_c *c, coop_m *m)
{
coop_t *old, *newthread;
/* coop_mutex_unlock (m); */
newthread = coop_qget (&(m->waiting));
if (newthread != NULL)
{
m->owner = newthread;
}
else
{
m->owner = NULL;
/*fixme* Should we really wait here? Isn't it OK just to proceed? */
newthread = coop_wait_for_runnable_thread();
if (newthread == coop_global_curr)
coop_abort ();
}
coop_global_curr->top = &old;
old = coop_global_curr;
coop_global_curr = newthread;
QT_BLOCK (coop_yieldhelp, old, &(c->waiting), newthread->sp);
coop_mutex_lock (m);
return 0;
}
int
coop_condition_variable_timed_wait_mutex (coop_c *c,
coop_m *m,
const scm_t_timespec *abstime)
{
coop_t *old, *t;
#ifdef ETIMEDOUT
int res = ETIMEDOUT;
#elif defined (WSAETIMEDOUT)
int res = WSAETIMEDOUT;
#else
int res = 0;
#endif
/* coop_mutex_unlock (m); */
t = coop_qget (&(m->waiting));
if (t != NULL)
{
m->owner = t;
}
else
{
m->owner = NULL;
coop_global_curr->timeoutp = 1;
coop_global_curr->wakeup_time.tv_sec = abstime->tv_sec;
coop_global_curr->wakeup_time.tv_usec = abstime->tv_nsec / 1000;
coop_timeout_qinsert (&coop_global_sleepq, coop_global_curr);
t = coop_wait_for_runnable_thread();
}
if (t != coop_global_curr)
{
coop_global_curr->top = &old;
old = coop_global_curr;
coop_global_curr = t;
QT_BLOCK (coop_yieldhelp, old, &(c->waiting), t->sp);
/* Are we still in the sleep queue? */
old = &coop_global_sleepq.t;
for (t = old->next; t != &coop_global_sleepq.t; old = t, t = t->next)
if (t == coop_global_curr)
{
old->next = t->next; /* unlink */
res = 0;
break;
}
}
coop_mutex_lock (m);
return res;
}
int
coop_condition_variable_broadcast (coop_c *c)
{
coop_t *newthread;
while ((newthread = coop_qget (&(c->waiting))) != NULL)
{
coop_qput (&coop_global_runq, newthread);
}
return 0;
}
int
coop_condition_variable_signal (coop_c *c)
{
return coop_condition_variable_broadcast (c);
}
/* {Keys}
*/
static int n_keys = 0;
static int max_keys = 0;
static void (**destructors) (void *) = 0;
int
coop_key_create (coop_k *keyp, void (*destructor) (void *value))
{
if (n_keys >= max_keys)
{
int i;
max_keys = max_keys ? max_keys * 3 / 2 : 10;
destructors = realloc (destructors, sizeof (void *) * max_keys);
if (destructors == 0)
{
fprintf (stderr, "Virtual memory exceeded in coop_key_create\n");
exit (1);
}
for (i = n_keys; i < max_keys; ++i)
destructors[i] = NULL;
}
destructors[n_keys] = destructor;
*keyp = n_keys++;
return 0;
}
int
coop_setspecific (coop_k key, const void *value)
{
int n_keys = coop_global_curr->n_keys;
if (key >= n_keys)
{
int i;
coop_global_curr->n_keys = max_keys;
coop_global_curr->specific = realloc (n_keys
? coop_global_curr->specific
: NULL,
sizeof (void *) * max_keys);
if (coop_global_curr->specific == 0)
{
fprintf (stderr, "Virtual memory exceeded in coop_setspecific\n");
exit (1);
}
for (i = n_keys; i < max_keys; ++i)
coop_global_curr->specific[i] = NULL;
}
coop_global_curr->specific[key] = (void *) value;
return 0;
}
void *
coop_getspecific (coop_k key)
{
return (key < coop_global_curr->n_keys
? coop_global_curr->specific[key]
: NULL);
}
int
coop_key_delete (coop_k key)
{
return 0;
}
int
coop_condition_variable_destroy (coop_c *c)
{
return 0;
}
#ifdef GUILE_PTHREAD_COMPAT
#include "libguile/boehm-gc.h"
/* 1K room for the cond wait routine */
#if SCM_STACK_GROWS_UP
# define COOP_STACK_ROOM (256)
#else
# define COOP_STACK_ROOM (-256)
#endif
static void *
dummy_start (void *coop_thread)
{
coop_t *t = (coop_t *) coop_thread;
int res;
t->sp = (qt_t *) (&t + COOP_STACK_ROOM);
pthread_mutex_init (&t->dummy_mutex, NULL);
pthread_mutex_lock (&t->dummy_mutex);
coop_child = 0;
do
res = pthread_cond_wait (&coop_cond_quit, &t->dummy_mutex);
while (res == EINTR);
return 0;
}
static void *
mother (void *dummy)
{
pthread_mutex_lock (&coop_mutex_create);
while (!coop_quitting_p)
{
int res;
pthread_create (&coop_child->dummy_thread,
NULL,
dummy_start,
coop_child);
mother_awake_p = 0;
do
res = pthread_cond_wait (&coop_cond_create, &coop_mutex_create);
while (res == EINTR);
}
return 0;
}
#endif
coop_t *
coop_create (coop_userf_t *f, void *pu)
{
coop_t *t;
#ifndef GUILE_PTHREAD_COMPAT
void *sto;
#endif
#ifdef GUILE_PTHREAD_COMPAT
t = coop_qget (&coop_deadq);
if (t)
{
t->sp = t->base;
t->specific = 0;
t->n_keys = 0;
}
else
#endif
{
t = scm_malloc (sizeof (coop_t));
t->specific = NULL;
t->n_keys = 0;
#ifdef GUILE_PTHREAD_COMPAT
coop_child = t;
mother_awake_p = 1;
if (coop_quitting_p < 0)
{
coop_quitting_p = 0;
/* We can't create threads ourselves since the pthread
* corresponding to this stack might be sleeping.
*/
pthread_create (&coop_mother, NULL, mother, NULL);
}
else
{
pthread_cond_signal (&coop_cond_create);
}
/* We can't use a pthreads condition variable since "this"
* pthread could already be asleep. We can't use a COOP
* condition variable because they are not safe against
* pre-emptive switching.
*/
while (coop_child || mother_awake_p)
usleep (0);
#else
t->sto = scm_malloc (COOP_STKSIZE);
sto = COOP_STKALIGN (t->sto, QT_STKALIGN);
t->sp = QT_SP (sto, COOP_STKSIZE - QT_STKALIGN);
#endif
t->base = t->sp;
}
t->sp = QT_ARGS (t->sp, pu, t, (qt_userf_t *)f, coop_only);
t->joining = NULL;
coop_qput (&coop_global_runq, t);
coop_all_qput (&coop_global_allq, t);
return t;
}
static void
coop_only (void *pu, void *pt, qt_userf_t *f)
{
coop_global_curr = (coop_t *)pt;
(*(coop_userf_t *)f)(pu);
coop_abort();
/* NOTREACHED */
}
void
coop_abort ()
{
coop_t *old, *newthread;
/* Wake up any threads that are waiting to join this one */
if (coop_global_curr->joining)
{
while ((newthread = coop_qget ((coop_q_t *)(coop_global_curr->joining)))
!= NULL)
{
coop_qput (&coop_global_runq, newthread);
}
free (coop_global_curr->joining);
}
scm_I_am_dead = 1;
do {
newthread = coop_wait_for_runnable_thread();
} while (newthread == coop_global_curr);
scm_I_am_dead = 0;
coop_all_qremove (&coop_global_allq, coop_global_curr);
old = coop_global_curr;
coop_global_curr = newthread;
QT_ABORT (coop_aborthelp, old, (void *) NULL, newthread->sp);
}
static void *
coop_aborthelp (qt_t *sp, void *old, void *null)
{
coop_t *oldthread = (coop_t *) old;
if (oldthread->specific)
free (oldthread->specific);
#ifndef GUILE_PTHREAD_COMPAT
free (oldthread->sto);
free (oldthread);
#else
coop_qput (&coop_deadq, oldthread);
#endif
return NULL;
}
void
coop_join(coop_t *t)
{
coop_t *old, *newthread;
/* Create a join list if necessary */
if (t->joining == NULL)
{
t->joining = scm_malloc(sizeof(coop_q_t));
coop_qinit((coop_q_t *) t->joining);
}
newthread = coop_wait_for_runnable_thread();
if (newthread == coop_global_curr)
return;
old = coop_global_curr;
coop_global_curr = newthread;
QT_BLOCK (coop_yieldhelp, old, (coop_q_t *) t->joining, newthread->sp);
}
void
coop_yield()
{
coop_t *old = NULL;
coop_t *newthread;
newthread = coop_next_runnable_thread();
/* There may be no other runnable threads. Return if this is the
case. */
if (newthread == coop_global_curr)
return;
old = coop_global_curr;
coop_global_curr = newthread;
QT_BLOCK (coop_yieldhelp, old, &coop_global_runq, newthread->sp);
}
static void *
coop_yieldhelp (qt_t *sp, void *old, void *blockq)
{
((coop_t *)old)->sp = sp;
coop_qput ((coop_q_t *)blockq, (coop_t *)old);
return NULL;
}
/* Replacement for the system's sleep() function. Does the right thing
for the process - but not for the system (it busy-waits) */
void *
coop_sleephelp (qt_t *sp, void *old, void *blockq)
{
((coop_t *)old)->sp = sp;
/* old is already on the sleep queue - so there's no need to
do anything extra here */
return NULL;
}
unsigned long
scm_thread_usleep (unsigned long usec)
{
struct timeval timeout;
timeout.tv_sec = 0;
timeout.tv_usec = usec;
scm_internal_select (0, NULL, NULL, NULL, &timeout);
return 0; /* Maybe we should calculate actual time slept,
but this is faster... :) */
}
unsigned long
scm_thread_sleep (unsigned long sec)
{
time_t now = time (NULL);
struct timeval timeout;
unsigned long slept;
timeout.tv_sec = sec;
timeout.tv_usec = 0;
scm_internal_select (0, NULL, NULL, NULL, &timeout);
slept = time (NULL) - now;
return slept > sec ? 0 : sec - slept;
}
/*
Local Variables:
c-file-style: "gnu"
End:
*/