Make C or C++ programs to illustrate the Producer and Consumer synchronization problem. You will have to create several processes to simulate the producers and consumers. Use shared memory to implement the shared buffer among producers and consumers. Use semaphore to synchronize the processes. Here are some constraints for the problem.
• A shared buffer with 3 slots, initially are all empty.
• Two producers
– Randomly wait for a period of time and put product into the buffer.
– Wait if the buffer slots are all full
– Repeat 6 times.
• Three consumers
– Randomly wait for a period of time and fetch a product from the buffer.
– Wait if the buffer slots are all empty.
– Repeat 4 times.

1实验目的

制作C或C++程序来说明生产者和消费者同步问题。您必须创建几个流程来模拟生产者和消费者。使用共享内存实现生产者和消费者之间的共享缓冲区。使用信号量来同步进程。

2实验内容

•一个有3个插槽的共享缓冲区，最初都是空的。
•两个生产商
–随机等待一段时间，将产品放入缓冲区。
–如果缓冲槽已满，请等待
–重复6次。
•三个消费者
–随机等待一段时间，然后从缓冲区提取产品。
–如果缓冲槽全部为空，请等待。
–重复4次。
笔记：
•显示缓冲区的状态以及产品放入或移出缓冲区的时间。
•使用过程（非线程）模拟消费者和生产者。
•使用fork（）和CreateProcess（）等系统调用创建新流程。在Linux中使用系统调用，如shmget（）创建共享内存，使用semget（）创建信号量。
•在windows中，共享内存实现为内存映射文件，您可以使用函数CreateFileMapping创建共享内存。你可以找到更多关于https://docs.microsoft.com/en-us/windows/win32/memory/creating-named-shared-memory?redirectedfrom=MSDN.
•您可以使用CreateSemaphore和CreateSmutex在Windows中创建信号量和互斥锁。
•实施Windows版本和Linux版本。

3实验环境

3.1Windows

操作系统：Windows 10
处理器：AMD 3800X

3.2Linux虚拟机

操作系统：Ubantu 20.04.3
虚拟机软件：VMware Workstation 15
虚拟处理器：1个6核

4程序设计和实现

4.1Windows实现

4.1.1函数解释

4.1.1.1 CreateFileMapping()是用于创建一个文件映射内核对象的函数
HANDLE handleFileMapping = CreateFileMapping(INVALID_HANDLE_VALUE, NULL, PAGE_READWRITE, 0, sizeof(sharememory), SHM_NAME);
参数hFile使用INVALID_HANDLE_VALUE，表示在页面文件中创建一个可共享的文件映射，在本实验中用于作为共享内存
参数flProtect使用PAGE_READWRITE，表示以可读、写的方式打开映射
参数dwMaximumSizeLow使用sizeof(shm)，该数据为文件映射最大长度的低32位，表示该文件大小只有在4.1.1定义的数据结构sharememory一样大。
参数lpName使用SHM_NAME该值为宏定义，表示共享内存区名字

4.1.1.2 MapViewOfFile()是用于将一个文件映射对象映射到当前程序地址空间的函数
LPVOID WINAPI MapViewOfFile(
_In_HANDLE hFileMappingObject,
_In_DWORD dwDesiredAccess,
_In_DWORD dwFileOffsetHigh,
_In_DWORD dwFileOffsetLow,
_In_SIZE_T dwNumberOfBytesToMap
);
LPVOID shareMemoryAddress = MapViewOfFile(handleFileMapping, FILE_MAP_ALL_ACCESS, 0, 0, 0);
参数hFileMappingObject使用CreateFileMapping的返回句柄，表示将创建的对应的文件映射对象映射到程序地址空间
参数dwDesiredAccess使用FILE_MAP_ALL_ACCESS，表示可以使用文件所有权限，是与创建文件映射对象相对应的权限

4.1.1.3 UnmapViewOfFile()是用于停止当前程序的一个内存映射的函数
BOOL WINAPI UnmapViewOfFile(
_In_LPCVOID lpBaseAddress
);
参数pFile是函数MapViewOfFile()函数返回的文件映射对象句柄
该函数用于解除当前进程地址空间对一个文件映射对象的映射

4.1.1.4 OpenFileMapping()是用于打开一个已经存在的文件映射对象的函数，返回相应打开的句柄
HANDLE OpenFileMapping(
_In_DWORD dwDesiredAccess,
_In_BOOL bInheritHandle,
_In_LPCSTR lpName
);
参数dwDesireAccess使用FILE_MAP_ALL_ACCESS，表示打开该映射对象时具有全部权限，和创建文件对象对应
参数bInheritHandle使用FALSE，表示由该进程启动的新进程不允许继承该句柄，防止错误发生
参数lpName使用SHM_NAME，表示打开创建的名为SHM_NAME的文件映射对象

