通过共享内存和互斥锁、条件变量实现进程同步

目的：
使用 pthread_mutex_t 和 pthread_cond_t 以及共享内存、内存映射 达到跨进程通信的目的

步骤：

• 通过 shm_open 、ftrancate、mmap将共享内存进行内存映射
• 将 pthread_mutex_t 和 pthread_cond_t 保存在共享内存中，并且设置 PTHREAD_SHARED 属性，以达到跨进程使用的目的
• 消费者在没有通知的情况下超时等待10s后输出共享内存数据
• 生产者产生数据后，可以选择是否进行 notify
• 
  
  

结论：

• 通过设置 PTHREAD_SHARED 属性， pthread_mutex_t 和 pthread_cond_t 可以跨进程使用
• 如果生产者不进行notify，消费者将等到超时之后才能被唤醒
• 由于消费者进行wait的时候，并不拥有锁，因此如果在有一个进程/线程进行wait的时候，另外一个进程是可以获取锁并且进行条件唤醒的。此时，wait 的线程/进程将被唤醒，但是在 notify 到 wait 之间的时间间隔中，其他线程/进程也可以获取这个锁。因此，如果我们将 wait 和 nofity 方在独立进程的循环中，此时可以看到 notify 和 wait 并不是一对一交替执行的
• (补充)mmap 返回的指针，是可以被有亲缘关系的进程使用的
• (补充)对相同名字的共享内存文件进行内存映射，跨进程得到的内存地址也是相同的
  

1. #include
   
2. #include
   
3. #include
   
4. #include
   
5. #include
   
6. #include "sys.h"
   
7. #include
   
8. 
   
9. #include         // for shm_open
   
10. #include         /* For mode constants */
    
11. #include            /* For O_* constants */
    
12. #include          // for pthread_xx
    
13. #include  // for assert
    
14. 
    
15. struct SHM_MUTEX
    
16. {
    
17.     pthread_mutex_t _mutex;
    
18.     pthread_cond_t _cond;
    
19. };
    
20. 
    
21. 
    
22. class SHM_SYNC_COND
    
23. {
    
24. public:
    
25.     SHM_SYNC_COND() = default;
    
26.     bool init(const char *shm_name,size_t elm_size,size_t eml_count)
    
27.     {
    
28.         assert(_data == nullptr);
    
29.         assert(_shm_mutex == nullptr);
    
30.         size_t s = sizeof(SHM_MUTEX) + elm_size * eml_count;// 计算内存大小
    
31.         int shm_fd = shm_open(shm_name,O_CREAT | O_RDWR,0666);// 创建/打开共享内存文件
    
32.         exit_on_error(shm_fd < 0,"shm_open failed!");
    
33. 
    
34. 
    
35.         int ret = ftruncate(shm_fd,s); // 截断共享文件大小
    
36.         exit_on_error(ret < 0,"ftruncate failed!");
    
37. 
    
38. 
    
39.         void *addr = mmap(NULL,s,PROT_WRITE | PROT_READ,MAP_SHARED,shm_fd,0);// 将共享内存文件进行内存映射
    
40.         exit_on_error(addr == (void *)-1,"mmap failed");
    
41. 
    
42. 
    
43.         _shm_mutex = (SHM_MUTEX *)addr;// 获取共享内存锁
    
44.         _data = (char *)addr + sizeof(SHM_MUTEX);
    
45. 
    
46. 
    
47.         pthread_mutexattr_t mutexattr;// 设置 mutex 的 PTHREAD_PROCESS_SHARED 属性
    
48.         pthread_mutexattr_setpshared(&mutexattr,PTHREAD_PROCESS_SHARED);
    
49.         pthread_mutex_init(&_shm_mutex->_mutex,&mutexattr);
    
50.         pthread_mutexattr_destroy(&mutexattr);
    
51. 
    
52. 
    
53. 
    
54.         pthread_condattr_t condattr;// 设置 cond 的 PTHREAD_PROCESS_SHARED 属性
    
55.         pthread_condattr_setpshared(&condattr,PTHREAD_PROCESS_SHARED);
    
56.         pthread_cond_init(&_shm_mutex->_cond,&condattr);
    
57.         pthread_condattr_destroy(&condattr);
    
58.         return true;
    
59.     }
    
60. 
    
61. 
    
62.     void notify()// 跨进程进行条件变量通知
    
63.     {
    
64.         assert(_data != nullptr);
    
65.         assert(_shm_mutex != nullptr);
    
66.         pthread_mutex_lock(&_shm_mutex->_mutex);
    
67.         printf("notify..\r\n");
    
68.         pthread_cond_broadcast(&_shm_mutex->_cond);
    
69.         pthread_mutex_unlock(&_shm_mutex->_mutex);
    
70.     }
    
71. 
    
72. 
    
73.     void wait(int32_t wait_ms)// 跨进程进行条件变量等待
    
74.     {
    
75.         assert(_data != nullptr);
    
76.         assert(_shm_mutex != nullptr);
    
77. 
    
78. 
    
79. 
    
80.         struct timeval now;
    
81.         struct timespec abstime;
    
82.         gettimeofday(&now,NULL);
    
83.         printf("wait sec:%ld,nsec:%ld,wait_ms:%d\r\n",now.tv_sec,now.tv_usec*1000,wait_ms);
    
84.         abstime.tv_nsec = (now.tv_usec + wait_ms) * 1000;
    
85.         abstime.tv_sec = now.tv_sec + (abstime.tv_nsec) / 1000000000;
    
86.         abstime.tv_nsec = abstime.tv_nsec % 1000000000;
    
87. 
    
88. 
    
89.         pthread_mutex_lock(&_shm_mutex->_mutex);
    
90.         printf("wait sec:%ld,nsec:%ld\r\n",abstime.tv_sec,abstime.tv_nsec);
    
91.         //pthread_cond_wait(&_shm_mutex->_cond,&_shm_mutex->_mutex);
    
92.         pthread_cond_timedwait(&_shm_mutex->_cond,&_shm_mutex->_mutex,&abstime);
    
93.         pthread_mutex_unlock(&_shm_mutex->_mutex);
    
94.     }
    
95. 
    
96. 
    
97.     void *data_buf() const
    
98.     {
    
99.         assert(_data != nullptr);
    
100.         return _data;
     
101.     }
     
102. 
     
103. 
     
104. private:
     
105.     SHM_MUTEX * _shm_mutex{nullptr};
     
106.     void *_data{nullptr};
     
107. };
     
