producer_consumer.cpp

操作系统生产者消费者多线程的编程实现

// Producer_Consumer.cpp : Defines the entry point for the console application.
//

#include "stdafx.h"


#include <windows.h>
#include <tchar.h>
#include <process.h>
#include <stddef.h>
#include <stdlib.h>
#include <time.h>
#include <iostream.h>

#define DATA_SIZE 256
//#define CONSUMER   4
//#define PRODUCTER  2
#define N	1		//缓冲区大小
#define K	0		//生产总产品数
int pro_instan=0;
int con_instan=0;
int com_place=0;
int Mutex=1;
typedef struct msg_block_tag { /* Message block */
	volatile DWORD	f_ready, f_stop; 
		/* ready state flag, stop flag	*/
	volatile DWORD sequence; /* Message block sequence number	*/
	volatile DWORD nCons, nLost;
	time_t timestamp;
	HANDLE	mguard;	/* Mutex to uard the message block structure	*/
	HANDLE	mready; /* "Message ready" auto-reset event			*/
	DWORD	checksum; /* Message contents checksum		*/
	DWORD	data[DATA_SIZE]; /* Message Contents		*/
} MSG_BLOCK;
/*	One of the following conditions holds for the message block 	*/
/*	  1)	!f_ready || f_stop										*/
/*			 nothing is assured about the data		OR				*/
/*	  2)	f_ready && data is valid								*/
/*			 && checksum and timestamp are valid					*/ 
/*  Also, at all times, 0 <= nLost + nCons <= sequence				*/

/* Single message block, ready to fill with a new message 	*/
MSG_BLOCK mblock = { 0, 0, 0, 0, 0 }; 

DWORD WINAPI produce (void *);
DWORD WINAPI consume (void *);
DWORD WINAPI produce2(void *);
DWORD WINAPI consume2 (void *);
DWORD WINAPI consume3 (void *);
DWORD WINAPI consume4 (void *);
void MessageFill (MSG_BLOCK *);
void MessageDisplay (MSG_BLOCK *);

void ReportError(char * msg, int err_no, bool bl);

int main(int argc, char* argv[])
{
	DWORD Status, ThId;
	HANDLE produce_h, consume_h,produce_h2,consume_h2,consume_h3,consume_h4;
	int Producer,Consumer,Producer_2,Consumer_2,Buffersize,Products;
	/* Initialize the message block mutex and event (auto-reset) */
	mblock.mguard = CreateMutex (NULL, FALSE, NULL);
	mblock.mready = CreateEvent (NULL, FALSE, FALSE, NULL);


	printf("input the number of producer\n");
//	scanf("%d",Producer);
	cin>>Producer;
	printf("input the number of consumer\n");
//	scanf("%d",Consumer);
	cin>>Consumer;

//	cout>>"enter buffer size\n";
//	cin>>Buffersize;

//	cout>>"enter the number of products\n";
//	cin>>Products;

	Producer_2=Producer;
	Consumer_2=Consumer;


	/* Create the two threads */
	if(Producer>0)
	{
	produce_h = (HANDLE)CreateThread(NULL, 0, produce, NULL, 0, &ThId);
	if (produce_h == NULL) 
		ReportError (_T("Cannot create producer thread"), 1, TRUE);
	Producer--;
	}

	if(Producer>0)
	{
	produce_h2 = (HANDLE)CreateThread(NULL, 0, produce2, NULL, 0, &ThId);
	if (produce_h2 == NULL) 
		ReportError (_T("Cannot create producer2 thread"), 1, TRUE);
	Producer--;
	}

	if(Consumer>0)
	{
	consume_h = (HANDLE)CreateThread(NULL, 0, consume, NULL, 0, &ThId);
	if (consume_h == NULL) 
		ReportError (_T("Cannot create consumer thread"), 2, TRUE);
	Consumer--;
	}

	if(Consumer>0)
	{
	consume_h2 = (HANDLE)CreateThread(NULL, 0, consume2, NULL, 0, &ThId);
	if (consume_h2 == NULL) 
		ReportError (_T("Cannot create consumer2 thread"), 2, TRUE);
	Consumer--;
	}

	if(Consumer>0)
	{
	consume_h3 = (HANDLE)CreateThread(NULL, 0, consume3, NULL, 0, &ThId);
	if (consume_h3 == NULL) 
		ReportError (_T("Cannot create consumer3 thread"), 2, TRUE);
	Consumer--;
	}

/*	consume_h4 = (HANDLE)CreateThread(NULL, 0, consume4, NULL, 0, &ThId);
	if (consume_h4 == NULL) 
		ReportError (_T("Cannot create consumer4 thread"), 2, TRUE);
*/	
	/* Wait for the producer and consumer to complete */

	if(Consumer_2>0)
	{
	Status = WaitForSingleObject (consume_h, INFINITE);
	if (Status != WAIT_OBJECT_0) 
		ReportError (_T("Failed waiting for consumer thread"), 3, TRUE);
	Consumer_2--;
	}

	if(Consumer_2>0)
	{
	Status = WaitForSingleObject (consume_h2, INFINITE);
	if (Status != WAIT_OBJECT_0) 
		ReportError (_T("Failed waiting for consumer2 thread"), 3, TRUE);
	Consumer_2--;
	}

	if(Consumer_2>0)
	{
	Status = WaitForSingleObject (consume_h3, INFINITE);
	if (Status != WAIT_OBJECT_0) 
		ReportError (_T("Failed waiting for consumer3 thread"), 3, TRUE);
	Consumer_2--;
	}

/*	Status = WaitForSingleObject (consume_h4, INFINITE);
	if (Status != WAIT_OBJECT_0) 
		ReportError (_T("Failed waiting for consumer4 thread"), 3, TRUE);
*/
	if(Producer_2>0)
	{
	Status = WaitForSingleObject (produce_h, INFINITE);
	if (Status != WAIT_OBJECT_0) 
		ReportError (__T("Failed waiting for producer thread"), 4, TRUE);
	Producer_2--;
	}
	
	if(Producer_2>0)
	{
	Status = WaitForSingleObject (produce_h2, INFINITE);
	if (Status != WAIT_OBJECT_0) 
		ReportError (__T("Failed waiting for producer2 thread"), 4, TRUE);
	Producer_2--;
	}

//	DeleteCriticalSection (&mblock.mguard);
	CloseHandle (mblock.mguard);
	CloseHandle (mblock.mready);

	_tprintf (_T("Producer and consumer threads have terminated\n"));
	_tprintf (_T("Messages produced: %d, Consumed: %d, Known Lost: %d\n"),
		mblock.sequence, mblock.nCons, mblock.nLost);
	return 0;
}

