src/thread.c

   1
   2 #include <config.h>
   3 #include "lisp.h"
   4 #include "buffer.h"
   5 #include "blockinput.h"
   6 #include <pthread.h>
   7 #include "systime.h"
   8 #include "sysselect.h"
   9
  10 void mark_byte_stack P_ ((struct byte_stack *));
  11 void mark_backtrace P_ ((struct backtrace *));
  12 void mark_catchlist P_ ((struct catchtag *));
  13 void mark_stack P_ ((char *, char *));
  14 void flush_stack_call_func P_ ((void (*) (char *, void *), void *));
  15
  16 /* Get the next thread as in circular buffer.  */
  17 #define NEXT_THREAD(x)(x->next_thread ? x->next_thread : all_threads)
  18
  19 /* condition var .. w/ global lock */
  20
  21 static pthread_cond_t thread_cond;
  22
  23 static struct thread_state primary_thread;
  24
  25 static struct thread_state *all_threads = &primary_thread;
  26
  27 __thread struct thread_state *current_thread = &primary_thread;
  28
  29 pthread_mutex_t global_lock;
  30
  31 /* Used internally by the scheduler, it is the next that will be executed.  */
  32 static pthread_t next_thread;
  33
  34 Lisp_Object minibuffer_mutex;
  35
  36 /* Choose the next thread to be executed.  */
  37 static void
  38 thread_schedule ()
  39 {
  40   struct thread_state *it = current_thread;
  41   do
  42     {
  43       it = NEXT_THREAD (it);
  44     }
  45   while (it->blocked && it != current_thread);
  46
  47   next_thread = it->pthread_id;
  48 }
  49
  50 /* Schedule a new thread and block the caller until it is not scheduled
  51    again.  */
  52 static inline void
  53 reschedule (char *end, int wait)
  54 {
  55   current_thread->stack_top = end;
  56   thread_schedule ();
  57
  58   if (next_thread != current_thread->pthread_id)
  59     pthread_cond_broadcast (&thread_cond);
  60
  61   if (!wait)
  62     return;
  63
  64   pthread_mutex_unlock (&global_lock);
  65
  66   pthread_mutex_lock (&global_lock);
  67
  68   while (current_thread->pthread_id != next_thread)
  69     pthread_cond_wait (&thread_cond, &global_lock);
  70 }
  71
  72 static void
  73 mark_one_thread (struct thread_state *thread)
  74 {
  75   register struct specbinding *bind;
  76   struct handler *handler;
  77   Lisp_Object tem;
  78
  79   for (bind = thread->m_specpdl; bind != thread->m_specpdl_ptr; bind++)
  80     {
  81       mark_object (bind->symbol);
  82       mark_object (bind->old_value);
  83     }
  84
  85 #if (GC_MARK_STACK == GC_MAKE_GCPROS_NOOPS \
  86      || GC_MARK_STACK == GC_MARK_STACK_CHECK_GCPROS)
  87   mark_stack (thread->stack_bottom, thread->stack_top);
  88 #else
  89   {
  90     register struct gcpro *tail;
  91     for (tail = thread->m_gcprolist; tail; tail = tail->next)
  92       for (i = 0; i < tail->nvars; i++)
  93         mark_object (tail->var[i]);
  94   }
  95 #endif
  96
  97   if (thread->m_byte_stack_list)
  98     mark_byte_stack (thread->m_byte_stack_list);
  99
 100   mark_catchlist (thread->m_catchlist);
 101
 102   for (handler = thread->m_handlerlist; handler; handler = handler->next)
 103     {
 104       mark_object (handler->handler);
 105       mark_object (handler->var);
 106     }
 107
 108   mark_backtrace (thread->m_backtrace_list);
 109
 110   if (thread->m_current_buffer)
 111     {
 112       XSETBUFFER (tem, thread->m_current_buffer);
 113       mark_object (tem);
 114     }
 115
 116   mark_object (thread->m_last_thing_searched);
 117
 118   if (thread->m_saved_last_thing_searched)
 119     mark_object (thread->m_saved_last_thing_searched);
 120 }
 121
 122 static void
 123 mark_threads_callback (char *end, void *ignore)
 124 {
 125   struct thread_state *iter;
 126
 127   current_thread->stack_top = end;
 128   for (iter = all_threads; iter; iter = iter->next_thread)
 129     {
 130       Lisp_Object thread_obj;
 131       XSETTHREAD (thread_obj, iter);
 132       mark_object (thread_obj);
 133       mark_one_thread (iter);
 134     }
 135 }
 136
 137 void
 138 mark_threads (void)
 139 {
 140   flush_stack_call_func (mark_threads_callback, NULL);
 141 }
 142
 143 void
 144 unmark_threads (void)
 145 {
 146   struct thread_state *iter;
 147
 148   for (iter = all_threads; iter; iter = iter->next_thread)
 149     if (iter->m_byte_stack_list)
 150       unmark_byte_stack (iter->m_byte_stack_list);
 151 }
 152
 153 static void
 154 thread_yield_callback (char *end, void *ignore)
 155 {
 156   reschedule (end, 1);
 157 }
 158
 159 void
 160 thread_yield (void)
 161 {
 162   /* Note: currently it is safe to check this here, but eventually it
 163      will require a lock to ensure non-racy operation.  */
 164   /* Only yield if there is another thread to yield to.  */
 165   if (all_threads->next_thread)
 166     flush_stack_call_func (thread_yield_callback, NULL);
 167 }
 168
 169 DEFUN ("yield", Fyield, Syield, 0, 0, 0,
 170        doc: /* Yield to the next thread.  */)
 171      (void)
 172 {
 173   thread_yield ();
 174   return other_threads_p () ? Qt : Qnil;
 175 }
 176
 177 static Lisp_Object
 178 invoke_thread_function (void)
 179 {
 180   Lisp_Object iter;
 181
 182   int count = SPECPDL_INDEX ();
 183
 184   /* Set up specpdl.  */
 185   for (iter = current_thread->initial_specpdl;
 186        !EQ (iter, Qnil);
 187        iter = XCDR (iter))
 188     {
 189       /* We may bind a variable twice -- but it doesn't matter because
 190          there is no way to undo these bindings without exiting the
 191          thread.  */
 192       specbind (XCAR (XCAR (iter)), XCDR (XCAR (iter)));
 193     }
 194   current_thread->initial_specpdl = Qnil;
 195
 196   Ffuncall (1, &current_thread->func);
 197   return unbind_to (count, Qnil);
 198 }
 199
 200 static Lisp_Object
 201 do_nothing (Lisp_Object whatever)
 202 {
 203   return whatever;
 204 }
 205
 206 static void *
 207 run_thread (void *state)
 208 {
 209   struct thread_state *self = state;
 210   struct thread_state **iter;
 211   struct gcpro gcpro1;
 212   Lisp_Object buffer;
 213   char stack_pos;
 214
 215   self->stack_top = self->stack_bottom = &stack_pos;
 216
 217   self->m_specpdl_size = 50;
 218   self->m_specpdl = xmalloc (self->m_specpdl_size
 219                              * sizeof (struct specbinding));
 220   self->m_specpdl_ptr = self->m_specpdl;
 221   self->pthread_id = pthread_self ();
 222
 223   /* Thread-local assignment.  */
 224   current_thread = self;
 225
 226   /* We need special handling to set the initial buffer.  Our parent
 227      thread is very likely to be using this same buffer so we will
 228      typically wait for the parent thread to release it first.  */
 229   XSETBUFFER (buffer, self->m_current_buffer);
 230   GCPRO1 (buffer);
 231   self->m_current_buffer = 0;
 232
 233   pthread_mutex_lock (&global_lock);
 234
 235   set_buffer_internal (XBUFFER (buffer));
 236
 237   /* It might be nice to do something with errors here.  */
 238   internal_condition_case (invoke_thread_function, Qt, do_nothing);
 239
 240   blocal_unbind_thread (get_current_thread ());
 241
 242   /* Unlink this thread from the list of all threads.  */
 243   for (iter = &all_threads; *iter != self; iter = &(*iter)->next_thread)
 244     ;
 245   *iter = (*iter)->next_thread;
 246
 247   thread_schedule ();
 248   pthread_cond_broadcast (&thread_cond);
 249
 250   xfree (self->m_specpdl);
 251
 252   pthread_mutex_unlock (&global_lock);
 253
 254   return NULL;
 255 }
 256
 257 DEFUN ("run-in-thread", Frun_in_thread, Srun_in_thread, 1, 1, 0,
 258        doc: /* Start a new thread and run FUNCTION in it.
 259 When the function exits, the thread dies.  */)
 260      (function)
 261      Lisp_Object function;
 262 {
 263   char stack_pos;
 264   pthread_t thr;
 265   struct thread_state *new_thread;
 266   struct specbinding *p;
 267
 268   /* Can't start a thread in temacs.  */
 269   if (!initialized)
 270     abort ();
 271
 272   new_thread = ALLOCATE_PSEUDOVECTOR (struct thread_state, m_gcprolist,
 273                                       PVEC_THREAD);
 274   memset ((char *) new_thread + OFFSETOF (struct thread_state, m_gcprolist),
 275           0, sizeof (struct thread_state) - OFFSETOF (struct thread_state,
 276                                                       m_gcprolist));
 277
 278   new_thread->func = function;
 279   new_thread->blocked = 0;
 280   new_thread->initial_specpdl = Qnil;
 281   new_thread->m_last_thing_searched = Qnil; /* copy from parent? */
 282   new_thread->m_saved_last_thing_searched = Qnil;
 283   new_thread->m_current_buffer = current_thread->m_current_buffer;
 284   new_thread->stack_bottom = &stack_pos;
 285
 286   for (p = specpdl; p != specpdl_ptr; ++p)
 287     {
 288       if (!p->func)
 289         {
 290           Lisp_Object sym = p->symbol;
 291           if (!SYMBOLP (sym))
 292             sym = XCAR (sym);
 293           new_thread->initial_specpdl
 294             = Fcons (Fcons (sym, find_symbol_value (sym)),
 295                      new_thread->initial_specpdl);
 296         }
 297     }
 298
 299   /* We'll need locking here.  */
 300   new_thread->next_thread = all_threads;
 301   all_threads = new_thread;
 302
 303   if (pthread_create (&thr, NULL, run_thread, new_thread))
 304     {
 305       /* Restore the previous situation.  */
 306       all_threads = all_threads->next_thread;
 307       error ("Could not start a new thread");
 308     }
 309
 310   return Qnil;
 311 }
 312
 313 /* Get the current thread as a lisp object.  */
 314 Lisp_Object
 315 get_current_thread (void)
 316 {
 317   Lisp_Object result;
 318   XSETTHREAD (result, current_thread);
 319   return result;
 320 }
 321
 322 /* Get the main thread as a lisp object.  */
 323 Lisp_Object
 324 get_main_thread (void)
 325 {
 326   Lisp_Object result;
 327   XSETTHREAD (result, &primary_thread);
 328   return result;
 329 }
 330
 331 /* Is the current an user thread.  */
 332 int
 333 user_thread_p (void)
 334 {
 335   struct thread_state *it = all_threads;
 336   pthread_t self = pthread_self ();
 337   do
 338     {
 339       if (it->pthread_id == self)
 340         return 1;
 341     }
 342   while (it = it->next_thread);
 343
 344   return 0;
 345 }
 346
 347 DEFUN ("make-mutex", Fmake_mutex, Smake_mutex, 0, 0, 0,
 348        doc: /* Make a mutex.  */)
 349      ()
 350 {
 351   Lisp_Object ret;
 352   struct Lisp_Mutex *mutex = allocate_mutex ();
 353   mutex->owner = 0;
 354   XSETMUTEX (ret, mutex);
 355   return ret;
 356 }
 357
 358 DEFUN ("mutex-lock", Fmutex_lock, Smutex_lock, 1, 1, 0,
 359        doc: /* Lock a mutex.  */)
 360      (val)
 361      Lisp_Object val;
 362 {
 363   struct Lisp_Mutex *mutex = XMUTEX (val);
 364   while (1)
 365     {
 366       if (mutex->owner == 0 || mutex->owner == pthread_self ())
 367         {
 368           mutex->owner = pthread_self ();
 369           return Qt;
 370         }
 371
 372       thread_yield ();
 373     }
 374
 375   return Qt;
 376 }
 377
 378 DEFUN ("mutex-unlock", Fmutex_unlock, Smutex_unlock, 1, 1, 0,
 379        doc: /* Unlock a mutex.  */)
 380      (val)
 381      Lisp_Object val;
 382 {
 383   struct Lisp_Mutex *mutex = XMUTEX (val);
 384   mutex->owner = 0;
 385   return Qt;
 386 }
 387
 388 int
 389 thread_select (n, rfd, wfd, xfd, tmo)
 390   int n;
 391   SELECT_TYPE *rfd, *wfd, *xfd;
 392   EMACS_TIME *tmo;
 393 {
 394   char end;
 395   int ret;
 396   current_thread->blocked = 1;
 397
 398   reschedule (&end, 0);
 399
 400   pthread_mutex_unlock (&global_lock);
 401
 402   ret = select (n, rfd, wfd, xfd, tmo);
 403   current_thread->blocked = 0;
 404
 405   pthread_mutex_lock (&global_lock);
 406   pthread_cond_broadcast (&thread_cond);
 407
 408   while (current_thread->pthread_id != next_thread)
 409     pthread_cond_wait (&thread_cond, &global_lock);
 410
 411   return ret;
 412 }
 413
 414 int
 415 other_threads_p (void)
 416 {
 417   return all_threads->next ? 1 : 0;
 418 }
 419
 420 Lisp_Object
 421 thread_notify_kill_buffer (register struct buffer *b)
 422 {
 423   register Lisp_Object tem;
 424   struct thread_state *it = all_threads;
 425   for (; it; it = it->next_thread)
 426     {
 427       if (b == it->m_current_buffer)
 428         {
 429           register Lisp_Object buf;
 430           XSETBUFFER (buf, it->m_current_buffer);
 431           tem = Fother_buffer (buf, Qnil, Qnil);
 432           it->m_current_buffer = XBUFFER (tem);
 433           if (b == it->m_current_buffer)
 434             return Qnil;
 435         }
 436     }
 437
 438   return Qt;
 439 }
 440
 441 void
 442 init_threads_once (void)
 443 {
 444   primary_thread.size = PSEUDOVECSIZE (struct thread_state, m_gcprolist);
 445   primary_thread.next = NULL;
 446   primary_thread.func = Qnil;
 447   primary_thread.initial_specpdl = Qnil;
 448   XSETPVECTYPE (&primary_thread, PVEC_THREAD);
 449   minibuffer_mutex = Fmake_mutex ();
 450 }
 451
 452 void
 453 init_threads (void)
 454 {
 455   pthread_mutex_init (&global_lock, NULL);
 456   pthread_cond_init (&thread_cond, NULL);
 457   pthread_mutex_lock (&global_lock);
 458
 459   primary_thread.pthread_id = pthread_self ();
 460   primary_thread.blocked = 0;
 461   primary_thread.m_last_thing_searched = Qnil;
 462   next_thread = primary_thread.pthread_id;
 463 }
 464
 465 void
 466 syms_of_threads (void)
 467 {
 468   DEFVAR_LISP ("minibuffer-mutex", &minibuffer_mutex,
 469                doc: /* Mutex for the minibuffer.  */);
 470
 471   defsubr (&Srun_in_thread);
 472   defsubr (&Syield);
 473   defsubr (&Smake_mutex);
 474   defsubr (&Smutex_lock);
 475   defsubr (&Smutex_unlock);
 476 }