Multiple Producers and Consumers

A. First Edition

This is a small simulation of my project which has three major tasks, namely graphic displaying, view collecting, mpi

communicating. In order to create a multiple-threaded producers and consumers synchronization mechanism, I choose the 

conditional wait which is superior to busy waiting. After a night and a morning coding, debugging and testing, I finished

this small simulation and were just ready to post it for future reference when developing my project. Then suddenly I fell

into the old "new" trap. If these tasks are a kind of sequential, is there any benefit to develop them in multiple-threaded?

Hence, the sub title suits perfectly the dilemma: the pro and con of multiple-threaded!

B.The problem

D.The major functions

You must understand the basic semantics of conditional wait. As far as I understand the key is the mutual exclusive "wait"

condition to prevent from deadlock. As for "signal" or "broadcast", you can do it at any time even before you unlock mutex 

or not. The extreme case is that an irrelevant thread can also call "signal" or "broadcast".

E.Further improvement

F.File listing

1. multiplethread.cpp

file name: multiplethread.cpp

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>
#include <sys/time.h>

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t  cond  = PTHREAD_COND_INITIALIZER;


void* graphicProc(void*param);
void* viewProc(void*param);
void* mpiProc(void*param);
int  frameNumber = 0;


typedef void*(*ThreadProc)(void*);

const int MaxBufferNumber=5;
const int MaxFrameNumber=1000;
const int MaxBufferSize=1024;
const int MaxGraphicJobNumber=100;
const int MaxViewJobNumber=2000;
const int MaxMpiJobNumber=5000;

int eventIndex[MaxBufferNumber];
int requestIndex[MaxBufferNumber];
int pixelIndex[MaxBufferNumber];

int graphicRuns=0, viewRuns=0, mpiRuns=0;

char eventBuffer[MaxBufferNumber][MaxBufferSize];
char requestBuffer[MaxBufferNumber][MaxBufferSize];
char pixelBuffer[MaxBufferNumber][MaxBufferSize];

pthread_t graphicThread, viewThread, mpiThread;

//for graphicThread, eventIndex is producer, pixelIndex is consumer
int graphicEventIndex=0, graphicPixelIndex=MaxBufferNumber-1;
//for viewThread,eventIndex is producer, request is consumer
int viewEventIndex=0, viewRequestIndex=0;
//for mpiThread, requestIndex is producer, pixelIndex is consumer
int mpiRequestIndex=0, mpiPixelIndex=0;


bool createEvent();

void graphicProcessEvent();
void viewProcessEvent();
void mpiProcessRequest();


struct timeval start, end;

int i;
float result;

int totalSec, totalUsec;


void printStatus()
{
	printf("graphic: event=%d,pixel=%d\n", graphicEventIndex, graphicPixelIndex);
	printf("view: event=%d, request=%d\n", viewEventIndex, viewRequestIndex);
	printf("mpi: request=%d, pixel=%d\n", mpiRequestIndex, mpiPixelIndex);
}

int main()
{
	pthread_mutex_init(&mutex, NULL);
	pthread_cond_init(&cond, NULL);

	sprintf(pixelBuffer[graphicPixelIndex], "empty starting pixel buffer\n");
	gettimeofday(&start, NULL);
	pthread_create(&graphicThread, NULL, graphicProc, NULL);
	pthread_create(&mpiThread, NULL, mpiProc, NULL);
	pthread_create(&viewThread,NULL, viewProc, NULL);

	printf("main thread finishes creating...\n");

	pthread_join(graphicThread, NULL);
	pthread_join(viewThread, NULL);
	pthread_join(mpiThread, NULL);

	printf("graphic runs:%d, view runs:%d, mpi runs:%d\n", graphicRuns, viewRuns, mpiRuns);		
	return 0;
}


void* graphicProc(void* param)
{
	int i;
	int usec;
	printf("graphic thread starts...\n");
	while (frameNumber<MaxFrameNumber)
	{
		//printf("graphic threads starts new frame...\n");
			//eventIndex=%d, pixelIndex=%d\n", graphicEventIndex, graphicPixelIndex);
		//printStatus();
		pthread_mutex_lock(&mutex);

		//only one case we stop graphic thread, that is when no space for new events
		//another case to speed up view by checking if there is pending new event in view
		while ((graphicEventIndex+1)%MaxBufferNumber==mpiPixelIndex)
			//graphicEventIndex!=viewEventIndex)
		{
			pthread_cond_wait(&cond, &mutex);
		}

		//if we have event space to add new event
			//createNewEvent();			
		for (i=0; i<MaxGraphicJobNumber; i++)
		{
			sprintf(eventBuffer[graphicEventIndex], "new event#%d of frame number %d\n",
				graphicEventIndex, frameNumber);
		}
		graphicEventIndex=(graphicEventIndex+1)%MaxBufferNumber;
		
		//this is only meaningful for very first frame
		if ((graphicPixelIndex+1)%MaxBufferNumber!=mpiPixelIndex)
		{
			//there is some new incoming pixels to display
			graphicPixelIndex=(graphicPixelIndex+1)%MaxBufferNumber;
		}		
		//printStatus();
		printf("graphic thread is displaying pixel buffer:'%s' of frame number %d\n", 
			pixelBuffer[graphicPixelIndex], frameNumber);
		frameNumber++; 

		pthread_cond_broadcast(&cond);
		pthread_mutex_unlock(&mutex);
		graphicRuns++;
		gettimeofday(&end, NULL);
		usec=(end.tv_sec-start.tv_sec)*1000000+end.tv_usec-start.tv_usec;
		printf("fps:%f\n", 1000000.0/(float)usec);
		memcpy(&start, &end, sizeof(timeval));
	}
}


