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 void pthread_pause_handler(const int signal, siginfo_t *info, void *ptr) {
  16         (void)signal; (void)info; (void)ptr;
  17         pthread_user_data_internal_t *td = (pthread_user_data_internal_t *)(info->si_value.sival_ptr);
  18         (void)td;
  19
  20         //Do nothing when there are more signals pending (to cleanup the queue)
  21         sigset_t pending;
  22         sigpending(&pending);
  23         if(sigismember(&pending, PTHREAD_XSIG_STOP)) return;
  24
  25         //Keep waiting for signals until we are supposed to be running
  26         sigset_t sigset;
  27         sigfillset(&sigset);
  28         sigdelset(&sigset, PTHREAD_XSIG_STOP);
  29
  30         //printf("RCV: %p = %p\n", (void *)pthread_user_data_internal(pthread_self()), (void *)td);
  31
  32         //if(!pthread_user_data_internal(pthread_self())->running) {
  33         if(!td->running) {
  34                 sigsuspend(&sigset);
  35         }
  36 }
  37
  38 void pthread_pause_enable() {
  39         //Add thread to internal registry
  40         pthread_user_data_internal(pthread_self());
  41
  42         //Nesting signals too deep is not good for stack
  43         //You can get runtime stats using following command:
  44         //grep -i sig /proc/$(pgrep binary)/status
  45         struct rlimit sigq = {.rlim_cur = 32, .rlim_max=32};
  46         setrlimit(RLIMIT_SIGPENDING, &sigq);
  47
  48         //Prepare signal mask
  49         sigset_t sigset;
  50         sigemptyset(&sigset);
  51         sigaddset(&sigset, PTHREAD_XSIG_STOP);
  52
  53         //Setup signal handler
  54         //signal(PTHREAD_XSIG_STOP, pthread_pause_handler);
  55         const struct sigaction pause_sa = {
  56                 .sa_sigaction = pthread_pause_handler,
  57                 .sa_mask = sigset,
  58                 .sa_flags = SA_SIGINFO,
  59                 .sa_restorer = NULL
  60         };
  61         sigaction(PTHREAD_XSIG_STOP, &pause_sa, NULL);
  62
  63         //Unblock signal
  64         pthread_sigmask(SIG_UNBLOCK, &sigset, NULL);
  65 }
  66
  67 void pthread_pause_disable() {
  68         //Add thread to internal registry
  69         pthread_user_data_internal(pthread_self());
  70
  71         //Block signal
  72         sigset_t sigset;
  73         sigemptyset(&sigset);
  74         sigaddset(&sigset, PTHREAD_XSIG_STOP);
  75         pthread_sigmask(SIG_BLOCK, &sigset, NULL);
  76 }
  77
  78 int pthread_pause_reschedule(pthread_t thread) {
  79         //Send signal to initiate pause handler
  80         //printf("SND: %p\n", (void *)pthread_user_data_internal(thread));
  81         //while(pthread_kill(thread, PTHREAD_XSIG_STOP) == EAGAIN) usleep(1000);
  82         while(pthread_sigqueue(thread, PTHREAD_XSIG_STOP,
  83                 (const union sigval){.sival_ptr=pthread_user_data_internal(thread)}
  84                 ) == EAGAIN) usleep(1000);
  85         return 0;
  86 }
  87
  88 int pthread_extra_yield() {
  89         //Yield to both schedulers
  90         pthread_pause_reschedule(pthread_self());
  91         return pthread_yield();
  92 }
  93
  94 int pthread_pause(pthread_t thread) {
  95         //Set thread as paused and notify it via signal (wait when queue full)
  96         pthread_user_data_internal(thread)->running = 0;
  97         pthread_pause_reschedule(thread);
  98         return 0;
  99 }
 100
 101 int pthread_unpause(pthread_t thread) {
 102         //Set thread as running and notify it via signal (wait when queue full)
 103         pthread_user_data_internal(thread)->running = 1;
 104         pthread_pause_reschedule(thread);
 105         return 0;
 106 }
 107
 108
 109
 110 // Wrappers ///////////////////////////////////////////////////////////
 111
 112
 113 typedef struct pthread_extra_wrapper_t {
 114         void *(*start_routine)(void *);
 115         void *restrict arg;
 116 } pthread_extra_wrapper_t;
 117
 118 void *pthread_extra_thread_wrapper(void *arg) {
 119         pthread_extra_wrapper_t task = *((pthread_extra_wrapper_t*)arg);
 120         free(arg);
 121
 122         //Register new thread to user data structure
 123         pthread_user_data_internal(pthread_self());
 124
 125         //TODO: user_data should do this automaticaly?
 126         pthread_cleanup_push(pthread_user_data_cleanup, (void *)pthread_self());
 127
 128         //Check if we should be running according to pthread_pause sub-scheduler
 129         pthread_pause_reschedule(pthread_self());
 130
 131         return task.start_routine(task.arg);
 132
 133         pthread_cleanup_pop(1); //Needed by pthread_cleanup_push() macro
 134 }
 135
 136 int pthread_extra_create(pthread_t *restrict thread,
 137                           const pthread_attr_t *restrict attr,
 138                           void *(*start_routine)(void *),
 139                           void *restrict arg) {
 140
 141         pthread_extra_wrapper_t *task = malloc(sizeof(pthread_extra_wrapper_t));
 142         assert(task != NULL);
 143         task->start_routine=start_routine;
 144         task->arg=arg;
 145         return pthread_create(thread, attr, pthread_extra_thread_wrapper, task);
 146 }