void ReportError(char * msg, int err_no, bool bl){
	_tprintf(msg);
	exit(err_no);
}

DWORD WINAPI produce (void *arg)
/* Producer thread - Create new messages at random intervals */
{
	
	srand ((DWORD)time(NULL)); /* Seed the random # generator 	*/
	if(pro_instan>=K)return 0;
	while (!mblock.f_stop&&pro_instan<K) {
		/* Random Delay */
		if(pro_instan>=K)return 0;
		Sleep(rand()%30000); /* wait a long period for the next message */
			/* Adjust the divisor to change message generation rate */
		/* Get the buffer, fill it */
		
		WaitForSingleObject (mblock.mguard, INFINITE);
		__try 
//		if(Mutex>0)
		{
//			Mutex--;
				_tprintf(_T("\n\nProducer1 Enter"));
			if (!mblock.f_stop&&com_place<N&&pro_instan<K) {
				mblock.f_ready = 0;
				MessageFill (&mblock);
//				MessageDisplay (&mblock);
				mblock.f_ready = 1;
				mblock.sequence++;
				SetEvent(mblock.mready); /* Signal that a message is ready. */
				pro_instan++;
				com_place++;

				_tprintf (_T("\nproducer1 produce one project and producers product"));
				printf("%d",pro_instan);
	//			_tprintf (_T("\nnow there's "));
	//			printf("%d",com_place);
	//			_tprintf(_T(" in the common place"));
				_tprintf(_T("\nbuffer size is %d"),com_place);
			}
//			else if(pro_instan==K)
//			{
//				ReleaseMutex (mblock.mguard);
//				return 0;
//			}
		} 
		__finally
		{ 
			_tprintf(_T("\nProducer1 leave\n"));
			ReleaseMutex (mblock.mguard); 
//			Mutex++;
//			_tprintf(_T("\nProducer1 leave")); 
		}
//		if(pro_instan==K)return 0;
	}

//	_tprintf(_T("all the acts are completed"));
	return 0;
}

DWORD WINAPI produce2 (void *arg)
/* Producer thread - Create new messages at random intervals */
{
	
	srand ((DWORD)time(NULL)); /* Seed the random # generator 	*/
	if(pro_instan>=K)return 0;
	while (!mblock.f_stop&&pro_instan<K) {
		/* Random Delay */
		if(pro_instan>=K)return 0;
		Sleep(rand()%4000); /* wait a long period for the next message */
			/* Adjust the divisor to change message generation rate */
		/* Get the buffer, fill it */
		
		WaitForSingleObject (mblock.mguard, INFINITE);
		__try 
//		if(Mutex>0)
		{
//			Mutex--;
			_tprintf(_T("\n\nProducer2 Enter"));
			if (!mblock.f_stop&&com_place<N&&pro_instan<K) {
				mblock.f_ready = 0;
				MessageFill (&mblock);
//				MessageDisplay (&mblock);
				mblock.f_ready = 1;
				mblock.sequence++;
				SetEvent(mblock.mready); /* Signal that a message is ready. */
				pro_instan++;