src/thread.c

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