Boost::Asio:async_read 问题

发布于 2024-11-14 05:36:37 字数 6827 浏览 1 评论 0原文

我正在编写一个简单的 ProxyServer 来分析包并将它们发送到另一个服务器实例,例如这样:

client ->我的代理-> SQLServer ->
client <- MyProxy <- SQLServer <-

它应该在无限循环中运行。 我现在的问题是代理似乎丢失了包,有时甚至挂起。 当我添加大量调试信息(写入控制台)时,ProxyServer 更加稳定。看来 ProxyServer 太快了.. :-)

我很确定我做错了什么,这是我的会话类的代码(该代码源自 Boost::Asio 示例)。

#include "session.h"

#include <iostream>
using namespace std;

session::session(boost::asio::io_service& io_service)
: socket_(io_service)
, sqlsocket_(io_service)
, io_service_(io_service)
, resolver(io_service)
{
    cout << "session::session()" << endl;
}

session::~session()
{
    cout << "session::~session()" << endl;
    cout << "closing session ..." << endl;
}

tcp::socket& session::socket()
{
    return socket_;
}

void session::start()
{
    cout << "session::start()" << endl;
    cout << "starting session ..." << endl;

    // connect to the sqlserver database
    tcp::resolver::query query("192.168.1.50", "1317");
    tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
    tcp::endpoint endpoint = *endpoint_iterator;

    sqlsocket_.async_connect(endpoint,
        boost::bind(&session::handle_sqlserver_connect, this,
        boost::asio::placeholders::error, ++endpoint_iterator));

    // TODO: connect to the connector
}

void session::handle_read(const boost::system::error_code& error,
                          size_t bytes_transferred)
{
    cout << "session::handle_read()" << endl;
    if (!error)
    {
        cout << "session::handle_read() (read: " 
             << bytes_transferred << ")"
             << endl;
        boost::asio::async_write(sqlsocket_,
            boost::asio::buffer(data_, bytes_transferred),
            boost::bind(&session::handle_sqlserver_write, this,
            boost::asio::placeholders::error, bytes_transferred));
    }
    else
    {
        delete this;
    }
}

void session::handle_sqlserver_read(const boost::system::error_code& error,
                                    size_t bytes_transferred)
{
    cout << "session::handle_sqlserver_read()" << endl;
    if (!error)
    {
        cout << "session::handle_sqlserver_read() (read: " 
             << bytes_transferred << ")"
             << endl;
        boost::asio::async_write(socket_,
            boost::asio::buffer(data_, bytes_transferred),
            boost::bind(&session::handle_write, this,
            boost::asio::placeholders::error, bytes_transferred));
    }
    else
    {
        delete this;
    }
}

