c++ - 如何避免与 `asio::ip::tcp::iostream` 的数据竞争？

标签 c++ multithreading tcp c++14 boost-asio

我的问题

当使用两个线程通过 asio::ip::tcp::iostream 发送和接收数据时，如何避免数据竞争？

设计

我正在编写一个使用 asio::ip::tcp::iostream 进行输入和输出的程序。该程序通过端口 5555 接受来自(远程)用户的命令，并通过相同的 TCP 连接向用户发送消息。因为这些事件(从用户收到的命令或发送给用户的消息)异步发生，所以我有单独的传输和接收线程。

在这个玩具版本中，命令是“一”、“二”和“退出”。当然“退出”退出程序。其他命令什么都不做，任何无法识别的命令都会导致服务器关闭 TCP 连接。

传输的消息是简单的序列号消息，每秒发送一次。

在这个玩具版本和我尝试编写的真实代码中，传输和接收进程都使用阻塞 IO，因此似乎没有使用 std::的好方法mutex 或其他同步机制。 (在我的尝试中，一个进程会获取互斥锁然后阻塞，这不会为此工作。)

构建和测试

为了构建和测试它，我在 64 位 Linux 机器上使用 gcc 7.2.1 版和 valgrind 3.13。构建:

g++ -DASIO_STANDALONE -Wall -Wextra -pedantic -std=c++14 concurrent.cpp -o concurrent -lpthread

为了测试，我使用以下命令运行服务器:

valgrind --tool=helgrind --log-file=helgrind.txt ./concurrent

然后我在另一个窗口中使用 telnet 127.0.0.1 5555 创建到服务器的连接。 helgrind 正确指出存在数据竞争，因为 runTx 和 runRx 都试图异步访问同一个流:

==16188== Possible data race during read of size 1 at 0x1FFEFFF1CC by thread #1

==16188== Locks held: none

... many more lines elided

并发.cpp

#include <asio.hpp>
#include <iostream>
#include <fstream>
#include <thread>
#include <array>
#include <chrono>

class Console {
public:
    Console() :
        want_quit{false},
        want_reset{false}
    {}
    bool getQuitValue() const { return want_quit; }
    int run(std::istream *in, std::ostream *out);
    bool wantReset() const { return want_reset; }
private:
    int runTx(std::istream *in);
    int runRx(std::ostream *out);
    bool want_quit;
    bool want_reset;
};

int Console::runTx(std::istream *in) {
    static const std::array<std::string, 3> cmds{
        "quit", "one", "two", 
    };
    std::string command;
    while (!want_quit && !want_reset && *in >> command) {
        if (command == cmds.front()) {
            want_quit = true;
        }
        if (std::find(cmds.cbegin(), cmds.cend(), command) == cmds.cend()) {
            want_reset = true;
            std::cout << "unknown command [" << command << "]\n";
        } else {
            std::cout << command << '\n';
        }
    }
    return 0;
}

int Console::runRx(std::ostream *out) {
    for (int i=0; !(want_reset || want_quit); ++i) {
        (*out) << "This is message number " << i << '\n';
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
        out->flush();
    }
    return 0;
}

int Console::run(std::istream *in, std::ostream *out) {
    want_reset = false;
    std::thread t1{&Console::runRx, this, out};
    int status = runTx(in);
    t1.join();
    return status;
}

int main()
{
    Console con;
    asio::io_service ios;
    // IPv4 address, port 5555
    asio::ip::tcp::acceptor acceptor(ios, 
            asio::ip::tcp::endpoint{asio::ip::tcp::v4(), 5555});
    while (!con.getQuitValue()) {
        asio::ip::tcp::iostream stream;
        acceptor.accept(*stream.rdbuf());
        con.run(&stream, &stream);
        if (con.wantReset()) {
            std::cout << "resetting\n";
        }
    }
}

最佳答案

是的，您正在共享作为流基础的套接字，没有同步

Sidenote, same with the boolean flags, which can easily be "fixed" by changing:

std::atomic_bool want_quit;
std::atomic_bool want_reset;

如何解决

说实话，我觉得没有什么好的解决办法。您自己说过:操作是异步的，因此如果您尝试同步执行它们，就会遇到麻烦。

你可以试着想想黑客。如果我们基于相同的底层套接字(文件描述符)创建一个单独的流对象会怎样。这不会非常容易，因为这样的流不是 Asio 的一部分。

但我们可以使用 Boost Iostreams 破解一个:

#define BOOST_IOSTREAMS_USE_DEPRECATED
#include <boost/iostreams/device/file_descriptor.hpp>
#include <boost/iostreams/stream.hpp>

// .... later:

    // HACK: procure a _separate `ostream` to prevent the race, using the same fd
    namespace bio = boost::iostreams;
    bio::file_descriptor_sink fds(stream.rdbuf()->native_handle(), false); // close_on_exit flag is deprecated
    bio::stream<bio::file_descriptor_sink> hack_ostream(fds);

    con.run(stream, hack_ostream);