108. 
     
109. 
     
110. struct IPC_DATA // 跨进程通信时使用的结构体
     
111. {
     
112.     pid_t _pid{0};
     
113.     char  _msg[256];
     
114. };
     
115. 
     
116. 
     
117. 
     
118. int main()
     
119. {
     
120.     const char *shm_name = "shm_cond_name"; // 共享内存文件名
     
121.     SHM_SYNC_COND shm_cond;
     
122.     size_t elm_size = sizeof(IPC_DATA);
     
123.     size_t elm_count = 10;
     
124.     shm_cond.init(shm_name, elm_size,elm_count);// 创建共享锁以及内存映射
     
125.     IPC_DATA *datas = (IPC_DATA *)shm_cond.data_buf();// 获取开辟的内存映射数据地址
     
126. 
     
127. 
     
128.     pid_t pid = fork();// 创建子进程
     
129.     if (pid == -1)
     
130.     {
     
131.         exit_on_error(true,"fork failed");
     
132.     }
     
133. 
     
134. 
     
135.     if (pid == 0) // 子进程，向 IPC 数据写入数据，然后进行 notify
     
136.     {
     
137.         usleep(100);
     
138.         printf("child to write msg\r\n");
     
139.         for (int i = 0; i < elm_count;++i)
     
140.         {
     
141.             datas[i]._pid = pid;
     
142.             snprintf(datas[i]._msg,sizeof(datas[i]._msg),"msg from child,pid:%d,index:%d\r\n",getpid(),i);
     
143.         }
     
144. 
     
145. 
     
146.         shm_cond.notify();
     
147.         printf("child finished!\r\n");
     
148.         usleep(10);
     
149.         return 0;
     
150.     }
     
151.     else // 父进程，等待子进程进行条件变量通知，或者等待10s超时，最后输出 IPC_DATA 内容
     
152.     {
     
153.         int index = 0;
     
154.         printf("parent wait for msg\r\n");
     
155.         shm_cond.wait(1000*1000*10); // 如果没有生产者进行notify，则需要等待10s超时，此处在10s内就已经被唤醒，说明条件变量跨进程生效
     
156.         while (index < elm_count)
     
157.         {
     
158.             printf("wait for index:%d!msg:%s",index,datas[index++]._msg);
     
159.         }
     
160. 
     
161. 
     
162.         printf("parent to exit!\r\n");
     
163.         usleep(100);
     
164.         return 0;
     
165.     }
     
166.     return 0;
     
167. }
     
168. 
     
169. 
     
170. // 输出:
     
171. // parent wait for msg
     
172. // wait sec:1673590541,nsec:138637000,wait_ms:10000000
     
173. // wait sec:1673590551,nsec:138637000
     
174. // child to write msg
     
175. // notify..
     
176. // child finished!
     
177. // wait for index:1!msg:msg from child,pid:7529,index:0
     
178. // wait for index:2!msg:msg from child,pid:7529,index:1
     
179. // wait for index:3!msg:msg from child,pid:7529,index:2
     
180. // wait for index:4!msg:msg from child,pid:7529,index:3
     
181. // wait for index:5!msg:msg from child,pid:7529,index:4
     
182. // wait for index:6!msg:msg from child,pid:7529,index:5
     
183. // wait for index:7!msg:msg from child,pid:7529,index:6
     
184. // wait for index:8!msg:msg from child,pid:7529,index:7
     
185. // wait for index:9!msg:msg from child,pid:7529,index:8
     
186. // wait for index:10!msg:msg from child,pid:7529,index:9
     
187. // parent to exit!

复制代码