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?

Answer 2

OS have no idea that you want to distribute a JOB between multiple threads, so OS run first available thread with highest priority, it continue running that thread until current thread wait for something( user input, disk or network activity or ... ) or another thread with higher priority become available. - in order to boost performance and avoid too much context switch, OS try to dispatch request on same thread( opposite to your need ).

For example consider MonitorThread scheduled for run and its call to sigwait is satisfied, now it do its action (while this thread own CPU, OS have no way to schedule other threads) when its action complete (end of while) it call sigwait, if there is any pending signal there, OS will pick up that signal and go through loop again , this thread will do this process until there is no more pending signal and sigwait block its operation and next time the operation will be repeated with other scheduled thread.

Now in order to dispatch signals equally I have to implement my own mechanism( since OS have no mechanism for this ). I can create a condition_variable for each thread and force each of them first wait for their own condition_variable and then call sigwait and after completing execution it will set condition_variable of next thread and operation go as is!.