c/pthread_extra/pthread_pause.c

   1 #define __PTHREAD_EXTRA_INTERNAL
   2
   3 #include <pthread.h>
   4 #include <pthread_extra.h>
   5 #include <signal.h>
   6 #include <semaphore.h>
   7 #include <errno.h>
   8 #include <unistd.h>
   9 #include <sys/resource.h>
  10 //#include <sys/siginfo.h>
  11 //#include <stdio.h>
  12 #include <stdlib.h>
  13 #include <assert.h>
  14 //#include <sys/time.h>
  15
  16 //Mutex that ensures proper serialization of (un)pause calls
  17 //pthread_mutex_t pthread_pause_mutex = PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP;
  18
  19 //Semaphore that ensures proper serialization of (un)pause signals
  20 sem_t pthread_pause_sem;
  21 //Once control to init the semaphore (and possibly other stuff)
  22 pthread_once_t pthread_pause_once_ctrl = PTHREAD_ONCE_INIT;
  23 void pthread_pause_once(void) {
  24         sem_init(&pthread_pause_sem, 0, 1);
  25 }
  26 void pthread_pause_init() { pthread_once(&pthread_pause_once_ctrl, &pthread_pause_once); }
  27
  28 ///When this variable is nonzero, only referenced thread is allowed to run
  29 ///Access has to be protected by pthread_user_data_lock() and pthread_pause_sem;
  30 pthread_t pthread_pause_holder = PTHREAD_XNULL;
  31
  32 void pthread_pause_handler(const int signal, siginfo_t *info, void *ptr) {
  33         (void)signal; (void)info; (void)ptr;
  34         int run = info->si_value.sival_int;
  35         //(void)td;
  36
  37         //Post semaphore to confirm that signal is handled
  38         sem_post(&pthread_pause_sem);
  39
  40         //Keep waiting for signals until we are supposed to be running
  41         if(!run) {
  42                 sigset_t sigset;
  43                 sigfillset(&sigset);
  44                 sigdelset(&sigset, PTHREAD_XSIG_STOP);
  45                 sigsuspend(&sigset);
  46         }
  47 }
  48
  49 void pthread_pause_enable() {
  50         pthread_pause_init(); //Make sure semaphore is init'd
  51
  52         //Nesting signals too deep is not good for stack
  53         //You can get runtime stats using following command:
  54         //grep -i sig /proc/$(pgrep binary)/status
  55         //struct rlimit sigq = {.rlim_cur = 32, .rlim_max=32};
  56         //setrlimit(RLIMIT_SIGPENDING, &sigq);
  57
  58         //Prepare signal mask
  59         sigset_t sigset;
  60         sigemptyset(&sigset);
  61         sigaddset(&sigset, PTHREAD_XSIG_STOP);
  62
  63         //Setup signal handler
  64         //signal(PTHREAD_XSIG_STOP, pthread_pause_handler);
  65         const struct sigaction pause_sa = {
  66                 .sa_sigaction = pthread_pause_handler,
  67                 .sa_mask = sigset,
  68                 .sa_flags = SA_SIGINFO | SA_RESTART,
  69                 .sa_restorer = NULL
  70         };
  71
  72         sigaction(PTHREAD_XSIG_STOP, &pause_sa, NULL);
  73
  74         //Unblock signal
  75         pthread_sigmask(SIG_UNBLOCK, &sigset, NULL);
  76
  77         //Add thread to internal registry
  78         pthread_user_data_internal(pthread_self()); //Only now, when signals are unblocked!
  79 }
  80
  81 void pthread_pause_disable() {
  82         //pthread_user_data_lock();
  83         pthread_pause_init(); //Make sure semaphore is init'd
  84
  85         //Add thread to internal registry
  86         //pthread_user_data_internal(pthread_self()); //DEADLOCKS!
  87
  88         //Block signal
  89         sigset_t sigset;
  90         sigemptyset(&sigset);
  91         sigaddset(&sigset, PTHREAD_XSIG_STOP);
  92
  93         //Make sure all signals are dispatched before we block them
  94         //Maybe not a good idea, causes DEADLOCKS!
  95         //sem_wait(&pthread_pause_sem);
  96         pthread_sigmask(SIG_BLOCK, &sigset, NULL);
  97         //sem_post(&pthread_pause_sem);
  98
  99         //pthread_user_data_unlock();
 100 }
 101
 102 /*
 103 int pthread_pause_reschedule(pthread_t thread) {
 104         //Send signal to initiate pause handler
 105         //printf("SND: %p\n", (void *)pthread_user_data_internal(thread));
 106         //while(pthread_kill(thread, PTHREAD_XSIG_STOP) == EAGAIN) usleep(1000);
 107         while(pthread_sigqueue(thread, PTHREAD_XSIG_STOP,
 108                 (const union sigval){.sival_ptr=pthread_user_data_internal(thread)}
 109                 ) == EAGAIN) usleep(1000);
 110         return 0;
 111 }
 112 */
 113
 114 int pthread_pause_reschedule(pthread_t thread) {
 115         //Decide if the thread should run and signal it
 116         pthread_user_data_lock();
 117
 118         //Wait for semaphore which means signal queue is empty
 119         pthread_pause_init(); //Make sure semaphore is init'd
 120         sem_wait(&pthread_pause_sem);
 121
 122         //Only call this if you already acquired pthread_pause_sem semaphore!!!!
 123         //Otherwise call pthread_pause_reschedule()
 124
 125         //Check if thread has running flag
 126         int run = (pthread_user_data_internal(thread)->running);
 127         //Check if privileged (single thread) mode is active
 128         if(!pthread_equal(pthread_pause_holder, PTHREAD_XNULL) && !pthread_equal(pthread_pause_holder, thread)) {
 129                 run = 0;
 130         }
 131
 132         //Send signal to initiate pause handler (keep trying while SigQueue is full)
 133         //while(pthread_kill(thread, PTHREAD_XSIG_STOP) == EAGAIN) usleep(1000);
 134         //printf("Sched %lu = %d (self: %lu, lck: %lu)\n", thread, run, pthread_self(), pthread_pause_holder);
 135         while(pthread_sigqueue(thread, PTHREAD_XSIG_STOP,
 136                 (const union sigval){.sival_int=run}
 137                 ) == EAGAIN) usleep(1000);
 138
 139         //Wait for signal to be delivered
 140         sem_wait(&pthread_pause_sem);
 141         sem_post(&pthread_pause_sem);
 142         pthread_user_data_unlock();
 143
 144         return 0;
 145 }
 146
 147 int pthread_extra_yield() {
 148         //Yield to both schedulers
 149         pthread_pause_reschedule(pthread_self());
 150         return pthread_yield();
 151 }
 152
 153 ///Pause specified thread (block until it is paused)
 154 int pthread_pause(pthread_t thread) {
 155         pthread_user_data_lock();
 156         //Set thread as paused and notify it via signal (wait when queue full)
 157         pthread_user_data_internal(thread)->running = 0;
 158         pthread_pause_reschedule(thread);
 159         pthread_user_data_unlock();
 160         return 0;
 161 }
 162
 163 ///UnPause specified thread (block until it is unpaused)
 164 int pthread_unpause(pthread_t thread) {
 165         pthread_user_data_lock();
 166         //Set thread as running and notify it via signal (wait when queue full)
 167         pthread_user_data_internal(thread)->running = 1;
 168         pthread_pause_reschedule(thread);
 169         pthread_user_data_unlock();
 170         return 0;
 171 }
 172
 173 ///Enter exclusive mode by pausing everyone else
 174 int pthread_pause_all() {
 175         //printf("Pause ALL\n");
 176         pthread_user_data_lock();
 177         //printf("Pause ALL+\n");
 178         //printf("Pause %p == %p\n", (void *)pthread_pause_holder, (void *)pthread_self());
 179         if(!pthread_equal(pthread_pause_holder,PTHREAD_XNULL)) assert(pthread_equal(pthread_pause_holder, pthread_self()));
 180         pthread_pause_holder = pthread_self();
 181         pthread_user_data_internal_iterate(&pthread_pause_reschedule, NULL);
 182         //printf("Pause ALL!\n");
 183         pthread_user_data_unlock();
 184         return 0;
 185 }
 186
 187 ///Leave exclusive mode by unpausing everyone else
 188 int pthread_unpause_all() {
 189         //printf("UnPause ALL\n");
 190         pthread_user_data_lock();
 191         //printf("UnPause ALL+\n");
 192         if(!pthread_equal(pthread_pause_holder,PTHREAD_XNULL)) assert(pthread_equal(pthread_pause_holder, pthread_self()));
 193         pthread_pause_holder = PTHREAD_XNULL;
 194         pthread_user_data_internal_iterate(&pthread_pause_reschedule, NULL);
 195         //printf("UnPause ALL!\n");
 196         pthread_user_data_unlock();
 197         return 0;
 198 }
 199
 200
 201 // Wrappers ///////////////////////////////////////////////////////////
 202
 203
 204 typedef struct pthread_extra_wrapper_t {
 205         void *(*start_routine)(void *);
 206         void *restrict arg;
 207 } pthread_extra_wrapper_t;
 208
 209 void *pthread_extra_thread_wrapper(void *arg) {
 210         pthread_extra_wrapper_t task = *((pthread_extra_wrapper_t*)arg);
 211         free(arg);
 212
 213         pthread_pause_enable();
 214
 215         //Register new thread to user data structure
 216         //pthread_user_data_internal(pthread_self()); //Perhaps already done in pthread_extra_yield() and pthread_pause_enable()??
 217
 218         //TODO: user_data should do this automaticaly?
 219         pthread_cleanup_push(pthread_user_data_cleanup, (void *)pthread_self());
 220
 221         //Check if we should be running according to pthread_pause sub-scheduler
 222         //pthread_pause_reschedule(pthread_self());
 223         pthread_extra_yield();
 224
 225         //Run actual task
 226         return task.start_routine(task.arg);
 227
 228         pthread_cleanup_pop(1); //Needed by pthread_cleanup_push() macro
 229 }
 230
 231 int pthread_extra_create(pthread_t *restrict thread,
 232                           const pthread_attr_t *restrict attr,
 233                           void *(*start_routine)(void *),
 234                           void *restrict arg) {
 235
 236         pthread_extra_wrapper_t *task = malloc(sizeof(pthread_extra_wrapper_t));
 237         assert(task != NULL);
 238         task->start_routine=start_routine;
 239         task->arg=arg;
 240         return pthread_create(thread, attr, pthread_extra_thread_wrapper, task);
 241 }