c++ - 在多线程 c++0x11 程序中结合 cntr +x 或 kill 信号使用 zmq::poll

Question

对于自定义服务器，我打算使用 int zmq::poll( zmq_pollitemt_t * items, int nitems, long timeout = -1)。函数，我认为它是 unix poll 函数的包装器，但在文件描述符旁边包含 zmq::socket_t。在我按 ctrl+x 或在终端中运行 $kill my_server_pid 之前，该函数按预期工作。我希望轮询以 -1 终止或抛出一个 zmq::error_t(派生自 std::exception)，其中包括一个 errno 和 strerr 消息。这应该表明存在中断。然后我的服务器应该优雅地处理信号并保存一些数据并关闭。

下面我有一段代码可以演示这个问题。首先，我展示了一些我的环境以及我如何编译它:

    mja@gijs:~/Desktop/sample_programs/zeromq/poll$ g++ -v
    Using built-in specs.
    COLLECT_GCC=g++
    COLLECT_LTO_WRAPPER=/usr/lib/gcc/x86_64-linux-gnu/4.6/lto-wrapper
    Target: x86_64-linux-gnu
    Configured with: ../src/configure -v --with-pkgversion='Ubuntu/Linaro      4.6.3-1ubuntu5' --with-bugurl=file:///usr/share/doc/gcc-4.6/README.Bugs --enable-languages=c,c++,fortran,objc,obj-c++ --prefix=/usr --program-suffix=-4.6 --enable-shared --enable-linker-build-id --with-system-zlib --libexecdir=/usr/lib --without-included-gettext --enable-threads=posix --with-gxx-include-dir=/usr/include/c++/4.6 --libdir=/usr/lib --enable-nls --with-sysroot=/ --enable-clocale=gnu --enable-libstdcxx-debug --enable-libstdcxx-time=yes --enable-gnu-unique-object --enable-plugin --enable-objc-gc --disable-werror --with-arch-32=i686 --with-tune=generic --enable-checking=release --build=x86_64-linux-gnu --host=x86_64-linux-gnu --target=x86_64-linux-gnu
    Thread model: posix
    gcc version 4.6.3 (Ubuntu/Linaro 4.6.3-1ubuntu5) 
    mja@gijs:~/Desktop/sample_programs/zeromq/poll$ pkg-config --modversion `enter code here`libzmq
    2.2.0
    mja@gijs:~/Desktop/sample_programs/zeromq/poll$ g++ -pthread -std=c++0x -Wall -Wextra -pedantic -o poll polling.cpp $(pkg-config --cflags --libs libzmq )
    mja@gijs:~/Desktop/sample_programs/zeromq/poll$

现在是 polling.cpp 的代码:

#include <zmq.hpp>
#include <thread>
#include <cstdlib>
#include <string>
#include <iostream>
#include <signal.h>

const char* bind_addres = "tcp://*:2345";
const char* connect_addres = "tcp://localhost:2345";

inline void
send_str( zmq::socket_t& sock, const std::string& s) throw (zmq::error_t) {
    zmq::message_t msg ( s.size() );
    memcpy( msg.data(), s.c_str(), msg.size() );
    sock.send( msg );
}

inline void recv_str( zmq::socket_t& sock, std::string& s) throw( zmq::error_t) {
    zmq::message_t msg;
    sock.recv(&msg);
    s = std::string( static_cast<const char*>(msg.data()), msg.size());
}
static int interrupted = 0;

static void handle_signal ( int signal ) 
{
    interrupted = signal;
    std::cerr << "Interrupted by signal: " << signal << std::endl;
}

void catch_signals (void) 
{
    struct sigaction action;
    action.sa_handler = handle_signal;
    action.sa_flags = 0;
    sigemptyset(&action.sa_mask);

    sigaction(SIGINT,  &action, NULL);
    sigaction(SIGTERM, &action, NULL);
}

int get_interrupted(void)
{
    std::cout << "interrupted = " << interrupted << std::endl;
    return interrupted;
}