void session::handle_write(const boost::system::error_code& error,
                           size_t bytes_transferred)
{
    static int count = 0;
    cout << ++count << ". session::handle_write()" << endl;
    if (!error)
    {
        cout << "session::handle_write() (read: " 
             << bytes_transferred << ")"
             << endl;
        socket_.async_read_some(boost::asio::buffer(data_, max_length),
            boost::bind(&session::handle_read, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
    }
    else
    {
        delete this;
    }
}

void session::handle_sqlserver_write(const boost::system::error_code& error,
                                     size_t bytes_transferred)
{
    cout << "session::handle_sqlserver_write()" << endl;
    if (!error)
    {
        cout << "session::handle_sqlserver_write() (read: " 
             << bytes_transferred << ")"
             << endl;
        sqlsocket_.async_read_some(boost::asio::buffer(data_, max_length),
            boost::bind(&session::handle_sqlserver_read, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
    }
    else
    {
        delete this;
    }
}

void session::handle_sqlserver_connect(const boost::system::error_code& error,
                                       tcp::resolver::iterator endpoint_iterator)
{
    cout << "session::handle_sqlserver_connect()" << endl;
    if (!error)
    {
        socket_.async_read_some(boost::asio::buffer(data_, max_length),
            boost::bind(&session::handle_read, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
    }
    else if (endpoint_iterator != tcp::resolver::iterator())
    {
        sqlsocket_.close();
        tcp::endpoint endpoint = *endpoint_iterator;
        sqlsocket_.async_connect(endpoint,
            boost::bind(&session::handle_sqlserver_connect, this,
            boost::asio::placeholders::error, ++endpoint_iterator));
    }
}

对于我的代理类型,我是否需要使用其他方法而不是 async_* ? 我正在从我的公司想要重新启动的一些旧项目中移植代码,但使用 boost 而不是以前使用的 Winsock 东西。

知道可能是什么问题吗?

旧代码做了这样的事情: 带有accept方法调用的main方法创建了两个线程

CreateThread(0, 0, (LPTHREAD_START_ROUTINE)listenatclient, (LPVOID)cs, 0, 0);
CreateThread(0, 0, (LPTHREAD_START_ROUTINE)listenatserver, (LPVOID)cs, 0, 0);

,这些线程调用了以下函数:

void listenatclient(LPVOID connection)
{
    connection_s* cs = (connection_s*)connection;
    char inMessagecli[MSG_SIZE];
    int rcount = 0;

    ...

    do
    {
        memset(inMessagecli, 0, MSG_SIZE);
        rcount = recv((SOCKET)cs->client, inMessagecli, MSG_SIZE, 0);
        if (rcount != SOCKET_ERROR)
        {
            // analyze package
            ...

            send((SOCKET)cs->server, inMessagecli, rcount, 0);
        }
    } while (rcount > 0);
}

void listenatserver(LPVOID connection)
{
    connection_s* cs = (connection_s*)connection;
    char inMessageserv[MSG_SIZE];
    int rcount = 0;

    do
    {
        memset(inMessageserv, 0, MSG_SIZE);
        rcount = recv((SOCKET)cs->server, inMessageserv, MSG_SIZE, 0);
        if (rcount != SOCKET_ERROR)
        {
            send((SOCKET)cs->client, inMessageserv, rcount, 0);         
        }
    } while (rcount > 0);
}

[编辑]: 我尝试同时为客户端和 sqlserver 运行 async_read 命令,但现在我总是崩溃,有时在 boost::bind 中,有时在 boost 库的其他部分。

似乎会创建 2 或 3 个连接(3 个会话)。在关闭第一个会话时,崩溃似乎发生在第二个会话中。

boost asio 不安全还是我在这里做了一些非常错误的事情:-)?

我在这里发布了小代理服务器的代码:

session.h : link

session.cpp : 链接

server.h:链接

server.cpp:链接

ProxyServer.cpp:链接

I'm writing a simple ProxyServer that analyzes packages and sends them to another server instance, e.g. something like this:

client -> MyProxy -> SQLServer ->
client <- MyProxy <- SQLServer <-

It should run in an infinite loop.
My problem now is that the proxy seems to loose packages, sometimes it even hangs.
When I add a lot of debug information (which is written to the console), the ProxyServer is
much more stable. It seems like the ProxyServer is too fast.. :-)

I'm pretty sure I'm doing something wrong, here is code of my session class (the code is derived from the Boost::Asio examples).

#include "session.h"

#include <iostream>
using namespace std;

session::session(boost::asio::io_service& io_service)
: socket_(io_service)
, sqlsocket_(io_service)
, io_service_(io_service)
, resolver(io_service)
{
    cout << "session::session()" << endl;
}

session::~session()
{
    cout << "session::~session()" << endl;
    cout << "closing session ..." << endl;
}

tcp::socket& session::socket()
{
    return socket_;
}

void session::start()
{
    cout << "session::start()" << endl;
    cout << "starting session ..." << endl;

    // connect to the sqlserver database
    tcp::resolver::query query("192.168.1.50", "1317");
    tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
    tcp::endpoint endpoint = *endpoint_iterator;

    sqlsocket_.async_connect(endpoint,
        boost::bind(&session::handle_sqlserver_connect, this,
        boost::asio::placeholders::error, ++endpoint_iterator));

    // TODO: connect to the connector
}

void session::handle_read(const boost::system::error_code& error,
                          size_t bytes_transferred)
{
    cout << "session::handle_read()" << endl;
    if (!error)
    {
        cout << "session::handle_read() (read: " 
             << bytes_transferred << ")"
             << endl;
        boost::asio::async_write(sqlsocket_,
            boost::asio::buffer(data_, bytes_transferred),
            boost::bind(&session::handle_sqlserver_write, this,
            boost::asio::placeholders::error, bytes_transferred));
    }
    else
    {
        delete this;
    }
}

void session::handle_sqlserver_read(const boost::system::error_code& error,
                                    size_t bytes_transferred)
{
    cout << "session::handle_sqlserver_read()" << endl;
    if (!error)
    {
        cout << "session::handle_sqlserver_read() (read: " 
             << bytes_transferred << ")"
             << endl;
        boost::asio::async_write(socket_,
            boost::asio::buffer(data_, bytes_transferred),
            boost::bind(&session::handle_write, this,
            boost::asio::placeholders::error, bytes_transferred));
    }
    else
    {
        delete this;
    }
}

void session::handle_write(const boost::system::error_code& error,
                           size_t bytes_transferred)
{
    static int count = 0;
    cout << ++count << ". session::handle_write()" << endl;
    if (!error)
    {
        cout << "session::handle_write() (read: " 
             << bytes_transferred << ")"
             << endl;
        socket_.async_read_some(boost::asio::buffer(data_, max_length),
            boost::bind(&session::handle_read, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
    }
    else
    {
        delete this;
    }
}

void session::handle_sqlserver_write(const boost::system::error_code& error,
                                     size_t bytes_transferred)
{
    cout << "session::handle_sqlserver_write()" << endl;
    if (!error)
    {
        cout << "session::handle_sqlserver_write() (read: " 
             << bytes_transferred << ")"
             << endl;
        sqlsocket_.async_read_some(boost::asio::buffer(data_, max_length),
            boost::bind(&session::handle_sqlserver_read, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
    }
    else
    {
        delete this;
    }
}

void session::handle_sqlserver_connect(const boost::system::error_code& error,
                                       tcp::resolver::iterator endpoint_iterator)
{
    cout << "session::handle_sqlserver_connect()" << endl;
    if (!error)
    {
        socket_.async_read_some(boost::asio::buffer(data_, max_length),
            boost::bind(&session::handle_read, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
    }
    else if (endpoint_iterator != tcp::resolver::iterator())
    {
        sqlsocket_.close();
        tcp::endpoint endpoint = *endpoint_iterator;
        sqlsocket_.async_connect(endpoint,
            boost::bind(&session::handle_sqlserver_connect, this,
            boost::asio::placeholders::error, ++endpoint_iterator));
    }
}

Do I need to use other methods instead of async_* for my type of proxy?
I'm porting the code from some old project that my company wants to restart again, but with boost instead of the Winsock stuff that was used before.

Any idea what could be the problem?

The old code did something like this:
The main method with the accept method call created two threads

CreateThread(0, 0, (LPTHREAD_START_ROUTINE)listenatclient, (LPVOID)cs, 0, 0);
CreateThread(0, 0, (LPTHREAD_START_ROUTINE)listenatserver, (LPVOID)cs, 0, 0);

and the threads called the following functions:

void listenatclient(LPVOID connection)
{
    connection_s* cs = (connection_s*)connection;
    char inMessagecli[MSG_SIZE];
    int rcount = 0;

    ...

    do
    {
        memset(inMessagecli, 0, MSG_SIZE);
        rcount = recv((SOCKET)cs->client, inMessagecli, MSG_SIZE, 0);
        if (rcount != SOCKET_ERROR)
        {
            // analyze package
            ...

            send((SOCKET)cs->server, inMessagecli, rcount, 0);
        }
    } while (rcount > 0);
}

void listenatserver(LPVOID connection)
{
    connection_s* cs = (connection_s*)connection;
    char inMessageserv[MSG_SIZE];
    int rcount = 0;

    do
    {
        memset(inMessageserv, 0, MSG_SIZE);
        rcount = recv((SOCKET)cs->server, inMessageserv, MSG_SIZE, 0);
        if (rcount != SOCKET_ERROR)
        {
            send((SOCKET)cs->client, inMessageserv, rcount, 0);         
        }
    } while (rcount > 0);
}

[EDIT]:
I tried to run the async_read commands for the client and the sqlserver simultaneously, but now I get crashes all the time, sometimes in boost::bind, sometimes in other parts of the boost library.

What seems to happen is that 2 or three conections are created ( 3 sessions). While closing the first session, the crash seems to happen in the second session.

Is boost asio not treadsafe or am I doing something terribly wrong here :-) ?

I posted the code for the little ProxyServer here:

session.h : link

session.cpp : link

server.h: link

server.cpp: link

ProxyServer.cpp: link

如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论, 你可以免费 注册 一个本站的账号。

评论(1

意中人 2024-11-21 05:36:37

我怀疑发生的情况是您的 async_read_some 调用之一返回“一些”数据,但不足以让 SQL Server 满意它已收到完整的请求。您的代码始终遵循 read_from_client -> 的路径发送到服务器->从服务器读取 ->发送到客户端。它不处理您需要 read_from_client -> 的情况send_to_server - > read_from_client - >发送到服务器->从服务器读取 ->发送到客户端。

到目前为止,您编写的代码执行的操作与原始代码不同。具体来说,旧代码同时侦听两个套接字上的读取。幸运的是,由于您使用的是 ASIO,因此您不需要搞乱线程。只需在两个套接字上同时发出 async_read_some 请求并异步处理它们即可。

I suspect what is happening is that one of your async_read_some calls is returning "some" data, but not enough for the SQL server to be satisfied that it has received a complete request. You code always follows the path of read_from_client -> send_to_server -> read_from_server -> send_to_client. It does not handle the case where you need read_from_client -> send_to_server - >read_from_client -> send_to_server -> read_from_server -> send_to_client.

The code you have written so far does not perform the same operations as the original. Specifically, the old code was simultaneously listening for reads on both sockets. Fortunately, since you're using ASIO, you don't need to mess with threads. Just issue issue simultaneous async_read_some requests on both sockets and deal with them asynchronously.

~没有更多了~
我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站,即表示您同意使用 Cookies 和您的相关数据。
原文