使用匿名管道是否为线程间通信引入了内存屏障
Does the use of an anonymous pipe introduce a memory barrier for interthread communication?
例如，假设我分配了一个带有new的结构，并将指针写入匿名管道的写入端。
如果我从相应的读取端读取指针，我能保证在结构体上看到"正确"的内容吗？
同样令人感兴趣的是，Unix上的socketpair（）和Windows上的TCP环回自连接的结果是否具有相同的保证。
上下文是一种服务器设计，它通过 select/epoll 集中事件调度
例如，假设我使用 new 分配一个结构，并将指针写入匿名管道的写入端。
如果我从相应的读取端读取指针，我能保证在结构体上看到"正确"的内容吗？
不。无法保证写入 CPU 已将写操作从其缓存中清除，并使其对可能执行读取的其他 CPU 可见。
同样令人感兴趣的是，Unix上的socketpair（）和Windows上的TCP环回自连接的结果是否具有相同的保证。
不。
在实践中，调用 write() （这是一个系统调用）最终将锁定内核中的一个或多个数据结构，这应该解决重新排序问题。例如，POSIX 需要后续读取才能查看调用前写入的数据，这意味着锁定（或某种获取/释放）本身。
至于这是否是调用的正式规范的一部分，可能不是。
指针只是一个内存地址，因此，如果您在同一进程中，指针将在接收线程上有效，并将指向相同的结构。如果你在不同的进程中，充其量你会立即得到一个内存错误，在更糟糕的情况下，你会读取（或写入）一个随机内存，这本质上是未定义的行为。
你会阅读正确的内容吗？不比指针位于两个线程共享的静态变量中更好也不更差：如果你想要一致性，你仍然需要做一些同步。
静态内存（由线程共享）、匿名管道、套接字对、tcp 环回等之间的传输地址类型是否重要？否：所有这些通道都传输字节，因此如果您传递内存地址，您将获得内存地址。剩下的就是同步，因为在这里你只是共享一个内存地址。
如果您不使用任何其他同步，则可能发生任何事情（我已经说过未定义的行为吗？
- 读取线程可以在写入内存之前访问内存，方法是写入一个提供过时数据
- 如果您忘记将结构成员声明为可变性，则读取线程可以继续使用缓存的值，此处再次获取过时数据 读取
- 线程可以读取部分写入的数据，这意味着不连贯的数据
一个有趣的问题，到目前为止，Cornstalks只有一个正确答案。
在同一个（多线程）进程中，没有保证，因为指针和数据遵循不同的路径到达目的地。隐式获取/释放保证不适用，因为结构数据不能通过缓存搭载在指针上，并且您正在正式处理数据争用。
但是，查看指针和结构数据本身如何到达第二个线程（分别通过管道和内存缓存），此机制很有可能不会造成任何伤害。将指针发送到对等线程需要 3 次系统调用（write() 次在发送线程中，select() 次，read() 次在接收线程中），这（相对）昂贵，并且当指针值可用时在接收线程中，结构数据可能很久以前就已经到达了。
请注意，这只是一个观察，机制仍然不正确。
，您的情况可能会简化为以下 2 线程模型：
int data = 0;
std::atomic<int*> atomicPtr{nullptr};
//...
void thread1()
{
data = 42;
atomicPtr.store(&integer, std::memory_order_release);
}
void thread2()
{
int* ptr = nullptr;
while(!ptr)
ptr = atomicPtr.load(std::memory_order_consume);
assert(*ptr == 42);
}
由于您有 2 个进程，因此您不能在它们之间使用一个原子变量，但由于您列出了窗口，因此您可以从消费部分省略atomicPtr.load(std::memory_order_consume)，因为 AFAIK，Windows 运行的所有体系结构都保证此负载是正确的加载端没有任何障碍。事实上，我认为没有多少架构可以说明该指令不是 NO-OP（我只听说过 DEC Alpha）
我同意Serge Ballesta的回答。在同一过程中，通过匿名管道发送和接收对象地址是可行的。
由于当消息大小低于 PIPE_BUF（通常为 4096 字节）时，write系统调用保证是原子的，因此多生产者线程不会弄乱彼此的对象地址（64 位应用程序为 8 字节）。
谈话很便宜，这里是 Linux 的演示代码（为简单起见，省略了防御性代码和错误处理程序）。只需复制并粘贴到pipe_ipc_demo.cc然后编译并运行测试。
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <string>
#include <list>
template<class T> class MPSCQ { // pipe based Multi Producer Single Consumer Queue
public:
MPSCQ();
~MPSCQ();
int producerPush(const T* t);
T* consumerPoll(double timeout = 1.0);
private:
void _consumeFd();
int _selectFdConsumer(double timeout);
T* _popFront();
private:
int _fdProducer;
int _fdConsumer;
char* _consumerBuf;
std::string* _partial;
std::list<T*>* _list;
static const int _PTR_SIZE;
static const int _CONSUMER_BUF_SIZE;
};
template<class T> const int MPSCQ<T>::_PTR_SIZE = sizeof(void*);
template<class T> const int MPSCQ<T>::_CONSUMER_BUF_SIZE = 1024;
template<class T> MPSCQ<T>::MPSCQ() :
_fdProducer(-1),
_fdConsumer(-1) {
_consumerBuf = new char[_CONSUMER_BUF_SIZE];
_partial = new std::string; // for holding partial pointer address
_list = new std::list<T*>; // unconsumed T* cache
int fd_[2];
int r = pipe(fd_);
_fdConsumer = fd_[0];
_fdProducer = fd_[1];
}
template<class T> MPSCQ<T>::~MPSCQ() { /* omitted */ }
template<class T> int MPSCQ<T>::producerPush(const T* t) {
return t == NULL ? 0 : write(_fdProducer, &t, _PTR_SIZE);
}
template<class T> T* MPSCQ<T>::consumerPoll(double timeout) {
T* t = _popFront();
if (t != NULL) {
return t;
}
if (_selectFdConsumer(timeout) <= 0) { // timeout or error
return NULL;
}
_consumeFd();
return _popFront();
}
template<class T> void MPSCQ<T>::_consumeFd() {
memcpy(_consumerBuf, _partial->data(), _partial->length());
ssize_t r = read(_fdConsumer, _consumerBuf, _CONSUMER_BUF_SIZE - _partial->length());
if (r <= 0) { // EOF or error, error handler omitted
return;
}
const char* p = _consumerBuf;
int remaining_len_ = _partial->length() + r;
T* t;
while (remaining_len_ >= _PTR_SIZE) {
memcpy(&t, p, _PTR_SIZE);
_list->push_back(t);
remaining_len_ -= _PTR_SIZE;
p += _PTR_SIZE;
}
*_partial = std::string(p, remaining_len_);
}
template<class T> int MPSCQ<T>::_selectFdConsumer(double timeout) {
int r;
int nfds_ = _fdConsumer + 1;
fd_set readfds_;
struct timeval timeout_;
int64_t usec_ = timeout * 1000000.0;
while (true) {
timeout_.tv_sec = usec_ / 1000000;
timeout_.tv_usec = usec_ % 1000000;
FD_ZERO(&readfds_);
FD_SET(_fdConsumer, &readfds_);
r = select(nfds_, &readfds_, NULL, NULL, &timeout_);
if (r < 0 && errno == EINTR) {
continue;
}
return r;
}
}
template<class T> T* MPSCQ<T>::_popFront() {
if (!_list->empty()) {
T* t = _list->front();
_list->pop_front();
return t;
} else {
return NULL;
}
}
// = = = = = test code below = = = = =
#define _LOOP_CNT 5000000
#define _ONE_MILLION 1000000
#define _PRODUCER_THREAD_NUM 2
struct TestMsg { // all public
int _threadId;
int _msgId;
int64_t _val;
TestMsg(int thread_id, int msg_id, int64_t val) :
_threadId(thread_id),
_msgId(msg_id),
_val(val) { };
};
static MPSCQ<TestMsg> _QUEUE;
static int64_t _SUM = 0;
void* functor_producer(void* arg) {
int my_thr_id_ = pthread_self();
TestMsg* msg_;
for (int i = 0; i <= _LOOP_CNT; ++ i) {
if (i == _LOOP_CNT) {
msg_ = new TestMsg(my_thr_id_, i, -1);
} else {
msg_ = new TestMsg(my_thr_id_, i, i + 1);
}
_QUEUE.producerPush(msg_);
}
return NULL;
}
void* functor_consumer(void* arg) {
int msg_cnt_ = 0;
int stop_cnt_ = 0;
TestMsg* msg_;
while (true) {
if ((msg_ = _QUEUE.consumerPoll()) == NULL) {
continue;
}
int64_t val_ = msg_->_val;
delete msg_;
if (val_ <= 0) {
if ((++ stop_cnt_) >= _PRODUCER_THREAD_NUM) {
printf("All done, _SUM=%ldn", _SUM);
break;
}
} else {
_SUM += val_;
if ((++ msg_cnt_) % _ONE_MILLION == 0) {
printf("msg_cnt_=%d, _SUM=%ldn", msg_cnt_, _SUM);
}
}
}
return NULL;
}
int main(int argc, char* const* argv) {
pthread_t consumer_;
pthread_create(&consumer_, NULL, functor_consumer, NULL);
pthread_t producers_[_PRODUCER_THREAD_NUM];
for (int i = 0; i < _PRODUCER_THREAD_NUM; ++ i) {
pthread_create(&producers_[i], NULL, functor_producer, NULL);
}
for (int i = 0; i < _PRODUCER_THREAD_NUM; ++ i) {
pthread_join(producers_[i], NULL);
}
pthread_join(consumer_, NULL);
return 0;
}
这是测试结果（2 * sum(1..5000000) == (1 + 5000000) * 5000000 == 25000005000000）：
$ g++ -o pipe_ipc_demo pipe_ipc_demo.cc -lpthread
$ ./pipe_ipc_demo ## output may vary except for the final _SUM
msg_cnt_=1000000, _SUM=251244261289
msg_cnt_=2000000, _SUM=1000708879236
msg_cnt_=3000000, _SUM=2250159002500
msg_cnt_=4000000, _SUM=4000785160225
msg_cnt_=5000000, _SUM=6251640644676
msg_cnt_=6000000, _SUM=9003167062500
msg_cnt_=7000000, _SUM=12252615629881
msg_cnt_=8000000, _SUM=16002380952516
msg_cnt_=9000000, _SUM=20252025092401
msg_cnt_=10000000, _SUM=25000005000000
All done, _SUM=25000005000000
此处显示的技术用于我们的生产应用。一种典型的用法是使用者线程充当日志编写器，工作线程几乎可以异步写入日志消息。是的，几乎意味着有时当管道已满时，写入器线程可能会在write()中被阻塞，这是操作系统提供的可靠拥塞控制功能。
- 将字符串存储在c++中的稳定内存中
- C++ 指针的内存地址和指向数组的内存地址如何相同？
- Win32编译器选项和内存分配
- 当vector是tje全局变量时，c++中vector的内存管理
- 带内存和隔离功能的SQLite
- 是否可以通过C++扩展强制多个python进程共享同一内存
- 迭代时从向量和内存中删除对象
- 在C++中打印指向不同基元数据类型的指针的内存地址
- 这个指针和内存代码打印是什么？我不知道是打印垃圾还是如何打印我需要的值
- 多个文件的内存分配错误"在抛出 'std :: bad_alloc' what ()： std ：： bad_alloc 的实例后终止调用" [C++]
- 为什么示例代码访问IUnknown中已删除的内存
- 进程间通信堆内存
- 使用用户模式和内核之间共享内存的慢速通信
- VC++和Qt应用程序之间使用共享内存进行通信
- 进程间通信 -- 在共享内存中锁定互斥体
- 使用匿名管道是否为线程间通信引入了内存屏障
- Boost::asio，共享内存和进程间通信
- 使用通信器进行远程内存访问
- 共享内存进程间通信
- Java -> C++ 使用 SWIG over JNI 的通信。防止C++错误和内存泄漏
