c++ - 0MQ : on passing a context to a pool of threads a Segmentation fault was thrown

Question

我正在构建一个小程序 ( here )，其中 main() 向所有工作线程发送相同的 msg 并且工作线程简单地打印消息。

我想我完全遵循了this教程，来自官方指南。

我通过 void 指针将 0MQ context 传递给线程，然后将其转换回 zmq::context_t *。但我仍然遇到了段错误。以下是 GDB 中 coredump 的信息:

Core was generated by `./test 1'.
Program terminated with signal SIGSEGV, Segmentation fault.
#0  zmq::ctx_t::check_tag (this=this@entry=0xabadcafe) at src/ctx.cpp:89
89  src/ctx.cpp: No such file or directory.
[Current thread is 1 (Thread 0x7fcdaae3f700 (LWP 1428))]
(gdb) bt
#0  zmq::ctx_t::check_tag (this=this@entry=0xabadcafe) at src/ctx.cpp:89
#1  0x00007fcdad17c275 in zmq_socket (ctx_=0xabadcafe, type_=7) at src/zmq.cpp:245
#2  0x0000000000402da1 in zmq::socket_t::init (this=0x7fcdaae3edf0, context_=..., type_=7) at /usr/include/zmq.hpp:649
#3  0x0000000000402ac3 in zmq::socket_t::socket_t (this=0x7fcdaae3edf0, context_=..., type_=7) at /usr/include/zmq.hpp:463
#4  0x0000000000401f18 in task1 (arg=0x170ac20) at test.cpp:21
#5  0x00000000004050fe in std::_Bind_simple<void (*(void*))(void*)>::_M_invoke<0ul>(std::_Index_tuple<0ul>) (this=0x170e0d8)
    at /usr/include/c++/5/functional:1531
#6  0x0000000000405008 in std::_Bind_simple<void (*(void*))(void*)>::operator()() (this=0x170e0d8) at /usr/include/c++/5/functional:1520
#7  0x0000000000404f98 in std::thread::_Impl<std::_Bind_simple<void (*(void*))(void*)> >::_M_run() (this=0x170e0c0) at /usr/include/c++/5/thread:115
#8  0x00007fcdacc48c80 in ?? () from /usr/lib/x86_64-linux-gnu/libstdc++.so.6
#9  0x00007fcdacf196ba in start_thread (arg=0x7fcdaae3f700) at pthread_create.c:333
#10 0x00007fcdac6b782d in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:109
(gdb) quit

用 g++ -std=c++11 test.cpp -o test -lzmq -lpthread -g 编译的程序。

Answer 1

在此用例中，无线程 Context() 可以提供帮助:

native API 推荐这种做法用于纯 inproc:// 基于传输类的多线程用法:

The zmq_init() function initialises a ØMQ context.

The io_threads argument specifies the size of the ØMQ thread pool to handle I/O operations. If your application is using only the inproc transport for messaging you may set this to zero, otherwise set it to at least one.

Thread safety

A ØMQ context is thread safe and may be shared among as many application threads as necessary, without any additional locking required on the part of the caller.

This function is deprecated by zmq_ctx_new(3).

最近的 { API/c++ wrapper } 将此步骤(最初在 { v2.1.11 API/c++ wrapper } 中通过 Context( NUM_io_threads ) 在实例化时可用)移动到 { v4. 2.2 } 通过设置调用在实例化后(但套接字关联前)状态拆分过程
zmq_ctx_set( *ctx, ZMQ_IO_THREADS, 0 )/p>

---

int main( int argc, char* argv[] )
{   
    int worker_num;
    if (  argc != 2 ) { cout << "1 parameter pls" << endl;                           exit(1); } // EXIT[1]
    try
    {
      if ( !has_only_digits( string( argv[1] ) ) ) { cout << "digit pls" << endl;    exit(1); } // EXIT[2]
      worker_num =     stoi( string( argv[1] ) );
    }
    catch ( exception e ) { cout << "exception while processing parameters" << endl; exit(1); } // EXIT[3]
 // _______________________________________  // ZeroMQ inproc use-case W/O IO-thread
    zmq::context_t context( 0 );             // 
 // _______________________________________  // 

    zmq::socket_t  distask_socket( context, ZMQ_PUSH );
                   distask_socket.bind( "inproc://task_publisher" );

    vector<thread> pool;

    for ( int i = 0; i < worker_num; i++ )
    {
      cout << "main() : creating thread, " << i << endl;
      pool.push_back( thread(           task1,
                              (void *) &context
                              )
                      );
    }

    for ( int i = 0; i < worker_num; i++ )
    {
      zmq::message_t msg( 6 );
      memcpy( (void *) msg.data(), "World", 6 ); 
      distask_socket.send( msg );
    }

    for ( auto &t : pool )
      t.join();

    exit(0);
}

---

值得压榨最后一点性能并消除延迟源。