Nanomsg 具有多线程应用程序的非阻塞双向套接字

Question

我正在使用 Nanomsg 在我的系统中使用 C++ 进行 IPC。我想创建一个后台线程来处理发送和接收消息。我使用对范式，并使用 nn_poll 检查套接字 fd 是否可写或可读，如果可读则读取；如果可写，则从消息队列中弹出一项并发送。我的问题是我创建的背景线程使用了很多 CPU 使用率，因为 nn_poll 循环中没有睡眠，有没有办法减少 CPU 使用率但仍然使延迟就像没有睡眠一样？下面是我的示例代码。谢谢。

服务器.cpp

#include <iostream>
#include <thread>
#include <string>
#include <queue>
#include <utility>
#include <mutex>
#include <nanomsg/pair.h>
#include <nanomsg/nn.h>

class Nanomsg {
private:
    bool _server;
    bool _stop;

    int _sock;
    std::string _url;
    std::thread _th;
    std::queue<std::string> _queue;

    std::mutex _queueMutex;

    void _start() {
        _sock = nn_socket(AF_SP, NN_PAIR);

        if (_sock < 0) {
            std::cout << "failed to create socket" << std::endl;
            return;
        }

        int rc = 0;

        if (_server) {
            rc = nn_bind(_sock, _url.c_str());
        } else {
            rc = nn_connect(_sock, _url.c_str());
        }

        if (rc < 0) {
            std::cout << "failed to connect/bind socket" << std::endl;
            return;
        }

        struct nn_pollfd pfd{};
        pfd.fd = _sock;
        pfd.events = NN_POLLIN | NN_POLLOUT;

        while (!_stop) {
            std::cout << "ssasd" << std::endl;
            rc = nn_poll(&pfd, 1, 2000);

            if (rc == 0) {
                std::cout << "timeout" << std::endl;
                continue;
            }

            if (rc == -1) {
                std::cout << "error!" << std::endl;
                return;
            }

            if (pfd.revents & NN_POLLIN) {
                char *buf = nullptr;
                int rbs = nn_recv(_sock, &buf, NN_MSG, 0);

                if (rbs < 0) {
                    continue;
                }

                std::string r(buf, rbs);

                std::cout << "received [" << r << "]" << std::endl;

                nn_freemsg(buf);
            }

            if (pfd.revents & NN_POLLOUT) {
                std::cout << "asd" << std::endl;
                if (_queue.empty()) {
                    continue;
                }

                {
                    std::lock_guard<std::mutex> lock(_queueMutex);
                    auto msg = _queue.front();

                    std::cout << "send [" << msg << "]" << std::endl;

                    rc = nn_send(_sock, msg.c_str(), msg.length(), 0);
                    if (rc >= 0) {
                        _queue.pop();
                    }
                }
            }
        }

    }

public:
    Nanomsg() : _sock(0), _server(false), _stop(false), _url("ipc:///tmp/test.ipc") {

    }

    Nanomsg(std::string url, bool server) : _url(std::move(url)), _sock(0), _server(server), _stop(false) {

    }

    void start() {
        _th = std::thread([=]() {
            _start();
        });
    }

    void stop() {
        _stop = true;

        if (_th.joinable()) {
            _th.join();
        }
    }

    void send(const std::string& msg) {
        {
            std::lock_guard<std::mutex> lock(_queueMutex);
            _queue.push(msg);
        }
    }

};

int main() {

    Nanomsg server("ipc:///tmp/test.ipc", true);

    server.start();

    while (true) {
        server.send("test");
        std::this_thread::sleep_for(std::chrono::seconds(3));
    }

    return 0;
}

客户端.cpp

#include <iostream>
#include <thread>
#include <string>
#include <queue>
#include <utility>
#include <mutex>
#include <nanomsg/pair.h>
#include <nanomsg/nn.h>

struct nn_pollf {
    int fd;
    short events;
    short revents;
};

class Nanomsg {
private:
    bool _server;
    bool _stop;

    int _sock;
    std::string _url;
    std::thread _th;
    std::queue<std::string> _queue;

    std::mutex _queueMutex;

    void _start() {
        _sock = nn_socket(AF_SP, NN_PAIR);

        if (_sock < 0) {
            std::cout << "failed to create socket" << std::endl;
            return;
        }

        int rc = 0;

        if (_server) {
            rc = nn_bind(_sock, _url.c_str());
        } else {
            rc = nn_connect(_sock, _url.c_str());
        }

        if (rc < 0) {
            std::cout << "failed to connect/bind socket" << std::endl;
            return;
        }

        struct nn_pollfd pfd{};
        pfd.fd = _sock;
        pfd.events = NN_POLLIN | NN_POLLOUT;

        while (!_stop) {
            std::cout << "ssasd" << std::endl;
            rc = nn_poll(&pfd, 1, 2000);

            if (rc == 0) {
                std::cout << "timeout" << std::endl;
                continue;
            }

            if (rc == -1) {
                std::cout << "error!" << std::endl;
                return;
            }

            if (pfd.revents & NN_POLLIN) {
                char *buf = nullptr;
                int rbs = nn_recv(_sock, &buf, NN_MSG, 0);

                if (rbs < 0) {
                    continue;
                }

                std::string r(buf, rbs);

                std::cout << "received [" << r << "]" << std::endl;

                nn_freemsg(buf);
            }

            if (pfd.revents & NN_POLLOUT) {
                std::cout << "asd" << std::endl;
                if (_queue.empty()) {
                    continue;
                }

                {
                    std::lock_guard<std::mutex> lock(_queueMutex);
                    auto msg = _queue.front();

                    std::cout << "send [" << msg << "]" << std::endl;

                    rc = nn_send(_sock, msg.c_str(), msg.length(), 0);
                    if (rc >= 0) {
                        _queue.pop();
                    }
                }
            }
        }

    }

public:
    Nanomsg() : _sock(0), _server(false), _stop(false), _url("ipc:///tmp/test.ipc") {

    }

    Nanomsg(std::string url, bool server) : _url(std::move(url)), _sock(0), _server(server), _stop(false) {

    }

    void start() {
        _start();
//        _th = std::thread([=]() {
//            _start();
//        });
    }

    void stop() {
        _stop = true;

        if (_th.joinable()) {
            _th.join();
        }
    }

    void send(const std::string& msg) {
        {
            std::lock_guard<std::mutex> lock(_queueMutex);
            _queue.push(msg);
        }
    }

};

int main() {

    Nanomsg client("ipc:///tmp/test.ipc", false);

    client.start();

    return 0;
}

Answer 1

如果没有什么可以发送也没有什么可以接收，让你的线程休眠 一毫秒。几乎是您当前设计中唯一可以做的事情。

如果可能，您可以使用nanomsg next generation (nng) 并给它asynchronous interface 一个机会。似乎您自己无论如何都在实现一个异步接口，那么为什么不也使用 nanomsg 呢？它们具有您操作系统的网络 API 的所有可用功能，因此应该能够提供最佳延迟而不会浪费 CPU 时间。

创建一个异步 I/O 句柄并使用 nng_aio_alloc(3) 设置回调。致电nng_recv_aio(3) 以获取收到数据的通知。不要管理自己的 发送队列，在void Nanomsg::send() 中使用nng_send_aio(3)。

不幸的是，nng 是一个单独的库，而您使用的是经典的 nanomsg。我注意到只有写到中间..