void req_thread ( zmq::context_t* c ){


    zmq::socket_t sock( *c, ZMQ_REQ);
    sock.connect(connect_addres);

    zmq::pollitem_t items[]{
        { sock, 0, ZMQ_POLLIN, 0}
    };

    while (true){
        try {
            // zmq 3.x.x takes ms instead of us so change to eg 1000 or be patient.
            int rc = zmq::poll(items, 1, 1000000);
            if (rc > 0){
                if ( items[0].revents & ZMQ_POLLIN){
                    std::string s;
                    recv_str(sock, s);
                    std::cout << s << std::endl;
                }
            }
            else if ( rc == 0){ //timeout
                send_str( sock, "Hello");
            }
            else{
                std::cout << __func__ << " " << __LINE__ << get_interrupted() << std::endl;
            }
        }
        catch( zmq::error_t& e ){
            std::cout << __func__ << " " << __LINE__ << e.what() << std::endl;
        }
    }
}

void rep_thread ( zmq::context_t* c ){

    zmq::socket_t sock( *c, ZMQ_REP);
    sock.bind(bind_addres);

    zmq::pollitem_t items[]{
        { sock, 0, ZMQ_POLLIN, 0}
    };

    while (true){
        try{
            int rc = zmq::poll(items, 1 , -1);
            if (rc > 0){
                if ( items[0].revents & ZMQ_POLLIN){
                    std::string s;
                    recv_str(sock, s);
                    s+=" world!";
                    send_str(sock, s);
                }
            }
            else{
                std::cout << __func__ << " " << __LINE__ << get_interrupted() << std::endl;
            }
        }
        catch( zmq::error_t& e ){
            std::cout << __func__ << " " << __LINE__ << e.what() << std::endl;
        }
    }
}

int main(){

    zmq::context_t context(1);
    catch_signals();

    std::thread t1 ( rep_thread, &context);
    std::thread t2 ( req_thread, &context);

    t1.join();
    t2.join();

    return 0;
}

最后，我展示了一些示例输出，展示了我的问题，即 zmq_poll 似乎不受在终端中按 ctrl+c 的影响:

    mja@gijs:~/Desktop/sample_programs/zeromq/poll$ ./poll 
    Hello world!
    Hello world!
    Hello world!
    ^CInterrupted by signal: 2
    Hello world!
    Hello world!
    ^Z
    [1]+  Stopped                 ./poll
    mja@gijs:~/Desktop/sample_programs/zeromq/poll$ kill -9 %1

    [1]+  Stopped                 ./poll
    mja@gijs:~/Desktop/sample_programs/zeromq/poll$ 

因此，从终端的输出可以看出，没有抛出 zmq::error_t，zmq::poll() 也不返回 -1；

在下一个示例 simple_poll.cpp 中它应该如何工作，可以看到抛出了一个 zmq::error_t:

#include<zmq.hpp>
#include<signal.h>
#include<iostream>


static int interrupted = 0;

static void handle_signal ( int signal ) 
{
    interrupted = signal;
    /*just to show that the signal handler works*/
    std::cerr << "Interrupted by signal: " << signal << std::endl;
}

void catch_signals (void) 
{
    struct sigaction action;
    action.sa_handler = handle_signal;
    action.sa_flags = 0;
    sigemptyset(&action.sa_mask);

    sigaction(SIGINT,  &action, NULL);
    sigaction(SIGTERM, &action, NULL);
}

using namespace std;

int main(){

    zmq::context_t context(1);
    catch_signals();
    zmq::socket_t sock( context, ZMQ_REP );
    /*listen on port 2346 on all available interfaces.*/
    sock.bind("tcp://*:2346");

    zmq::pollitem_t items[] = {
        {sock, 0 , ZMQ_POLLIN, 0}
    };

    try {
        /*wait for a event*/
        zmq::poll( items, 1, -1);
        /*for zmq users read message and respond*/
    }
    catch (zmq::error_t& e){
        cout << "error occured: " <<e.what() << endl;
        cout << "We were interrupted by: " << interrupted << endl;
    }
    return 0;
}

这会在终端中按 ctrl+x 产生以下结果，表明 zmq::error_t 已被捕获并且信号已被处理。

mja@gijs:~/Desktop/sample_programs/zeromq/poll$ ./simple_poll ^CInterrupted by signal: 2
error occured: Interrupted system call
We were interrupted by: 2
mja@gijs:~/Desktop/sample_programs/zeromq/poll$

Answer 1

您有一个信号处理程序 - 但您没有在其中执行任何操作。在您的处理程序中，中断轮询循环(而不是 true，检查您在信号处理程序中设置的某些条件。)

为了争论起见，假设您正在使用 c++11，请尝试类似...

// Global which indicates that we are running..
std::atomic<bool> running = true;


// In your handler - reset this flag
static void handle_signal ( int signal ) {
  running = false;
}

现在你的循环变成:

while (running)
{
:
}