Making threads get equal timeslices

Question

I am running this program where I have multiple threads. Three threads are generating signals for the same parent process. There are four handler threads for handling the signals generated by the signal generating threads. I have a monitoring thread which also receives the signals and processes according. However, I have a situation. I can see that the signals are not divided equally. I mean the signals are directed to the same process. I have four handler threads and one monitoring thread waiting for the signal. So anyone of them can receive the signal. I was expecting it to be distributed uniformly. However, I could see that at time a whole burst of signals are received by the handler threads. The very next time the whole burst of signals are handled by the monitor thread. Why isn't it uniform. I have added a sleep call after the handler/ monitor threads complete handling one signal. So as soon as the handler/monitor completes one signal it should have given the other opportunity to handle the next signal. However, the output doesn't show the case

#include <pthread.h>
#include <signal.h>
#include <stdlib.h>
#include <iostream>
#include <unistd.h>
#include <sys/types.h>
#include <sys/time.h>
#include <signal.h>
#include <cstdio>
#include <stdlib.h>

#define NUM_SENDER_PROCESSES 3
#define NUM_HANDLER_PROCESSES 4
#define NUM_SIGNAL_REPORT 10
#define MAX_SIGNAL_COUNT 100000


using namespace std;

volatile int usrsig1_handler_count = 0;
int usrsig2_handler_count = 0;
int usrsig1_sender_count = 0;
int usrsig2_sender_count = 0;
int monitor_count = 0;
int usrsig1_monitor_count = 0;
int usrsig2_monitor_count = 0;
double time_1[10];
double time_2[10];
int lock_1 = 0;
int lock_2 = 0;
int lock_3 = 0;
int lock_4 = 0;
int lock_5 = 0;


double timestamp() {
  struct timeval tp;
  gettimeofday(&tp, NULL);
  return (double)tp.tv_sec + tp.tv_usec / 1000000.;
}

void sleepMs(double seconds) {
  usleep((unsigned int)(seconds*1000000));
}

void *senderfunc(void *parm) {
  srand(time(0));
  while(true) {
    int signal_id = rand()%2 + 1;
    if(signal_id == 1) {
      while(__sync_lock_test_and_set(&lock_3,1) != 0) {
      }
      usrsig1_sender_count++;
      lock_3 = 0;
      kill(getpid(), SIGUSR1);
    } else {
      while(__sync_lock_test_and_set(&lock_4,1) != 0) {
      }
      usrsig2_sender_count++;
      lock_4 = 0;
      kill(getpid(), SIGUSR2);
    }


    int r = rand()%10 + 1;
    double s = (double)r/100;
    sleepMs(s);
  }
}

void *handlerfunc(void *parm)
{
  int *index = (int *)parm;
  sigset_t set;
  sigemptyset(&set);
  //cout << (*index) << endl;
  if((*index) % 2 == 0) {
    sigaddset(&set, SIGUSR1);
  } else {
    sigaddset(&set, SIGUSR2);
  }


  int sig;

  while(true) {
    sigwait(&set, &sig);
    //cout << "Handler" << endl;
    if (sig == SIGUSR1) {
      while(__sync_lock_test_and_set(&lock_1,1) != 0) {
      }
      usrsig1_handler_count++;        
      lock_1 = 0;
    } else if(sig == SIGUSR2) {
      while(__sync_lock_test_and_set(&lock_2,1) != 0) {
      }
      usrsig2_handler_count++;
      lock_2 = 0;
    }

    sleepMs(0.0001);
  }

}

