Задача "Производители-Потребители", реализация с помощью Events и циклического буфера

@sergeyotro · Регистрация: 08.10.2011

Доброго времени суток.

Написал программу для решения сабжевой задачи, с условиями 1 производитель, 1 потребитель.
Программа работает, потоки синхронизируются, тупиков не возникает.

Однако, условия поменялись и нужно изменить текст программы таким образом, что бы производителей было несколько, а потребитель один. Попытался реализовать такие условия, но не могу уместить в голове синхронизацию не только между производителем-потребителем, но и между производителями, что бы они делили буфер и не пытались туда писать одновременно.

В реализации этого условия и прошу помочь. Улучшения кода по первому условию так же приветствуется.

Ниже - текст программы с 1 производителем и 1 потребителем.

C++

#include <iostream>
#include <conio.h>
#include <windows.h>
 
using namespace std;
 
unsigned long uThrProdID1, uThrConsID;
 
HANDLE hEventWrite, hEventRead;
HANDLE hThrProd1, hThrCons;
void ProducerThread(char name[11]);
void ConsumerThread(void *pParams);
 
#define SIZE_BUF 25
 
int cycleBuf[SIZE_BUF];
int tail = 0;
int head = 0;
int count = 0;
 
void FlushBuf(void);
void PutChar(int sym);
int GetChar(void);
 
int main()
{
    FlushBuf();
 
    hEventWrite = CreateEvent(NULL, FALSE, FALSE, NULL);
    hEventRead = CreateEvent(NULL, FALSE, TRUE, NULL);
 
    hThrProd1 = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)ProducerThread, "Producer 1", 0, &uThrProdID1);
 
    hThrCons = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)ConsumerThread, NULL, 0, &uThrConsID);
 
    getch();
    return 0;
}
 
void ProducerThread(char name[11])
{
    WaitForSingleObject(hEventRead, INFINITE);
    while(1)
    {
        Sleep(rand()%500);
 
        while (count < SIZE_BUF)
            PutChar(rand());
 
        cout <<  name << ": There are " << count << " symbols in buf" << endl << endl;
        SetEvent(hEventWrite);
    }
}
 
void ConsumerThread(void *pParams)
{
    WaitForSingleObject(hEventWrite, INFINITE);
    while(1)
    {
        Sleep(rand()%50);
 
        if (count == 0)
        {
            SetEvent(hEventRead);
            WaitForSingleObject(hEventWrite, INFINITE);
        }
 
        GetChar();
        cout << "Consumer: There are " << count << " symbols in buf now." << endl << endl;
    }
}
 
void FlushBuf(void)
{
    tail = 0;
    head = 0;
    count = 0;
}
 
void PutChar(int sym)
{
    if (count < SIZE_BUF)
    {
        cycleBuf[tail] = sym;
        count++;
        tail++;
        if (tail == SIZE_BUF) tail = 0;
    }
}
 
int GetChar(void)
{
    int sym = 0;
    if (count > 0)
    {
        sym = cycleBuf[head];
        count--;
        head++;
        if (head == SIZE_BUF) head = 0;
    }
    return sym;
}

@niXman · 23.10.2011, 20:08

Сообщение от sergeyotro

не могу уместить в голове синхронизацию не только между производителем-потребителем, но и между производителями

использую блокируемую очередь.

@sergeyotro · 23.10.2011, 22:28 **[ТС]**

Сообщение от niXman

использую блокируемую очередь.

Спасибо за подсказку. Надеюсь я её правильно понял.

Вот код изменённой программы, работает правильно. Надеюсь, кому-нибудь пригодится.

C++

#include <iostream>
#include <conio.h>
#include <windows.h>
#define MAX_PRODUCERS 6
#define SIZE_BUF 25
//#define READ_OUTPUT
 
using namespace std;
 
DWORD uThrProdID[MAX_PRODUCERS], uThrConsID;
 
HANDLE hEventWrite, hEventRead, hEventBuf;
HANDLE hThrProdArray[MAX_PRODUCERS], hThrCons;
 
void ProducerThread(char name[11]);
void ConsumerThread(void *pParams);
 
int cycleBuf[SIZE_BUF];
int tail = 0;
int head = 0;
int count = 0;
 