事实上，这在没有竞争的情况下运行(在同一个套接字上同时读取和写入 are fine，只要您不共享包装它们的非线程安全 Asio 对象)。

我推荐的是:

不要那样做。这是一个kludge。您使事情复杂化，显然是为了避免使用异步代码。我会咬紧牙关。

将 IO 机制从服务逻辑中分离出来并不太费力。您最终将摆脱随机限制(您可以考虑与多个客户端打交道，您可以完全不使用任何线程，等等)。

如果您想了解一些中间立场，请查看堆栈协程 (http://www.boost.org/doc/libs/1_66_0/doc/html/boost_asio/reference/spawn.html)

list

仅供引用

Note I refactored to remove the need for pointers. You're not transferring ownership, so a reference will do. In case you didn't know how to pass the reference to a bind/std::thread constructor, the trick is in the std::ref you'll see.

[For stress testing I have greatly reduced the delays.]

Live On Coliru 

#include <boost/asio.hpp>
#include <iostream>
#include <fstream>
#include <thread>
#include <array>
#include <chrono>

class Console {
public:
    Console() :
        want_quit{false},
        want_reset{false}
    {}
    bool getQuitValue() const { return want_quit; }
    int run(std::istream &in, std::ostream &out);
    bool wantReset() const { return want_reset; }
private:
    int runTx(std::istream &in);
    int runRx(std::ostream &out);
    std::atomic_bool want_quit;
    std::atomic_bool want_reset;
};

int Console::runTx(std::istream &in) {
    static const std::array<std::string, 3> cmds{
        {"quit", "one", "two"}, 
    };
    std::string command;
    while (!want_quit && !want_reset && in >> command) {
        if (command == cmds.front()) {
            want_quit = true;
        }
        if (std::find(cmds.cbegin(), cmds.cend(), command) == cmds.cend()) {
            want_reset = true;
            std::cout << "unknown command [" << command << "]\n";
        } else {
            std::cout << command << '\n';
        }
    }
    return 0;
}

int Console::runRx(std::ostream &out) {
    for (int i=0; !(want_reset || want_quit); ++i) {
        out << "This is message number " << i << '\n';
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
        out.flush();
    }
    return 0;
}

int Console::run(std::istream &in, std::ostream &out) {
    want_reset = false;
    std::thread t1{&Console::runRx, this, std::ref(out)};
    int status = runTx(in);
    t1.join();
    return status;
}

#define BOOST_IOSTREAMS_USE_DEPRECATED
#include <boost/iostreams/device/file_descriptor.hpp>
#include <boost/iostreams/stream.hpp>

int main()
{
    Console con;
    boost::asio::io_service ios;

    // IPv4 address, port 5555
    boost::asio::ip::tcp::acceptor acceptor(ios, boost::asio::ip::tcp::endpoint{boost::asio::ip::tcp::v4(), 5555});

    while (!con.getQuitValue()) {
        boost::asio::ip::tcp::iostream stream;
        acceptor.accept(*stream.rdbuf());

        {
            // HACK: procure a _separate `ostream` to prevent the race, using the same fd
            namespace bio = boost::iostreams;
            bio::file_descriptor_sink fds(stream.rdbuf()->native_handle(), false); // close_on_exit flag is deprecated
            bio::stream<bio::file_descriptor_sink> hack_ostream(fds);

            con.run(stream, hack_ostream);
        }

        if (con.wantReset()) {
            std::cout << "resetting\n";
        }
    }
}

测试:

netcat localhost 5555 <<<quit
This is message number 0
This is message number 1
This is message number 2

和

commands=( one two one two one two one two one two one two one two three )
while sleep 0.1; do echo ${commands[$(($RANDOM%${#commands}))]}; done | (while netcat localhost 5555; do sleep 1; done)

无限期地运行，偶尔会重置连接(当发送命令“三”时)。

关于c++ - 如何避免与 `asio::ip::tcp::iostream` 的数据竞争？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/48117493/

上一篇：java - 在同一端口上监听 TCP 和 UDP 请求

下一篇：sockets - 关于 UDP 服务器如何将响应发送回 UDP 客户端的困惑

相关文章：

c++ - 如何终止线程池中的所有预分配线程？

java - 同步(this)和观察者通知

c++ - 尝试连接到 tcp 套接字时连接被拒绝 (linux)

c++ - 将 boost IOStreams 与 std::ostream_iterator 结合使用

c++ - 一个应用程序可以运行另一个应用程序的代码吗？

c++ - 在具有容量/调整大小的类中初始化 vector

python - 如何从 Python 中的运行中调用 Thread 派生类方法？

.net - 在什么情况下我应该使用不同的 .NET 线程方法？

c - AT 命令 ESP8266 01 : AT+CIPSTART: How to fix response Link type Error/Can't connect with TCP

c# - TCP Socket Server - 一种检测传入数据的方法

©2024 IT工具网  联系我们