dimanche 17 mai 2015

Boost.Asio concurrency problems on threads

i'm trying to make this program that has a few features:

-Networking with Boost.Asio (yeah, i got most of it figured out).

-Multi-threading with Boost.Thread and Boost.Asio.

The important aspects of the program are a few threads, one that will constantly try to read and one that will check the contents of an std::vector and write them to a textbox (i'm using Nana C++ for the GUI) and on the socket. So i know that i need to use atomics in some points, because i also need these threads to be able to close themselves (using a bool maybe?) and be able to read and write in the vector. But this is proving very difficult.

What i really want to know is: How can i have these threads doing the write and read operations and accessing the vector without damaging the data or causing undefined behavior? And how can i "kill" the threads without doing any harm?

I do know that this is very confuse, so the following is the code i've built so far:

short port;
wstring password;
atomic<bool> stop = false;
vector<wstring> msgQueue;
atomic<vector<wstring>> sLog;
boost::asio::io_service io_service;

class session
{
public:
    session(boost::asio::io_service& io_service)
        : socket_(io_service)
    {
    }

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

    void start()
    {
        thread(read);
        while (!stop)
        {
            if (!msgQueue.empty())
            {
                std::string sending = ws2s(msgQueue.front());
                msgQueue.erase(msgQueue.begin());
                socket_.write_some(boost::asio::buffer(sending.c_str(), sending.length()));
            }
        }
    }

    void read()
    {
        while (!stop)
        {
            socket_.read_some(boost::asio::buffer(data_, max_length));
        }
    }

    void handle_read(const boost::system::error_code& error,
        size_t bytes_transferred)
    {
        if (!error)
        {
            //boost::asio::async_write(socket_,
                //boost::asio::buffer(data_, bytes_transferred),
                //);
        }
        else
        {
            delete this;
        }
    }

    void handle_write(const boost::system::error_code& error)
    {
        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
        {
            delete this;
        }
    }

private:
    tcp::socket socket_;
    enum { max_length = 1024 };
    char data_[max_length];
};

class server
{
public:
    server(boost::asio::io_service& io_service, short port)
        : io_service_(io_service),
        acceptor_(io_service, tcp::endpoint(tcp::v4(), port))
    {
        session* new_session = new session(io_service_);
        acceptor_.async_accept(new_session->socket(),
            boost::bind(&server::handle_accept, this, new_session,
            boost::asio::placeholders::error));
    }

    void handle_accept(session* new_session,
        const boost::system::error_code& error)
    {
        if (!error)
        {
            new_session->start();
            new_session = new session(io_service_);
            acceptor_.async_accept(new_session->socket(),
                boost::bind(&server::handle_accept, this, new_session,
                boost::asio::placeholders::error));
        }
        else
        {
            delete new_session;
        }
    }

private:
    boost::asio::io_service& io_service_;
    tcp::acceptor acceptor_;
};

    void praca()
    {
        try
        {
            server s(io_service, port);
            sLog; // push_back() here?
            io_service.run();
        }
        catch (std::exception& e)
        {
            std::string raps(const_cast<char*>(e.what()));
            MsgBox(L"Error", s2ws(raps));
        }
    }

    void start()
    {
        stop = false;
        thread(praca);
        append(L"Server is now running. Waiting for client connection...", false);
    }

I do know this code has major flaws and that it would certainly cause a lot of exceptions if it compiled.

I also forgot to mention, i wanted to use a password while sending/receiving or just on connection time, but i think i can figure that out once the current mess is solved.

Any suggestions on how can i do this, or if i should use something else instead of what i've been trying so far (the methods or the libraries)?

Btw, i'm using Visual Studio 2013 on Windows 8.1.

Thanks in advance.

Aucun commentaire:

Enregistrer un commentaire