stream processing with pthread

stream-processing.c

// http://archive.linux.or.jp/JM/html/glibc-linuxthreads/man3/pthread_cond_wait.3.html
// http://d.hatena.ne.jp/mononoco/20080522/1211452506
// http://codezine.jp/article/detail/1894?p=2
#include<stdio.h>
#include<pthread.h>

typedef unsigned char byte_t;
typedef struct {
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    int id;
    byte_t* ptr;
    int end;
} thread_arg_t;

#define NTHREADS (4)
byte_t buffer[8];


void* slave(void* _arg)
{
    thread_arg_t *arg = (thread_arg_t*)_arg;
    byte_t *ptr = arg->ptr;
    
    for (;;)
    {
        pthread_mutex_lock(&(arg->mutex));
        
        pthread_cond_wait(&(arg->cond), &(arg->mutex));
        printf("pong %d\n", arg->id);
        
        // process audio buffer
        ptr[0] *= 2;
        ptr[1] *= 2;
        
        //mprintf("%d %d\n", ptr[0], ptr[1]);
        
        pthread_mutex_unlock(&(arg->mutex));
    }
    
    pthread_exit(0);
}

void fill_data_from(byte_t *ptr)
{
    static int count = 0;
    // 
    //
    //
    sleep(1);
    ptr[0] = count++;
    sleep(1);
    ptr[1] = count++;
}

int main()
{
    int i;
    int end;
    pthread_t threads[NTHREADS];
    thread_arg_t args[NTHREADS];
    
    // initialize thread parameter and create thread object
    for (i = 0; i < NTHREADS; i++)
    {
        args[i].id = i; 
        args[i].ptr = buffer + (8 / NTHREADS * i);
        pthread_mutex_init(&(args[i].mutex), NULL);
        pthread_cond_init(&(args[i].cond), NULL);
        pthread_create(threads + i, NULL, slave, (void*)(args + i));
    }
    
    // processing audio data
    // main thread as master
    end = 0;
    while (!end)
    {
        for (i = 0; !end && i < NTHREADS; i++)
        {
            fill_data_from(args[i].ptr);
            // notify
            printf("ping %d\n", i);
            pthread_cond_signal(&(args[i].cond));
        }
    }
    
    // finalize
    for (i = 0; i < NTHREADS; i++)
        args[i].end = 1;
    for (i = 0; i < NTHREADS; i++)
        pthread_join(threads[i], NULL);
}