4.1.1.5 CreateSemaphore()是用于创建一个信号量的函数，返回对应信号量的句柄
HANDLE CreateSemaphore(
LPSECURITY_ATTRIBUTES lpSemaphoreAttributes,
LONG lInitialCount,
LONG lMaximumCount,
LPCTSTR lpName
);
参数lInitialCount和lMaximumCount分别表示该信号量初始值和最大可以到达的值，实验中设置如下：
sem_empty 初始值：BUFFER_LEN 最大值：BUFFER_LEN
sem_full 初始值：0 最大值：BUFFER_LEN
sem_mutex 初始值：1 最大值：1
参数lpName是信号量的名字

4.1.1.6 CloseHandle()是用于关闭现有已打开句柄的函数
BOOL CloseHandle(
HANDLE hObject
);
参数hFileMapping是函数OpenFileMapping()的返回值，是一个已经打开的文件映射对象句柄
该函数解除了对该进程对文件映射对象句柄的使用，防止内核泄漏

4.1.2程序代码

/**** 2021-11-25* zhj12399* m.cpp 生产者**/
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <windows.h>#define TIME_PRODUCER 6
#define TIME_CONSUMER 4struct buffer
{//定义缓冲区int s[3];//要求是三个共享区int head;int tail;int is_empty;
};struct sharedmemory
{//定义共享内存struct buffer data;//缓冲区HANDLE full;//有数据的缓冲区个数，初值为0HANDLE empty;//表示空缓冲区的个数，初值为kHANDLE mutex;//互斥访问临界区的信号量，初值为1
};HANDLE MakeShared()
{ //创建共享内存，由filemapping实现//创建一个临时文件映射对象HANDLE hMapping = CreateFileMapping(INVALID_HANDLE_VALUE,NULL, PAGE_READWRITE, 0, sizeof(struct sharedmemory), "BUFFER");if (hMapping == NULL){//映射对象无退出程序printf("CreateFileMapping error\n");exit(0);}//在文件映射上创建视图，返回起始虚地址LPVOID pData = MapViewOfFile(hMapping, FILE_MAP_ALL_ACCESS, 0, 0, 0);if (pData == NULL){printf("MapViewOfFile error\n");exit(0);}if (pData != NULL){ZeroMemory(pData, sizeof(struct sharedmemory));}//解除当前地址空间映射UnmapViewOfFile(pData);return (hMapping);
}int main()
{HANDLE hMapping = MakeShared();//打开文件映射HANDLE hFileMapping = OpenFileMapping(FILE_MAP_ALL_ACCESS, FALSE, "BUFFER");if (hFileMapping == NULL){printf("OpenFileMapping error\n");exit(0);}LPVOID pFile = MapViewOfFile(hFileMapping, FILE_MAP_ALL_ACCESS, 0, 0, 0);if (pFile == NULL){printf("MapViewOfFile error\n");exit(0);}// 创建共享内存struct sharedmemory *addr = (struct sharedmemory *) (pFile);addr->data.head = 0;addr->data.tail = 0;addr->data.is_empty = 1;HANDLE empty = CreateSemaphore(NULL, 3, 3, "EMPTY");HANDLE full = CreateSemaphore(NULL, 0, 3, "FULL");HANDLE mutex = CreateMutex(NULL, FALSE, "MUTEX");UnmapViewOfFile(pFile);//停止当前程序的一个内存映射pFile = NULL;CloseHandle(hFileMapping);//关闭现有已打开句柄//创建子进程PROCESS_INFORMATION sub[5];for (int i = 0; i < 2; i++){//生产者printf("Produce %d created.\n", i + 1);TCHAR szFilename[MAX_PATH];TCHAR szCmdLine[MAX_PATH];PROCESS_INFORMATION pi;sprintf(szFilename, "./p.exe");sprintf(szCmdLine, "\"%s\"", szFilename);STARTUPINFO si;ZeroMemory(&si, sizeof(STARTUPINFO));si.cb = sizeof(si);//创建子进程BOOL bCreatOK = CreateProcess(szFilename, szCmdLine, NULL, NULL, FALSE, 0, NULL, NULL, &si, &pi);sub[i] = pi;}//消费者for (int i = 2; i < 5; i++){printf("Consume %d created.\n", i - 1);TCHAR szFilename[MAX_PATH];TCHAR szCmdLine[MAX_PATH];PROCESS_INFORMATION pi;sprintf(szFilename, "./c.exe");sprintf(szCmdLine, "\"%s\"", szFilename);STARTUPINFO si;ZeroMemory(&si, sizeof(STARTUPINFO));si.cb = sizeof(si);//创建子进程BOOL bCreatOK = CreateProcess(szFilename, szCmdLine, NULL, NULL, FALSE, 0, NULL, NULL, &si, &pi);sub[i] = pi;}//等待子进程结束for (int i = 0; i < 5; i++){WaitForSingleObject(sub[i].hProcess, INFINITE);}//关闭子进程句柄for (int i = 0; i < 5; i++){CloseHandle(sub[i].hProcess);}CloseHandle(hMapping);hMapping = INVALID_HANDLE_VALUE;return 0;
}

