lib/pthreads/pthreads.nit

   1 # This file is part of NIT (http://www.nitlanguage.org).
   2 #
   3 # Copyright 2014 Alexis Laferrière <alexis.laf@xymus.net>
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16
  17 # Main POSIX threads support and intro the classes `Thread`, `Mutex` and `Barrier`
  18 module pthreads is
  19         cflags "-pthread -Wno-unknown-attributes"
  20         ldflags "-pthread"
  21         pkgconfig "bdw-gc"
  22         new_annotation threaded
  23 end
  24
  25 #
  26 ## Native part
  27 #
  28 # Nity part at the bottom of the module.
  29 #
  30
  31 in "C Header" `{
  32         #include <pthread.h>
  33 `}
  34
  35 in "C" `{
  36         #include <string.h>
  37
  38         // TODO protect with: #ifdef WITH_LIBGC
  39         // We might have to add the next line to gc_chooser.c too, especially
  40         // if we get an error like "thread not registered with GC".
  41         #ifdef __APPLE__
  42                 #include "TargetConditionals.h"
  43                 #if TARGET_OS_IPHONE == 1
  44                         #define IOS
  45                 #endif
  46         #endif
  47
  48         #if !defined(IOS)
  49                 #define GC_THREADS
  50                 #include <gc.h>
  51         #endif
  52 `}
  53
  54 redef class Sys
  55
  56         # `NativePthread` for running thread
  57         private fun native_pthread_self: NativePthread `{
  58                 pthread_t *id = malloc(sizeof(pthread_t));
  59                 *id = pthread_self();
  60                 return id;
  61         `}
  62
  63         private var self_thread_key = new NativePthreadKey
  64
  65         private var main_thread_cache: nullable MainThread = null
  66         private var main_thread_mutex = new Mutex
  67
  68         # Handle to the program's main thread
  69         fun main_thread: MainThread
  70         do
  71                 var cache = main_thread_cache
  72                 if cache != null then return cache
  73
  74                 main_thread_mutex.lock
  75
  76                 # Recheck if cache has been updated since lock has been unlocked/locked
  77                 cache = main_thread_cache
  78                 if cache != null then
  79                         main_thread_mutex.unlock
  80                         return cache
  81                 end
  82
  83                 # Create a `MainThread` exactly once
  84                 var thread = new MainThread
  85                 thread.native = sys.native_pthread_self
  86                 main_thread_cache = thread
  87
  88                 main_thread_mutex.unlock
  89                 return thread
  90         end
  91 end
  92
  93
  94 # An atomic Int
  95 extern class AtomicInt in "C" `{ int* `}
  96         new(i: Int)`{
  97                 int* v = malloc(sizeof(int));
  98                 return v;
  99         `}
 100
 101         # Get the value and increment it by `i`
 102         fun get_and_increment_by(i: Int): Int `{
 103                 return __sync_fetch_and_add(self, i);
 104         `}
 105
 106         # Get the value and decrement it by `i`
 107         fun get_and_decrement_by(i: Int): Int `{
 108                 return __sync_fetch_and_sub(self, i);
 109         `}
 110
 111         # Get the value and increment it
 112         fun get_and_increment: Int `{
 113                 return __sync_fetch_and_add(self, 1);
 114         `}
 115
 116         # Get the value and decrement it
 117         fun get_and_decrement: Int `{
 118                 return __sync_fetch_and_sub(self, 1);
 119         `}
 120
 121         # Increment by `i` and get the new value
 122         fun increment_by_and_get(i: Int): Int `{
 123                 return __sync_add_and_fetch(self, i);
 124         `}
 125
 126         # Decrement by `i` and get the new value
 127         fun decrement_by_and_get(i: Int): Int `{
 128                 return __sync_sub_and_fetch(self, i);
 129         `}
 130
 131         # Increment the value and get the new one
 132         fun increment_and_get: Int `{
 133                 return __sync_add_and_fetch(self, 1);
 134         `}
 135
 136         # Decrement the value and get the new one
 137         fun decrement_and_get: Int `{
 138                 return __sync_sub_and_fetch(self,1);
 139         `}
 140
 141         # Get the current value
 142         fun value: Int `{
 143                 return *self;
 144         `}
 145
 146 end
 147
 148 private extern class NativePthread in "C" `{ pthread_t * `}
 149
 150         new create(nit_thread: Thread) import Thread.main_intern `{
 151                 pthread_attr_t attr;
 152                 pthread_attr_init(&attr);
 153
 154                 pthread_t thread;
 155                 int r = pthread_create(&thread, &attr, (void * (*)(void *))&Thread_main_intern, nit_thread);
 156
 157                 if (r == 0) {
 158                         pthread_t *pthread = malloc(sizeof(pthread_t));
 159                         memmove(pthread, &thread, sizeof(pthread_t));
 160                         return pthread;
 161                 }
 162                 return NULL;
 163         `}
 164
 165         new create_ex(nit_thread: Thread, attr: NativePthreadAttr) import Thread.main_intern `{
 166                 pthread_t thread;
 167                 int r = pthread_create(&thread, attr, (void * (*)(void *))&Thread_main_intern, nit_thread);
 168
 169                 if (r == 0) {
 170                         pthread_t *pthread = malloc(sizeof(pthread_t));
 171                         memmove(pthread, &thread, sizeof(pthread_t));
 172                         return pthread;
 173                 }
 174                 return NULL;
 175         `}
 176
 177         fun join: nullable Object `{
 178                 void *thread_return;
 179                 pthread_join(*self, &thread_return);
 180                 if(thread_return == NULL) thread_return = null_Object();
 181                 return (nullable_Object)thread_return;
 182         `}
 183
 184         fun equal(other: NativePthread): Bool `{ return pthread_equal(*self, *other); `}
 185
 186         fun kill(signal: Int): Int `{ return pthread_kill(*self, (int)signal); `}
 187 end
 188
 189 private extern class NativePthreadAttr in "C" `{ pthread_attr_t * `}
 190         new `{
 191                 pthread_attr_t attr;
 192                 int r = pthread_attr_init(&attr);
 193                 if (r == 0) {
 194                         pthread_attr_t *pattr = malloc(sizeof(pthread_attr_t));
 195                         memmove(pattr, &attr, sizeof(pthread_attr_t));
 196                         return pattr;
 197                 }
 198                 return NULL;
 199         `}
 200
 201         fun destroy `{
 202                 pthread_attr_destroy(self);
 203         `}
 204
 205         # Most features of this class are still TODO
 206         #
 207         # * pthread_attr_setaffinity_np(3)
 208         # * pthread_attr_setdetachstate
 209         # * pthread_attr_setguardsize
 210         # * pthread_attr_setinheritsched
 211         # * pthread_attr_setschedparam
 212         # * pthread_attr_setschedpolicy
 213         # * pthread_attr_setscope
 214         # * pthread_attr_setstack
 215         # * pthread_attr_setstackaddr
 216         # * pthread_attr_setstacksize
 217 end
 218
 219 private extern class NativePthreadMutex in "C" `{ pthread_mutex_t * `}
 220         new (attr: NativePthreadMutexAttr) `{
 221                 pthread_mutex_t *mutex = malloc(sizeof(pthread_mutex_t));
 222                 int r = pthread_mutex_init(mutex, attr);
 223                 if (r != 0) {
 224                         free(mutex);
 225                         return NULL;
 226                 }
 227                 return mutex;
 228         `}
 229
 230         fun destroy `{ pthread_mutex_destroy(self); `}
 231
 232         fun lock `{ pthread_mutex_lock(self); `}
 233         fun try_lock: Bool `{ return pthread_mutex_trylock(self); `}
 234         fun unlock `{ pthread_mutex_unlock(self); `}
 235 end
 236
 237 private extern class NativePthreadMutexAttr in "C" `{ pthread_mutexattr_t * `}
 238         new `{
 239                 pthread_mutexattr_t *attr = malloc(sizeof(pthread_mutexattr_t));
 240                 int r = pthread_mutexattr_init(attr);
 241                 if (r != 0) {
 242                         free(attr);
 243                         return NULL;
 244                 }
 245                 return attr;
 246         `}
 247
 248         fun destroy `{ pthread_mutexattr_destroy(self); `}
 249
 250         fun set_type_normal `{ pthread_mutexattr_settype(self, PTHREAD_MUTEX_NORMAL); `}
 251         fun set_type_recursive `{ pthread_mutexattr_settype(self, PTHREAD_MUTEX_RECURSIVE); `}
 252         fun set_type_errorcheck `{ pthread_mutexattr_settype(self, PTHREAD_MUTEX_ERRORCHECK); `}
 253
 254         # pthread_mutexattr_setpshared
 255         # pthread_mutexattr_setprotocol
 256         # pthread_mutexattr_setproceiling
 257         # pthread_mutexattr_setrobust_np
 258 end
 259
 260 private extern class NativePthreadKey in "C" `{ pthread_key_t * `}
 261         new `{
 262                 pthread_key_t *key = malloc(sizeof(pthread_key_t));
 263                 int r = pthread_key_create(key, NULL);
 264                 if (r != 0) {
 265                         free(key);
 266                         return NULL;
 267                 }
 268                 return key;
 269         `}
 270
 271         fun get: nullable Object `{
 272                 void *val = pthread_getspecific(*self);
 273                 if (val == NULL) val = null_Object();
 274                 return val;
 275         `}
 276
 277         fun set(val: nullable Object) `{
 278                 pthread_setspecific(*self, val);
 279         `}
 280 end
 281
 282 private extern class NativePthreadCond in "C" `{ pthread_cond_t * `}
 283         new `{
 284                 pthread_cond_t cond;
 285                 int r = pthread_cond_init(&cond, NULL);
 286                 if (r == 0) {
 287                         pthread_cond_t *pcond = malloc(sizeof(pthread_cond_t));
 288                         memmove(pcond, &cond, sizeof(pthread_cond_t));
 289                         return pcond;
 290                 }
 291                 return NULL;
 292         `}
 293
 294         fun destroy `{ pthread_cond_destroy(self); `}
 295
 296         fun signal: Int `{ return pthread_cond_signal(self); `}
 297
 298         fun broadcast `{ pthread_cond_broadcast(self);  `}
 299
 300         fun wait(mutex: NativePthreadMutex) `{ pthread_cond_wait(self, mutex); `}
 301 end
 302
 303 #
 304 ## Nity part
 305 #
 306 # Cannot be extracted from this module because of the callback from C to `Thread::run`
 307 #
 308
 309 # Handle to a thread
 310 #
 311 # Instances of this class are each used to launch and control a thread.
 312 abstract class Thread
 313         super Finalizable
 314
 315         # Type returned by `main`
 316         type E : nullable Object
 317
 318         private var native: nullable NativePthread = null
 319
 320         # Is this thread finished ? True when main returned
 321         var is_done = false
 322
 323         # Main method of this thread
 324         #
 325         # The returned valued is passed to the caller of `join`.
 326         fun main: E do return null
 327
 328         private fun main_intern: E
 329         do
 330                 # Register thread local data
 331                 sys.self_thread_key.set self
 332                 var r = main
 333                 self.is_done = true
 334                 return r
 335         end
 336
 337         # Start executing this thread
 338         #
 339         # Will launch `main` on a different thread.
 340         fun start
 341         do
 342                 if native != null then return
 343                 native = new NativePthread.create(self)
 344         end
 345
 346         # Join this thread to the calling thread
 347         #
 348         # Blocks until the method `main` returns or the target thread calls
 349         # `Sys::thread_exit`. Returns the object returned from the other thread.
 350         #
 351         # Stats the thread if now already done by a call to `start`.
 352         fun join: E
 353         do
 354                 if native == null then start
 355                 var r = native.join
 356                 native = null
 357                 return r.as(E)
 358         end
 359
 360         redef fun finalize
 361         do
 362                 if native == null then return
 363                 native.free
 364                 native = null
 365         end
 366 end
 367
 368 # The main thread of the program
 369 class MainThread
 370         super Thread
 371
 372         private init do end
 373 end
 374
 375 # Exit current thread and return `value` to caller of `Thread::join`
 376 fun exit_thread(value: nullable Object) `{ pthread_exit(value); `}
 377
 378 # Returns the handle to the running `Thread`
 379 fun thread: Thread
 380 do
 381         var key = sys.self_thread_key
 382         var val = key.get
 383         if val == null then
 384                 # This is the original thread, get `Sys::main_thread` and store it
 385                 var thread = sys.main_thread
 386                 key.set thread
 387                 return thread
 388         end
 389
 390         assert val isa Thread
 391         return val
 392 end
 393
 394 # Mutual exclusion synchronization tool
 395 #
 396 # Instances of this class can only be acquired by a single thread at any one
 397 # point in time. Uses the recursive protocol so they can be locked many time by
 398 # the same thread, must then be unlocked as many time.
 399 class Mutex
 400         super Finalizable
 401
 402         private var native: nullable NativePthreadMutex is noinit
 403
 404         init
 405         do
 406                 var attr = new NativePthreadMutexAttr
 407                 attr.set_type_recursive
 408                 native = new NativePthreadMutex(attr)
 409                 attr.destroy
 410                 attr.free
 411         end
 412
 413         # Acquire this lock, wait until it is available
 414         fun lock do native.lock
 415
 416         # Acquire this lock only if it is available
 417         #
 418         # Returns `true` if the lock has been acquired.
 419         fun try_lock: Bool do return native.try_lock
 420
 421         # Release this lock, unblocking all callers of `lock`
 422         fun unlock do native.unlock
 423
 424         redef fun finalize
 425         do
 426                 var native = self.native
 427                 if native != null then
 428                         native.destroy
 429                         native.free
 430                 end
 431                 self.native = null
 432         end
 433 end
 434
 435 # Condition variable
 436 class PthreadCond
 437         super FinalizableOnce
 438
 439         private var native = new NativePthreadCond
 440
 441         # Destroy `self`
 442         redef fun finalize_once do native.destroy
 443
 444         # Signal at least one thread waiting to wake up
 445         fun signal: Int do return native.signal
 446
 447         # Signal all the waiting threads to wake up
 448         fun broadcast do native.broadcast
 449
 450         # Make the current thread waiting for a signal ( `mutex` should be locked)
 451         fun wait(mutex: Mutex) do native.wait(mutex.native.as(not null))
 452 end
 453
 454 # Barrier synchronization tool
 455 #
 456 # Ensures that `count` threads call and block on `wait` before releasing them.
 457 class Barrier
 458         super Finalizable
 459
 460         private var mutex = new Mutex
 461         private var cond: nullable NativePthreadCond = new NativePthreadCond
 462
 463         # Number of threads that must be waiting for `wait` to unblock
 464         var count: Int
 465
 466         private var threads_waiting = 0
 467
 468         # Wait at this barrier and block until there are a `count` threads waiting
 469         fun wait
 470         do
 471                 mutex.lock
 472                 threads_waiting += 1
 473                 if threads_waiting == count then
 474                         threads_waiting = 0
 475                         cond.broadcast
 476                 else
 477                         cond.wait(mutex.native.as(not null))
 478                 end
 479                 mutex.unlock
 480         end
 481
 482         redef fun finalize
 483         do
 484                 var cond = self.cond
 485                 if cond != null then
 486                         cond.destroy
 487                         cond.free
 488                 end
 489                 self.cond = null
 490         end
 491 end
 492
 493 # Print `object` and '\n' with the same system call
 494 redef fun print(object)
 495 do
 496         sys.stdout.write(object.to_s+"\n")
 497 end