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 <errno.h>
   7 #include <unistd.h>
   8 #include <sys/resource.h>
   9 //#include <sys/siginfo.h>
  10 //#include <stdio.h>
  11 #include <stdlib.h>
  12 #include <assert.h>
  13 //#include <sys/time.h>
  14
  15 ///When this variable is nonzero, only referenced thread is allowed to run
  16 ///Access has to be protected by pthread_user_data_lock()
  17 pthread_t pthread_pause_holder = 0;
  18
  19 void pthread_pause_handler(const int signal, siginfo_t *info, void *ptr) {
  20         (void)signal; (void)info; (void)ptr;
  21         int run = info->si_value.sival_int;
  22         //(void)td;
  23
  24         //Do nothing when there are more signals pending (to cleanup the queue)
  25         sigset_t pending;
  26         sigpending(&pending);
  27         if(sigismember(&pending, PTHREAD_XSIG_STOP)) return;
  28
  29         //Keep waiting for signals until we are supposed to be running
  30         sigset_t sigset;
  31         sigfillset(&sigset);
  32         sigdelset(&sigset, PTHREAD_XSIG_STOP);
  33
  34         //printf("RCV: %p = %p\n", (void *)pthread_user_data_internal(pthread_self()), (void *)td);
  35
  36         //if(!pthread_user_data_internal(pthread_self())->running) {
  37         if(!run) {
  38                 sigsuspend(&sigset);
  39         }
  40 }
  41
  42 void pthread_pause_enable() {
  43         //Add thread to internal registry
  44         //pthread_user_data_internal(pthread_self());
  45
  46         //Nesting signals too deep is not good for stack
  47         //You can get runtime stats using following command:
  48         //grep -i sig /proc/$(pgrep binary)/status
  49         struct rlimit sigq = {.rlim_cur = 32, .rlim_max=32};
  50         setrlimit(RLIMIT_SIGPENDING, &sigq);
  51
  52         //Prepare signal mask
  53         sigset_t sigset;
  54         sigemptyset(&sigset);
  55         sigaddset(&sigset, PTHREAD_XSIG_STOP);
  56
  57         //Setup signal handler
  58         //signal(PTHREAD_XSIG_STOP, pthread_pause_handler);
  59         const struct sigaction pause_sa = {
  60                 .sa_sigaction = pthread_pause_handler,
  61                 .sa_mask = sigset,
  62                 .sa_flags = SA_SIGINFO,
  63                 .sa_restorer = NULL
  64         };
  65         sigaction(PTHREAD_XSIG_STOP, &pause_sa, NULL);
  66
  67         //Unblock signal
  68         pthread_sigmask(SIG_UNBLOCK, &sigset, NULL);
  69 }
  70
  71 void pthread_pause_disable() {
  72         //Add thread to internal registry
  73         //pthread_user_data_internal(pthread_self());
  74
  75         //Block signal
  76         sigset_t sigset;
  77         sigemptyset(&sigset);
  78         sigaddset(&sigset, PTHREAD_XSIG_STOP);
  79         pthread_sigmask(SIG_BLOCK, &sigset, NULL);
  80 }
  81
  82 /*
  83 int pthread_pause_reschedule(pthread_t thread) {
  84         //Send signal to initiate pause handler
  85         //printf("SND: %p\n", (void *)pthread_user_data_internal(thread));
  86         //while(pthread_kill(thread, PTHREAD_XSIG_STOP) == EAGAIN) usleep(1000);
  87         while(pthread_sigqueue(thread, PTHREAD_XSIG_STOP,
  88                 (const union sigval){.sival_ptr=pthread_user_data_internal(thread)}
  89                 ) == EAGAIN) usleep(1000);
  90         return 0;
  91 }
  92 */
  93
  94 int pthread_pause_reschedule(pthread_t thread) {
  95         //Decide if the thread should run
  96
  97         pthread_user_data_lock();
  98         //Check if thread has running flag
  99         int run = (pthread_user_data_internal(thread)->running);
 100         //Check if privileged (single thread) mode is active
 101         if((pthread_pause_holder != 0) && !pthread_equal(pthread_pause_holder, thread)) {
 102                 run = 0;
 103         }
 104         pthread_user_data_unlock();
 105
 106         //Send signal to initiate pause handler (keep trying while SigQueue is full)
 107         //while(pthread_kill(thread, PTHREAD_XSIG_STOP) == EAGAIN) usleep(1000);
 108         while(pthread_sigqueue(thread, PTHREAD_XSIG_STOP,
 109                 (const union sigval){.sival_int=run}
 110                 ) == EAGAIN) usleep(1000);
 111         return 0;
 112 }
 113
 114 int pthread_extra_yield() {
 115         //Yield to both schedulers
 116         pthread_pause_reschedule(pthread_self());
 117         return pthread_yield();
 118 }
 119
 120 int pthread_pause(pthread_t thread) {
 121         //Set thread as paused and notify it via signal (wait when queue full)
 122         pthread_user_data_lock();
 123         pthread_user_data_internal(thread)->running = 0;
 124         pthread_user_data_unlock();
 125         pthread_pause_reschedule(thread);
 126         return 0;
 127 }
 128
 129 int pthread_unpause(pthread_t thread) {
 130         //Set thread as running and notify it via signal (wait when queue full)
 131         pthread_user_data_lock();
 132         pthread_user_data_internal(thread)->running = 1;
 133         pthread_user_data_unlock();
 134         pthread_pause_reschedule(thread);
 135         return 0;
 136 }
 137
 138 int pthread_pause_all() {
 139         pthread_user_data_lock();
 140         if(pthread_pause_holder!=0) assert(pthread_equal(pthread_pause_holder, pthread_self()));
 141         pthread_pause_holder = pthread_self();
 142         pthread_user_data_unlock();
 143         //todo reschedule all
 144         return 0;
 145 }
 146
 147 int pthread_unpause_all() {
 148         pthread_user_data_lock();
 149         if(pthread_pause_holder!=0) assert(pthread_equal(pthread_pause_holder, pthread_self()));
 150         pthread_pause_holder = 0;
 151         pthread_user_data_unlock();
 152         //todo reschedule
 153         return 0;
 154 }
 155
 156
 157 // Wrappers ///////////////////////////////////////////////////////////
 158
 159
 160 typedef struct pthread_extra_wrapper_t {
 161         void *(*start_routine)(void *);
 162         void *restrict arg;
 163 } pthread_extra_wrapper_t;
 164
 165 void *pthread_extra_thread_wrapper(void *arg) {
 166         pthread_extra_wrapper_t task = *((pthread_extra_wrapper_t*)arg);
 167         free(arg);
 168
 169         //Register new thread to user data structure
 170         pthread_user_data_internal(pthread_self());
 171
 172         //TODO: user_data should do this automaticaly?
 173         pthread_cleanup_push(pthread_user_data_cleanup, (void *)pthread_self());
 174
 175         //Check if we should be running according to pthread_pause sub-scheduler
 176         //pthread_pause_reschedule(pthread_self());
 177         pthread_extra_yield();
 178
 179         //Run actual task
 180         return task.start_routine(task.arg);
 181
 182         pthread_cleanup_pop(1); //Needed by pthread_cleanup_push() macro
 183 }
 184
 185 int pthread_extra_create(pthread_t *restrict thread,
 186                           const pthread_attr_t *restrict attr,
 187                           void *(*start_routine)(void *),
 188                           void *restrict arg) {
 189
 190         pthread_extra_wrapper_t *task = malloc(sizeof(pthread_extra_wrapper_t));
 191         assert(task != NULL);
 192         task->start_routine=start_routine;
 193         task->arg=arg;
 194         return pthread_create(thread, attr, pthread_extra_thread_wrapper, task);
 195 }