void *monitorfunc(void *parm) {

  sigset_t set;
  sigemptyset(&set);

  sigaddset(&set, SIGUSR1);
  sigaddset(&set, SIGUSR2);

  int sig;  

  while(true) {
    sigwait(&set, &sig);
    //cout << "Monitor" << endl;
    if(sig == SIGUSR1) {
      time_1[usrsig1_monitor_count] = timestamp();
      usrsig1_monitor_count++;
    } else if(sig == SIGUSR2) {
      time_2[usrsig2_monitor_count] = timestamp();
      usrsig2_monitor_count++;
    }
    monitor_count++;
    //cout << monitor_count << endl;

    if(monitor_count == NUM_SIGNAL_REPORT) {
      double difference_1 = 0;
      double difference_2 = 0;
      if(usrsig1_monitor_count > 1) {
        for(int i=0; i<usrsig1_monitor_count-1; i++) {
          difference_1 = difference_1 + time_1[i+1] - time_1[i];
        }
        cout << "Interval SIGUSR1 = " << difference_1/(usrsig1_monitor_count-1)<< endl;
      }

      if(usrsig2_monitor_count > 1) {
        for(int i=0; i<usrsig2_monitor_count-1; i++) {
          difference_2 = difference_2 + time_2[i+1] - time_2[i];
        }
        cout << "Interval SIGUSR2 = " << difference_2/(usrsig2_monitor_count-1) << endl;
      }
      cout << "Count SIGUSR1 = " << usrsig1_sender_count << endl;
      cout << "Count SIGUSR2 = " << usrsig2_sender_count << endl; 
      monitor_count = 0;
      usrsig1_monitor_count = 0;
      usrsig2_monitor_count = 0;
    }

    sleepMs(0.001);

  }
}

int main(int argc, char **argv)
{
  if(argc != 2) {
    cout << "Required parameters missing. " << endl;
    cout << "Option 1 = 1 which means run for 30 seconds" << endl;
    cout << "Option 2 = 2 which means run until 100000 signals" << endl;
    exit(0);
  }

  int option = atoi(argv[1]);
  int i;

  pthread_t handlers[NUM_HANDLER_PROCESSES];
  pthread_t generators[NUM_SENDER_PROCESSES];
  pthread_t monitor;

  sigset_t set;
  sigset_t oldset;
  sigemptyset(&oldset);
  sigemptyset(&set);
  sigaddset(&set, SIGUSR1);
  sigaddset(&set, SIGUSR2);

  pthread_sigmask(SIG_BLOCK, &set, &oldset);


  int handler_mask[4] = {0,1,2,3};
  //Initializing the handler threads
  for(i=0; i<NUM_HANDLER_PROCESSES; i++) {
    pthread_create(&handlers[i], NULL, handlerfunc, (void *)&handler_mask[i]);
  }

  pthread_create(&monitor, NULL, monitorfunc, NULL);

  sleep(5);

  for(i=0; i<NUM_SENDER_PROCESSES; i++) {
    pthread_create(&generators[i], NULL, senderfunc, NULL);
  }

  if(option == 1) {
    cout << "Option 1 " << endl;
    //sleep(30);
    while(true){

    }
    exit(0);
  } else {
    while(true) {
      if((usrsig1_handler_count + usrsig2_handler_count) >= MAX_SIGNAL_COUNT) {
        cout << "Count SIGUSR1 = " << usrsig1_handler_count << endl;
        cout << "Count SIGUSR2 = " << usrsig2_handler_count << endl;
        exit(0);
      } else {
        pthread_yield();
      }
    }
  }
}

Here is my output

HandlerHandler

Handler
Handler
Monitor
Monitor
... whole bunch of Monitor messages
Monitor
Monitor
Handler
Handler
... whole bunch of Handler messages
Handler
Handler

You can see the burst of monitor followed by burst of handler. However, in the code once a handler/monitor handles a signal and goes for sigwait, I have added a sleep call so that the turn is passed to the next available thread. However, this is not helping. This should have made it uniform I guess. However, still monitor gets burst and print. Even though in monitor I have placed sleep after it has finished its job for a signal

Answer 1

What OS are you using? You need a realtime operating system (RTOS) to do that dependably. I had good results using a very small RTOS called On Time RTOS-32 to drive a robot controller. It can do time-slicing with guaranteed latency, or you can choose to do cooperative scheduling. It has a Window emulation set for a subset of the Win API. We used Visual Studio/VC++ as the IDE. There may be a pretty good variety of RTOS's available now. That was almost 10 years ago. Google for RTOS and "realtime operating system."

The alternative, which may be good enough on a fast, dedicated machine, is to write your own "OS within an OS" to schedule the threads "by hand." You would have to change the way the threads wait for assignments.

I am shooting in the dark. Why do you need to balance the load among the threads? What is the task? What is the budget? Does it have to run on particular hardware? On a particular OS? If so which? What happens if it goes plonk? Someone mumbles a bad word and goes to lunch, or a space probe nose dives into the sun?