POSIX timer with sigev_notify = SIGEV_THREAD_ID method

Question

In a multithreaded scenario, I am trying to implement a POSIX timer in which on timer expiration the next thread of the same process should be awakened (at a time only one thread is running and all others are in block state). In the timer, I am using sigev_notify = SIGEV_THREAD_ID as neither I want any handler to service the signal nor I want to create a new thread after the timer expiration .

//Import
#define _GNU_SOURCE
#define _POSIX_C_SOURCE 199309
#include <sched.h>
#include <unistd.h>
#include <sys/wait.h>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <signal.h>
#include <errno.h>
#include <semaphore.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <syscall.h>
#define NUM_THREADS 10

#define CLOCKID CLOCK_REALTIME

int ret;
//pthread_cond_t      condA[NUM_THREADS+1]  = PTHREAD_COND_INITIALIZER;
pthread_cond_t      condA  = PTHREAD_COND_INITIALIZER;
pthread_mutex_t     mutex  = PTHREAD_MUTEX_INITIALIZER;

sem_t sem[NUM_THREADS];
sem_t mute;

timer_t timer1;
pthread_t tid[NUM_THREADS];
int state = 0;
int thread_count = 1;
int arr_tid[NUM_THREADS];
struct itimerspec new_value, old_value;

struct sigaction action;
struct sigevent sevent;
sigset_t set;
int signum = SIGALRM;

void *threadA(void *data_)
{
    cpu_set_t my_set;        
    CPU_ZERO(&my_set);       
    CPU_SET(2, &my_set);     
    sched_setaffinity(0, sizeof(cpu_set_t), &my_set);

    //struct itimerspec new_value, old_value;

    int i = 0, value;
    int sid;
    FILE *fp;
    fp=fopen("ipc.out","a");    

    long int loopNum;
    int turn = (intptr_t)data_;
    struct timespec tval_result, tval_result2;

    if(thread_count < NUM_THREADS)
    {
        thread_count++;
        sid = syscall(SYS_gettid);
        arr_tid[turn] = sid;
        fprintf(fp,"thread_%d %d\n", turn, sid); 
        //printf("Blocked %d ->%d\n", turn, thread_count );
        pthread_mutex_lock(&mutex);
        pthread_cond_wait(&condA, &mutex);
        pthread_mutex_unlock(&mutex);
    }
    else
    {
        arr_tid[turn] = syscall(SYS_gettid);
    }

    for (value = 0; value < NUM_THREADS; ++value)
    {   
        printf("%d\n",arr_tid[value] );
    } 
    //printf("rpg\n");
    pthread_mutex_lock(&mutex);
    pthread_cond_broadcast(&condA);
    pthread_mutex_unlock(&mutex);
    //printf("unblocked\n");
    fclose(fp); 

    if (turn > 0)
        {
            if (sigwait (&set, &signum) == -1)
                perror ("sigwait");
            //sleep(1);
            //printf("thread %d is sleeping\n", turn);
        }
    while(1)
    {
        ret = sem_wait(&sem[turn]);
        if (ret)
        {
            printf("Error in Sem Post\n");
        }
        //printf("this isn't the end of the world!!!\n");
        sevent.sigev_notify = SIGEV_THREAD_ID;
        sevent._sigev_un._tid = arr_tid[(turn+1)%10];
        sevent.sigev_signo = signum;

        sigemptyset(&set);
        sigaddset(&set, signum);
        sigprocmask(SIG_BLOCK, &set, NULL);

        printf("Thread # -> %d\n", turn);
        clock_gettime(CLOCKID, &tval_result);
        do
        {
            clock_gettime(CLOCKID, &tval_result2);  
        } while( (tval_result2.tv_sec - tval_result.tv_sec)*1000000000+(tval_result2.tv_nsec - tval_result.tv_nsec)<=12000);       
               //printf("Timestamp : %ld %ld\n", tval_result2.tv_sec, tval_result2.tv_nsec);
       // printf("Before creating timer\n");

        new_value.it_interval.tv_sec = 0;
        new_value.it_interval.tv_nsec = 0;
        new_value.it_value.tv_sec = 0;
        new_value.it_value.tv_nsec = 15000;
        printf("next thread to be signalled %d\n", arr_tid[turn+1]);
        if (timer_settime (timer1, 0, &new_value, NULL) == -1)
            perror ("timer_settime");
        printf("yy\n");
        ret = sem_post(&sem[(state+1)%NUM_THREADS]);
        if (ret)
        {
            printf("Error in Sem Post\n");
        }
        state++; 
        //printf("yy\n");
        //sleep(1);
        if (sigwait (&set, &signum) == -1)
            perror ("sigwait");
    }
}

int main(int argc, char *argv[])
{
    int data = 0;
    int err, i;
    int sid = syscall(SYS_gettid);
    //struct itimerspec new_value, old_value;

    FILE *fp;
    fp=fopen("ipc.out","a");    
    fprintf(fp,"Mainthread %d\n",sid);  
    fclose(fp); 
    if (timer_create (CLOCK_REALTIME, &sevent, &timer1) == -1)
        perror ("timer_create");

    sem_init(&sem[0], 0, 1); 
    //sem_init(&sem[1], 0, 0); 
    //sem_init(&sem[2], 0, 0); 
    for ( i = 1; i < NUM_THREADS; ++i)
        {
            sem_init(&sem[i], 0, 0); 
        }
    while(data < NUM_THREADS)
    {
        //create our threads
        err = pthread_create(&tid[data], NULL, threadA, (void *)(intptr_t)data);
        if(err != 0)
            printf("\ncan't create thread :[%s]", strerror(err));
        data++;
    }
    pthread_exit(NULL);
}

Compilation: $ gcc filename.c -lrt -lpthread

I have used semaphores to synchronize the threads so that threads execute in a particular sequence.

I am not getting the desired output. It should be like after 15 microseconds the timer should expire and next thread should be awakened but it's not happening. I have used sigwait() to block the threads. Are there any other alternatives to block a thread and awake them using a timer? What are the signals that can be assigned in signum? Can I use pthread_cond_signal() instead of SIGALRM?

Answer 1

You are expecting all threads to get signaled one at a time on timer expiry, in sequential fashion, but the timer setting (timer_settime) has been done in all threads, which breaks the rule.

This is what needs to be done: The timer and the setting of time, needs to happen at the process level single time, and not per-thread.

Next, create all the threads, and store the thread-ids in an array, and make them wait based on some condition (pthread_cond_wait).

When the timer expires, and sends a signal, loop through the thread-ids and wake up, one thread at a time.

Answer 2

Um.. OK.

Initialize your semaphore to one unit and have all the threads wait for it.

If a thread gets the unit, it stores the current 'startTime' time in some static/gobal/whatever, (safe, since only one thread can have the semaunit at a time), and goes away to do its thing/s. When it's done, it gets the curent 'endTime' time, claculates how long it took as 'endTime-startTime' and subtracts that from the desired inter-thread interval time - now has how long is left before the next thread can go - 'remainingInterval'. If remainingInterval is less than 0 it makes it 0. It converts remainingInterval to whatever unit is required for usleep()/Sleep() and sleeps for that long, then sends its unit to the semaphore. Another thread can then run, and the thread that just did its work can loop round and wait on the semaphore again, if it wishes.

It is just not possible for more than one thread to run at a time, no matter how long/sort/whatever the job/s take.

No explicit timer, mutex, condvar etc etc is required, just one semaphore.