/**** 2021-11-25* zhj12399* p.cpp 生产者**/
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <windows.h>#define TIME_PRODUCER 6struct buffer
{//定义缓冲区int s[3];//要求是三个共享区int head;int tail;int is_empty;
};struct sharedmemory
{//定义共享内存struct buffer data;//缓冲区HANDLE full;//有数据的缓冲区个数，初值为0HANDLE empty;//表示空缓冲区的个数，初值为kHANDLE mutex;//互斥访问临界区的信号量，初值为1
};//显示缓冲区数据
void CurrentStatus(struct sharedmemory *a)
{printf("Current Data: ");for (int i = a->data.head;;){printf("%d ", a->data.s[i]);i++;i %= 3;if (i == a->data.tail){printf("\n");return;}}
}int main()
{HANDLE hMap = OpenFileMapping(FILE_MAP_ALL_ACCESS, FALSE, "BUFFER");if (hMap == NULL){printf("OpenFileMapping error!\n");exit(0);}LPVOID pFile = MapViewOfFile(hMap, FILE_MAP_ALL_ACCESS, 0, 0, 0);if (pFile == NULL){printf("MapViewOfFile error!\n");exit(0);}struct sharedmemory *addr = (struct sharedmemory *) (pFile);HANDLE full = OpenSemaphore(SEMAPHORE_ALL_ACCESS, FALSE, "FULL"); // 为现有的一个已命名信号机对象创建一个新句柄HANDLE empty = OpenSemaphore(SEMAPHORE_ALL_ACCESS, FALSE, "EMPTY");HANDLE mutex = OpenMutex(SEMAPHORE_ALL_ACCESS, FALSE, "MUTEX"); // 为现有的一个已命名互斥体对象创建一个新句柄。for (int i = 0; i < TIME_PRODUCER; i++){srand(GetCurrentProcessId() + i);Sleep(rand() % 1000);WaitForSingleObject(empty, INFINITE); //P(empty) 申请空缓冲WaitForSingleObject(mutex, INFINITE); //P(mutex) 申请进入缓冲区//向缓冲区添加数据int num = rand() % 1000;addr->data.s[addr->data.tail] = num;addr->data.tail = (addr->data.tail + 1) % 3;addr->data.is_empty = 0;SYSTEMTIME time;GetLocalTime(&time);printf("\nTime: %02d:%02d:%02d:%d\n", time.wHour, time.wMinute, time.wSecond, time.wMilliseconds);printf("Producer %d putting %d\n", GetCurrentProcessId(), num);if (addr->data.is_empty)printf("Empty\n");elseCurrentStatus(addr);ReleaseSemaphore(full, 1, NULL); //V(full) 释放一个产品ReleaseMutex(mutex);             //V(mutex) 退出缓冲区}UnmapViewOfFile(pFile); // 停止当前程序的一个内存映射pFile = NULL;CloseHandle(hMap); // 关闭句柄CloseHandle(mutex);CloseHandle(empty);CloseHandle(full);return 0;
}