//general idea: conditional wait conditions are 
//a) whether there is new pending tasks by checking if producer index is equal to consumer index
//b) whether there is more space for production by checking if (producerIndex+1)%MaxBufferNumber
//is equal to following producerIndex
		
void* viewProc(void* param)
{
	int i;
	char buffer[MaxBufferSize];
	printf("view thread starts...\n");
	while (frameNumber<MaxFrameNumber)
	{
		//printf("view thread starts a new frame...\n");
			// eventIndex=%d, requestIndex=%d...\n",viewEventIndex, viewRequestIndex);
		//printStatus();
		pthread_mutex_lock(&mutex);
		
		//if only stops when both condition are true
		//a)there is no more space for new request buffer
		//b)there is no event
		while ((viewRequestIndex+1)%MaxBufferNumber==mpiPixelIndex||
			viewEventIndex==graphicEventIndex)//no event
		{
			pthread_cond_wait(&cond, &mutex);
		}
		//viewProcessRequest();
		
		if (viewEventIndex!=graphicEventIndex)
		{
			//there is some new event and starting process event
			sprintf(buffer, " and view thread process this event#%d at frame#%d\n", viewEventIndex,
				frameNumber);
			strcat(eventBuffer[viewEventIndex], buffer);
			viewEventIndex=(viewEventIndex+1)%MaxBufferNumber;//finish processing event
			
		}
		//if there is still space to put new request in request buffer and
		//if there exists new pending event waiting for copying to request buffer
		if ((viewRequestIndex+1)%MaxBufferNumber!=mpiPixelIndex&&
			viewEventIndex!=viewRequestIndex)
		{
			//there is new processed event and there is space for new request
			for (i=0; i<MaxViewJobNumber; i++)
			{
				strcpy(requestBuffer[viewRequestIndex], eventBuffer[viewRequestIndex]);
			}
			viewRequestIndex=(viewRequestIndex+1)%MaxBufferNumber;//finish processing event
		}

		//printf("view thread finishes one frame...\n");
		pthread_cond_broadcast(&cond);
		pthread_mutex_unlock(&mutex);
		viewRuns++;
	}
}

void* mpiProc(void* param)
{
	int i;
	char buffer[MaxBufferSize];
	printf("mpi thread starts...\n");
	while (frameNumber<MaxFrameNumber)
	{
		//printf("mpi thread starts a new frame...\n");
			//requestIndex=%d, pixelIndex=%d\n",mpiRequestIndex, mpiPixelIndex);
		//printStatus();
		pthread_mutex_lock(&mutex);
		
		//it only stops for both two conditions
		//a) no new request
		//b) no more space to write pixel results
		while (mpiRequestIndex==viewRequestIndex||(mpiPixelIndex+1)%MaxBufferNumber==graphicEventIndex)
		{
			pthread_cond_wait(&cond, &mutex);
		}
		//mpiProcessRequest();
		if (mpiRequestIndex!=viewRequestIndex)
		{
			//a new request in buffer waiting for processing
			sprintf(buffer, " and mpi thread process this request#%d at frame number#%d\n",
				mpiRequestIndex, frameNumber);
			strcat(requestBuffer[mpiRequestIndex], buffer);
			mpiRequestIndex=(mpiRequestIndex+1)%MaxBufferNumber;
		}
		//this needs two conditions, 
		//a)there is pending finished request to be copied to pixel buffer
		//b)there is space for mpi thread to copy which means graphic thread is not processing the
		//target pixel buffer
		if ((mpiPixelIndex+1)%MaxBufferNumber!=graphicEventIndex&&mpiRequestIndex!=mpiPixelIndex)
		{
			for (i=0; i<MaxMpiJobNumber; i++)
			{
				strcpy(pixelBuffer[mpiPixelIndex], requestBuffer[mpiPixelIndex]);
			}
			mpiPixelIndex=(mpiPixelIndex+1)%MaxBufferNumber;
		}
		//printf("mpi thread finishes one frame...\n");
		
		pthread_cond_broadcast(&cond);
		pthread_mutex_unlock(&mutex);
		mpiRuns++;
	}
}

		

The result is like this:

....

....