使用信号量的共享内存同步
Shared memory sync using semaphore
我有两个代码：PRODUCER (PR) 和 CONSUMER (CO)。有一块内存 (Mat)（准确地说是 3D 矩阵）需要在两个程序之间共享。我目前正在使用基于 共享内存 的 IPC 函数在两个代码之间共享内存 space。
限制条件：
- PR 是
Mat 的所有者并执行迭代以更改矩阵的值。 CO 是 Mat 的用户，只读取值并用于进一步计算
- PR应该先写入数据然后应该等待CO读取并使用[=14=的值] 然后发信号 PR 继续进一步的迭代，它应该像这样继续。
我目前使用的是-
制作人
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#define NOT_READY -1
#define FILLED 0
#define TAKEN 1
#define nx (400)
#define ny (400)
#define nz (400)
struct Memory {
int status;
double u_x[nx+1][ny+2][nz+2];
}
int
main(int argc, char *argv[])
{
key_t ShmKEY;
int ShmID;
struct Memory *ShmPTR;
int i, j, k;
int niter = 5;
int sumX[niter],sumY[niter],sumZ[niter];
ShmKEY = ftok(".", 'x'); // getting the unique identifier key from directory location
ShmID = shmget(ShmKEY, sizeof(struct Memory), IPC_CREAT | 0666);
if (ShmID < 0) {
printf("*** shmget error (server) ***\n");
exit(1);
}
ShmPTR = (struct Memory *) shmat(ShmID, NULL, 0);
if ((int) ShmPTR == -1) {
printf("*** shmat error (server) ***\n");
exit(1);
}
printf("Server attached the memory to its virtual space...\n");
ShmPTR->status = NOT_READY; // setting the status to be not ready before filling it
for (int m = 0; m < niter; m++){
for (i=0; i<=nx; i++) for (j=0; j<=ny+1; j++) for (k=0; k<=nz+1; k++)
ShmPTR->u_x[i][j][k] = m; // filling the array with iteration number (just for depiction purpose)
ShmPTR->status = FILLED; // change the status to Filled
//printf("Please start the client in another window...\n");
while (ShmPTR->status != TAKEN)
sleep(1);
}
printf("Server has detected the completion of its child...\n");
shmdt((void *) ShmPTR);
printf("Server has detached its shared memory...\n");
shmctl(ShmID, IPC_RMID, NULL);
printf("Server has removed its shared memory...\n");
printf("Server exits...\n");
exit(0);
}
消费者
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#define NOT_READY -1
#define FILLED 0
#define TAKEN 1
#define nx (400)
#define ny (400)
#define nz (400)
struct Memory {
int status;
double u_x[nx+1][ny+2][nz+2];
}
int
main(void)
{
key_t ShmKEY;
int ShmID;
struct Memory *ShmPTR;
int i, j, k;
int niter = 5;
int sumX[niter];
ShmKEY = ftok(".", 'x');
ShmID = shmget(ShmKEY, sizeof(struct Memory), 0666);
if (ShmID < 0) {
printf("*** shmget error (client) ***\n");
exit(1);
}
printf("Client has received a shared memory...\n");
ShmPTR = (struct Memory *) shmat(ShmID, NULL, 0);
if ((int) ShmPTR == -1) {
printf("*** shmat error (client) ***\n");
exit(1);
}
printf("Client has attached the shared memory to it's virtual memory space...\n");
for (int m =0; m<niter; m++){
sumX[m] = 0;
while (ShmPTR->status != FILLED)
;
printf("Client found the data is ready, performing sanity check...\n");
// read the integers and check for the sum
for (i=0; i<=nx; i++) for (j=0; j<=ny+1; j++) for (k=0; k<=nz+1; k++)
sumX[m] += ShmPTR->u_x[i][j][k];
printf("Cycle %d : sumX-> %d\n", m,sumX[m);
ShmPTR->status = TAKEN;
printf("Client has informed server data have been taken...\n");
}
shmdt((void *) ShmPTR);
printf("Client has detached its shared memory...\n");
printf("Client exits...\n");
exit(0);
}
我现在正在做的是使用名为 status 的结构成员来防止竞争条件。从我读到现在，semaphores 在 IPC 中允许类似的事情。
问题：如何在此使用信号量使得需要共享的内存 space 只是数组而不是将其包装在具有自定义标志的结构中？
Edit1：或者如果它比此应用程序的信号量更好，那么也许 mutex 就此而言。
Edit2：遵循适用于此代码的@Stargateur 回答，但在 nx、ny 和 nz 是变量的生产代码中，如何定义 共享内存用于由成员组成的结构，该成员是可变长度多维数组？ （当然它会一直存在到调用 shmdt 和 shmctl ）
如果您打算将 PR 和 CO 作为单独的进程，您可以尝试将其中一个进程从另一个进程中 fork同步它们。在这种特殊情况下，我的建议是从 PR 进程中分叉 CO。以下是我的简要想法：
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/shm.h>
#include <sys/stat.h>
#include <sys/mman.h>
int main(int argc, char* argv[])
{
const char *name="SHARED";
const int SIZE = 4096;
pid_t pidA;
pidA = fork();
if (pidA < 0)
{
printf("forkA Failed" );
return 1;
}
else if (pidA == 0) // Child process A
{
// Read from the shared memory object.
exit(0);
}
else // Parent process
{
int shm_fd;
/* pointer to shared memory obect */
void* ptr;
/* create the shared memory object */
shm_fd = shm_open(name, O_CREAT | O_RDWR, 0666);
/* configure the size of the shared memory object */
ftruncate(shm_fd, SIZE);
/* memory map the shared memory object */
ptr = mmap(0, SIZE, PROT_WRITE, MAP_SHARED, shm_fd, 0);
/* write to the shared memory object */
// Wait for child to read matrix
wait(NULL);
printf("Program finished------\n");
}
}
但我们都知道线程更轻量级，因此更可取。您可以使用带有两个线程的信号量，如下所示：
#include <stdio.h>
#include <pthread.h>
#include <semaphore.h>
#include <unistd.h>
sem_t can_read,can_write; // declare two global semaphore
void* threadPR(void* arg)
{
while(true)
{
//wait
sem_wait(&can_write);
//Write to matrix
//signal
sem_post(&can_read);
}
}
void* threadCO(void* arg)
{
while(true)
{
//wait
sem_wait(&can_read);
//Read the matrix
//signal
sem_post(&can_write);
}
}
int main()
{
// initialize the semaphore
sem_init(&mutex, 0, 1);
// declare two threads
pthread_t t1,t2;
pthread_create(&t1,NULL,threadPR,NULL); // Run the PR thread
// do whatever needed before running CO
pthread_create(&t2,NULL,threadCO,NULL); // Run the CO thread
// wait for threads to join
pthread_join(t1,NULL);
pthread_join(t2,NULL);
// free the semaphore
sem_destroy(&mutex);
return 0;
}
您可以将所需的初始化作为全局变量添加到此实现中。
我会建议您使用两个信号量实现您的功能，一个用于解锁生产者，一个用于解锁消费者。
How to use semaphore in this such that the memory space which needs to be shared is just the array(s) and not wrapping it in a struct with a self defined flag ?
是的，但是为什么要将数据和与数据关联的信号量分开？
我会做以下事情：
struct Memory {
sem_t prod;
sem_t cons;
double u_x[nx+1][ny+2][nz+2];
};
// produser
sem_init(&ShmPTR->cons, !0, 0);
sem_init(&ShmPTR->prod, !0, 1);
for (int m = 0; m < niter; m++) {
sem_wait(&ShmPTR->prod);
// ...
sem_post(&ShmPTR->cons);
}
// consumer
for (int m =0; m<niter; m++) {
sem_wait(&ShmPTR->cons);
// ...
sem_post(&ShmPTR->prod);
}
Or maybe mutex for that matter if it is better than semaphore for this
application.
互斥量不能在进程之间共享。
顺便说一句，你使用 int 来迭代数组，你应该使用 size_t
我有两个代码：PRODUCER (PR) 和 CONSUMER (CO)。有一块内存 (Mat)（准确地说是 3D 矩阵）需要在两个程序之间共享。我目前正在使用基于 共享内存 的 IPC 函数在两个代码之间共享内存 space。
限制条件：
- PR 是
Mat的所有者并执行迭代以更改矩阵的值。 CO 是Mat的用户，只读取值并用于进一步计算 - PR应该先写入数据然后应该等待CO读取并使用[=14=的值] 然后发信号 PR 继续进一步的迭代，它应该像这样继续。
我目前使用的是-
制作人
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#define NOT_READY -1
#define FILLED 0
#define TAKEN 1
#define nx (400)
#define ny (400)
#define nz (400)
struct Memory {
int status;
double u_x[nx+1][ny+2][nz+2];
}
int
main(int argc, char *argv[])
{
key_t ShmKEY;
int ShmID;
struct Memory *ShmPTR;
int i, j, k;
int niter = 5;
int sumX[niter],sumY[niter],sumZ[niter];
ShmKEY = ftok(".", 'x'); // getting the unique identifier key from directory location
ShmID = shmget(ShmKEY, sizeof(struct Memory), IPC_CREAT | 0666);
if (ShmID < 0) {
printf("*** shmget error (server) ***\n");
exit(1);
}
ShmPTR = (struct Memory *) shmat(ShmID, NULL, 0);
if ((int) ShmPTR == -1) {
printf("*** shmat error (server) ***\n");
exit(1);
}
printf("Server attached the memory to its virtual space...\n");
ShmPTR->status = NOT_READY; // setting the status to be not ready before filling it
for (int m = 0; m < niter; m++){
for (i=0; i<=nx; i++) for (j=0; j<=ny+1; j++) for (k=0; k<=nz+1; k++)
ShmPTR->u_x[i][j][k] = m; // filling the array with iteration number (just for depiction purpose)
ShmPTR->status = FILLED; // change the status to Filled
//printf("Please start the client in another window...\n");
while (ShmPTR->status != TAKEN)
sleep(1);
}
printf("Server has detected the completion of its child...\n");
shmdt((void *) ShmPTR);
printf("Server has detached its shared memory...\n");
shmctl(ShmID, IPC_RMID, NULL);
printf("Server has removed its shared memory...\n");
printf("Server exits...\n");
exit(0);
}
消费者
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#define NOT_READY -1
#define FILLED 0
#define TAKEN 1
#define nx (400)
#define ny (400)
#define nz (400)
struct Memory {
int status;
double u_x[nx+1][ny+2][nz+2];
}
int
main(void)
{
key_t ShmKEY;
int ShmID;
struct Memory *ShmPTR;
int i, j, k;
int niter = 5;
int sumX[niter];
ShmKEY = ftok(".", 'x');
ShmID = shmget(ShmKEY, sizeof(struct Memory), 0666);
if (ShmID < 0) {
printf("*** shmget error (client) ***\n");
exit(1);
}
printf("Client has received a shared memory...\n");
ShmPTR = (struct Memory *) shmat(ShmID, NULL, 0);
if ((int) ShmPTR == -1) {
printf("*** shmat error (client) ***\n");
exit(1);
}
printf("Client has attached the shared memory to it's virtual memory space...\n");
for (int m =0; m<niter; m++){
sumX[m] = 0;
while (ShmPTR->status != FILLED)
;
printf("Client found the data is ready, performing sanity check...\n");
// read the integers and check for the sum
for (i=0; i<=nx; i++) for (j=0; j<=ny+1; j++) for (k=0; k<=nz+1; k++)
sumX[m] += ShmPTR->u_x[i][j][k];
printf("Cycle %d : sumX-> %d\n", m,sumX[m);
ShmPTR->status = TAKEN;
printf("Client has informed server data have been taken...\n");
}
shmdt((void *) ShmPTR);
printf("Client has detached its shared memory...\n");
printf("Client exits...\n");
exit(0);
}
我现在正在做的是使用名为 status 的结构成员来防止竞争条件。从我读到现在，semaphores 在 IPC 中允许类似的事情。
问题：如何在此使用信号量使得需要共享的内存 space 只是数组而不是将其包装在具有自定义标志的结构中？
Edit1：或者如果它比此应用程序的信号量更好，那么也许 mutex 就此而言。
Edit2：遵循适用于此代码的@Stargateur 回答，但在 nx、ny 和 nz 是变量的生产代码中，如何定义 共享内存用于由成员组成的结构，该成员是可变长度多维数组？ （当然它会一直存在到调用 shmdt 和 shmctl ）
如果您打算将 PR 和 CO 作为单独的进程，您可以尝试将其中一个进程从另一个进程中 fork同步它们。在这种特殊情况下，我的建议是从 PR 进程中分叉 CO。以下是我的简要想法：
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/shm.h>
#include <sys/stat.h>
#include <sys/mman.h>
int main(int argc, char* argv[])
{
const char *name="SHARED";
const int SIZE = 4096;
pid_t pidA;
pidA = fork();
if (pidA < 0)
{
printf("forkA Failed" );
return 1;
}
else if (pidA == 0) // Child process A
{
// Read from the shared memory object.
exit(0);
}
else // Parent process
{
int shm_fd;
/* pointer to shared memory obect */
void* ptr;
/* create the shared memory object */
shm_fd = shm_open(name, O_CREAT | O_RDWR, 0666);
/* configure the size of the shared memory object */
ftruncate(shm_fd, SIZE);
/* memory map the shared memory object */
ptr = mmap(0, SIZE, PROT_WRITE, MAP_SHARED, shm_fd, 0);
/* write to the shared memory object */
// Wait for child to read matrix
wait(NULL);
printf("Program finished------\n");
}
}
但我们都知道线程更轻量级，因此更可取。您可以使用带有两个线程的信号量，如下所示：
#include <stdio.h>
#include <pthread.h>
#include <semaphore.h>
#include <unistd.h>
sem_t can_read,can_write; // declare two global semaphore
void* threadPR(void* arg)
{
while(true)
{
//wait
sem_wait(&can_write);
//Write to matrix
//signal
sem_post(&can_read);
}
}
void* threadCO(void* arg)
{
while(true)
{
//wait
sem_wait(&can_read);
//Read the matrix
//signal
sem_post(&can_write);
}
}
int main()
{
// initialize the semaphore
sem_init(&mutex, 0, 1);
// declare two threads
pthread_t t1,t2;
pthread_create(&t1,NULL,threadPR,NULL); // Run the PR thread
// do whatever needed before running CO
pthread_create(&t2,NULL,threadCO,NULL); // Run the CO thread
// wait for threads to join
pthread_join(t1,NULL);
pthread_join(t2,NULL);
// free the semaphore
sem_destroy(&mutex);
return 0;
}
您可以将所需的初始化作为全局变量添加到此实现中。
我会建议您使用两个信号量实现您的功能，一个用于解锁生产者，一个用于解锁消费者。
How to use semaphore in this such that the memory space which needs to be shared is just the array(s) and not wrapping it in a struct with a self defined flag ?
是的，但是为什么要将数据和与数据关联的信号量分开？
我会做以下事情：
struct Memory {
sem_t prod;
sem_t cons;
double u_x[nx+1][ny+2][nz+2];
};
// produser
sem_init(&ShmPTR->cons, !0, 0);
sem_init(&ShmPTR->prod, !0, 1);
for (int m = 0; m < niter; m++) {
sem_wait(&ShmPTR->prod);
// ...
sem_post(&ShmPTR->cons);
}
// consumer
for (int m =0; m<niter; m++) {
sem_wait(&ShmPTR->cons);
// ...
sem_post(&ShmPTR->prod);
}
Or maybe mutex for that matter if it is better than semaphore for this application.
互斥量不能在进程之间共享。
顺便说一句，你使用 int 来迭代数组，你应该使用 size_t