/**** 2021-11-25* zhj12399* c.cpp消费者**/
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <windows.h>#define TIME_CONSUMER 4struct buffer
{//定义缓冲区int s[3];//要求是三个共享区int head;int tail;int is_empty;
};struct sharedmemory
{//定义共享内存struct buffer data;//缓冲区HANDLE full;//有数据的缓冲区个数，初值为0HANDLE empty;//表示空缓冲区的个数，初值为kHANDLE mutex;//互斥访问临界区的信号量，初值为1
};//显示缓冲区数据
void CurrentStatus(struct sharedmemory *a)
{printf("Current Data: ");for (int i = a->data.head;;){printf("%d ", a->data.s[i]);i++;i %= 3;if (i == a->data.tail){printf("\n");return;}}
}int main()
{HANDLE hMap = OpenFileMapping(FILE_MAP_ALL_ACCESS, FALSE, "BUFFER");if (hMap == NULL){printf("OpenFileMapping error!\n");exit(0);}LPVOID pFile = MapViewOfFile(hMap, FILE_MAP_ALL_ACCESS, 0, 0, 0);if (pFile == NULL){printf("MapViewOfFile error!\n");exit(0);}struct sharedmemory *addr = (struct sharedmemory *) (pFile);HANDLE full = OpenSemaphore(SEMAPHORE_ALL_ACCESS, FALSE, "FULL");HANDLE empty = OpenSemaphore(SEMAPHORE_ALL_ACCESS, FALSE, "EMPTY");HANDLE mutex = OpenMutex(SEMAPHORE_ALL_ACCESS, FALSE, "MUTEX");for (int i = 0; i < TIME_CONSUMER; i++){srand(GetCurrentProcessId() + i);Sleep(rand() % 1000);WaitForSingleObject(full, INFINITE);  //P(full) 申请一个产品WaitForSingleObject(mutex, INFINITE); //P(mutex) 申请进入缓冲区int num = addr->data.s[addr->data.head];addr->data.head = (addr->data.head + 1) % 3;if (addr->data.head == addr->data.tail)addr->data.is_empty = 1;elseaddr->data.is_empty = 0;SYSTEMTIME time;GetLocalTime(&time);printf("\nTime: %02d:%02d:%02d:%d\n", time.wHour, time.wMinute, time.wSecond, time.wMilliseconds);printf("Consumer %d removing %d\n", GetCurrentProcessId(), num);if (addr->data.is_empty)printf("Empty\n");elseCurrentStatus(addr);ReleaseSemaphore(empty, 1, NULL); //V(empty) 释放一个空缓冲ReleaseMutex(mutex);//V(mutex) 退出缓冲区}UnmapViewOfFile(pFile);pFile = NULL;CloseHandle(hMap);return 0;
}

4.1.3运行结果

首先编译

然后运行

可以看到生产者和消费者在按时间随机的进行P / V

4.2Linux实现

4.2.1函数解释

4.2.1.1定义缓冲区结构体
typedef struct buff
{
int buff[BUFFER_LEN];
int head;
int tail;
int empty;
} buffer;

4.2.1.2 semget()用于获取与某个建关联的信号量集标识
int semget(
key_t key,
int nsems,
int semflg
);
参数key：信号量集的键值
参数nsems：信号量个数
参数IPC_CREAT：由于键值不为IPC_PRIVATE，且键对应的信号量集不存在，在标志中指定IPC_CREAT可以创建新的信号量集

4.2.1.3 semctl()是用于执行在信号量集上的控制操作的函数
int semctl(
int semid,
int semnum,
int cmd,
union semun arg
);
sem_id：函数semget的返回值，标识一个信号量集
semnum：信号量集中的第几个信号量
cmd：控制操作命令，表示信号量初始化置值
arg：一个union semun的变量，对其中val进行赋值，用于对信号量进行初始化

4.2.1.4 semop()是用于信号量的值与相应资源使用情况相关的操作的函数
int semop(
int semid,
struct sembuf *sops,
size_t nsops
);
semid：信号集标识符，函数semget()返回值
sops：指向存储信号操作结构的数组指针
nsops：信号操作结构的数量，大于等于1

4.2.1.5 shmget()是用于创建共享内存对象的函数
int shmget(
key_t key,
size_t size,
int shmflg
);
key：作为共享内存的键值，当该值为0或IPC_PRIVATE时会建立新的共享内存对象
size：表示对该共享内存区域的访问模式及权限
IPC_CREAT：用于shmflg作为标志，当内存中不存在与键匹配的共享内存对象时创建一个共享内存

4.2.1.6 shmat()是用于把共享内存区对象映射到调用进程的地址空间的函数
void *shmat(
int shmid,
const void *shmaddr,
int shmflg
);
shm_id：共享内存标识符，为函数shmget()函数的返回值
shmaddr：该函数返回的一个附加好的共享内存地址

4.2.1.7 shmdt()是用于断开共享内存连接的函数
int shmdt(
const void *shmaddr
);
shmaddr：连接共享内存的起始地址，函数shmat()函数返回值
该函数断开了现进程与共享内存区的连接，为后面删除共享内存区准备

4.2.1.8 shmctl()是用于完成对共享内存控制的函数
int shmctl(
int shmid,
int cmd,
struct shmid_ds *buf
);
shm_id：共享内存标识符，为函数shmget()函数返回值
IPC_RMID：操作命令，表示删除这片共享内存

对信号量集定义P、V操作时，需要使用到sembuf结构体
struct sembuf{
unsigned short int sem_num;
short int sem_op;
short int sem_flg;
};
sem_num：标识信号量集中的第几个信号量，从_0_开始
sem_op：标识对信号量所进行的操作，有如下几种：
大于0：对该信号量执行挂出操作，即_V操作_，增加对值由sem_op决定
小于0：对该信号量执行等待操作，即_P操作_
等于0：表示调用者希望设置值semval为0，若为0则返回，否则信号量的semzcnt加1，阻塞等待
sem_flag：信号量操作属性标志，为0表示正常操作

4.2.2程序代码

/**** 2021-11-25* zhj12399* main.cpp**/
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <time.h>
#include <sys/wait.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <sys/shm.h>//定义缓冲区
struct buffer
{int s[3];//缓冲区内容int head;//缓冲区头指针int tail;//缓冲区尾指针int is_empty;//设置标志位，判断是否buff为空
};//显示缓冲区数据
void showdata(struct buffer *a)
{printf("Current Data:");for (int i = a->head;;){printf("%d ", a->s[i]);i++;i %= 3;if (i == a->tail){printf("\n\n");return;}}
}//P操作
void P(int semid, int n)
{sembuf temp;temp.sem_num = n; //索引temp.sem_op = -1; //操作值，Ptemp.sem_flg = 0; //访问标志semop(semid, &temp, 1);
}//V操作
void V(int semid, int n)
{sembuf temp;temp.sem_num = n; //索引temp.sem_op = 1;  //操作值，Vtemp.sem_flg = 0; //访问标志semop(semid, &temp, 1);
}int main()
{int semid = semget(1111, 3, IPC_CREAT | 0600); //创建信号量if (semid < 0){printf("semget error\n");exit(0);}semctl(semid, 0, SETVAL, 3); //empty信号量初值为3semctl(semid, 1, SETVAL, 0); //full信号量初值为0semctl(semid, 2, SETVAL, 1); //mutex信号量初值为1//申请共享内存int shmid = shmget(2222, sizeof(buffer), IPC_CREAT | 0600);if (shmid < 0){printf("shmget error\n");exit(0);}//将共享段加到当前进程空间buffer *addr = (buffer *) shmat(shmid, 0, 0);if (addr == (void *) -1){printf("shmat error\n");exit(0);}//为缓冲区结构头尾指针赋值addr->head = 0;addr->tail = 0;addr->is_empty = 1;for (int i = 0; i < 2; i++) //生产者{pid_t pid = fork();if (pid < 0){printf("producer fork error\n");exit(0);}if (pid == 0) //创建生产者{// 把共享内存区加到子进程的地址空间addr = (buffer *) shmat(shmid, 0, 0);if (addr == (void *) -1){printf("producer shmat error\n");exit(0);}for (int j = 0; j < 6; j++){srand((unsigned) time(NULL) + getpid());sleep(rand() % 1);P(semid, 0); //申请emptyP(semid, 2); //申请mutexint num = rand() % 1000;addr->s[addr->tail] = num;addr->tail = (addr->tail + 1) % 3;addr->is_empty = 0;time_t t;time(&t);printf("Time: %s", ctime(&t));printf("producer %d put %d\n", i, num);showdata(addr);V(semid, 1); //释放fullV(semid, 2); //释放mutex}shmdt(addr); //将共享段与子进程解除连接exit(0);}}for (int i = 0; i < 3; i++) //消费者{pid_t pid = fork();if (pid < 0){printf("consumer fork error!\n");exit(0);}if (pid == 0) //创建消费者{addr = (buffer *) shmat(shmid, 0, 0);if (addr == (void *) -1){printf("consumer shmat error!\n");exit(0);}for (int j = 0; j < 4; j++){srand((unsigned) time(NULL) + getpid());sleep(rand() % 1);P(semid, 1); //申请fullP(semid, 2); //申请mutexint num = addr->s[addr->head];addr->head = (addr->head + 1) % 3;if (addr->head == addr->tail){//头尾相等时缓冲为空addr->is_empty = 1;}else{addr->is_empty = 0;}time_t t;time(&t);printf("Time: %s", ctime(&t));printf("consumer %d take %d\n", i, num);if (addr->is_empty == 0){showdata(addr);}else{printf("Empty\n\n");}V(semid, 0); //释放emptyV(semid, 2); //释放mutex}shmdt(addr); //将共享段与子进程解除连接exit(0);}}//等待所有子进程完成while (wait(0) != -1);shmdt(addr);semctl(semid, IPC_RMID, 0); //删除信号量shmctl(shmid, IPC_RMID, 0); //删除共享段return 0;
}

4.2.3运行结果