void FlushBuf(void);
void PutChar(int sym);
int GetChar(void);
 
FILE *fw;
 
#ifdef READ_OUTPUT
FILE *fr;
#endif
 
int main()
{
    fw = fopen("bufw.txt", "w");
#ifdef READ_OUTPUT
    fr = fopen("bufr.txt", "w");
#endif
    FlushBuf();
 
    hEventWrite = CreateEvent(NULL, FALSE, FALSE, NULL);
    hEventBuf = CreateEvent(NULL, FALSE, TRUE, NULL);
    hEventRead = CreateEvent(NULL, FALSE, TRUE, NULL);
 
    for ( int i = 0; i < MAX_PRODUCERS; i++ )
    {
        char* prod = new char[11];
        sprintf(prod, "Producer %d", i+1);
        hThrProdArray[i] = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)ProducerThread, prod, 0, &uThrProdID[i]);
    }
 
    hThrCons = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)ConsumerThread, NULL, 0, &uThrConsID);
 
    getch();
 
    fclose(fw);
#ifdef READ_OUTPUT
    fclose(fr);
#endif
 
    CloseHandle(hEventBuf);
    CloseHandle(hEventRead);
    CloseHandle(hEventWrite);
 
    for ( int i = 0; i < MAX_PRODUCERS; i++ )
        CloseHandle(hThrProdArray[i]);
 
    CloseHandle(hThrCons);
    return 0;
}
 
void ProducerThread(char name[11])
{
    while(1)
    {
        WaitForSingleObject(hEventRead, INFINITE);
        Sleep(rand()%500);
        if (count < SIZE_BUF)
        {
            for ( int i = 0; i < 3; i++ )
            {
                PutChar(name[9]);
                putc(name[9], fw);
            }
            putc(13, fw);
        }
        cout <<  name << ": There are " << count << " symbols in buf" << endl << endl;
        SetEvent(hEventWrite);
    }
}
 
void ConsumerThread(void *pParams)
{
    WaitForSingleObject(hEventWrite, INFINITE);
    while(1)
    {
        Sleep(rand()%50);
        if (count == 0)
        {
            SetEvent(hEventRead);
            WaitForSingleObject(hEventWrite, INFINITE);
        }
#ifdef READ_OUTPUT
        putc(GetChar(), fr);
#else
        GetChar();
#endif
        cout << "Consumer: I've read one symbol. There are " << count << " symbols in buf now. " << endl << endl;
        SetEvent(hEventRead);
    }
}
 
void FlushBuf(void)
{
    tail = 0;
    head = 0;
    count = 0;
}
 
void PutChar(int sym)
{
    WaitForSingleObject(hEventBuf, INFINITE);
    if (count < SIZE_BUF)
    {
        cycleBuf[tail] = sym;
        count++;
        tail++;
        if (tail == SIZE_BUF) tail = 0;
    }
    SetEvent(hEventBuf);
}
 
int GetChar(void)
{
    int sym = 0;
    if (count > 0)
    {
        sym = cycleBuf[head];
        count--;
        head++;
        if (head == SIZE_BUF) head = 0;
    }
    return sym;
}

@niXman · 23.10.2011, 23:55

Сообщение от sergeyotro

Надеюсь я её правильно понял.

ну почти... я имел ввиду написать обертку вокруг std::queue в которой push() и pop() защитить мьютексами.

твое решение тоже не плохое.

08.04.2014, 18:51

а в какой среде вы разрабатывали?

@niXman 3210 / 1458 / 73 Регистрация: 09.08.2009 Сообщений: 3,441 Записей в блоге: 2
	23.10.2011, 20:08	2
	Сообщение от sergeyotro не могу уместить в голове синхронизацию не только между производителем-потребителем, но и между производителями использую блокируемую очередь. 1

@niXman 3210 / 1458 / 73 Регистрация: 09.08.2009 Сообщений: 3,441 Записей в блоге: 2
	23.10.2011, 23:55	4
	Сообщение от sergeyotro Надеюсь я её правильно понял. ну почти... я имел ввиду написать обертку вокруг std::queue в которой push() и pop() защитить мьютексами. твое решение тоже не плохое. 0

tanya234
	08.04.2014, 18:51	5
	а в какой среде вы разрабатывали?

Опции темы