Как правильно завершить работу tcp-сервера asio?

Как правильно выключить асинхронный сервер boost asio tcp? Мое текущее решение обычно заходит в тупик в деструкторе. Почему?

class connection;

typedef std::set<shared_ptr<connection>> connection_set;

class connection : public enable_shared_from_this<connection>
{    
    shared_ptr<tcp::socket>         socket_; 

    std::array<char, 8192>          data_;

    shared_ptr<connection_set>      connection_set_;
public:
    static shared_ptr<connection> create(shared_ptr<tcp::socket> socket, shared_ptr<connection_set> connection_set)
    {
        auto con = shared_ptr<connection>(new connection(std::move(socket), std::move(connection_set)));
        con->read_some();
        return con;
    }

    void on_next(const event& e)
    {   
        // async_write_some ...
    }

private:
    connection(shared_ptr<tcp::socket> socket, shared_ptr<connection_set> connection_set) 
        : socket_(std::move(socket))
        , connection_set_(std::move(connection_set))
    {
    }

    void handle_read(const boost::system::error_code& error, size_t bytes_transferred) 
    {       
        if(!error)
        {
            on_read(std::string(data_.begin(), data_.begin() + bytes_transferred));     
            read_some();
        }  
        else if (error != boost::asio::error::operation_aborted)
            connection_set_->erase(shared_from_this());
        else
            read_some();
    }

    void handle_write(const shared_ptr<std::vector<char>>& data, const boost::system::error_code& error, size_t bytes_transferred)
    {
        if(!error)          
        {
        }
        else if (error != boost::asio::error::operation_aborted)
            connection_set_->erase(shared_from_this());
    }

    void read_some()
    {
        socket_->async_read_some(boost::asio::buffer(data_.data(), data_.size()), std::bind(&connection::handle_read, shared_from_this(), std::placeholders::_1, std::placeholders::_2));
    }

    void on_read(std::string str)
    {
        // parse the string...
    }
};

class tcp_observer
{
    boost::asio::io_service             service_;
    tcp::acceptor                       acceptor_;
    std::shared_ptr<connection_set>     connection_set_;
    boost::thread                       thread_;

public:
    tcp_observer(unsigned short port)
        : acceptor_(service_, tcp::endpoint(tcp::v4(), port))
        , thread_(std::bind(&boost::asio::io_service::run, &service_))
    {
        start_accept(); 
    }

    ~tcp_observer()
    {
        // Deadlocks...
        service_.post([=]
        {
            acceptor_.close();
            connection_set_->clear();
        });
        thread_.join();
    }

    void on_next(const event& e)
    {
        service_.post([=]
        {
            BOOST_FOREACH(auto& connection, *connection_set_)
                connection->on_next(e);
        });
    }   
private:        
    void start_accept() 
    {
        auto socket = std::make_shared<tcp::socket>(service_);
        acceptor_.async_accept(*socket, std::bind(&tcp_observer::handle_accept, this, socket, std::placeholders::_1));
    }

    void handle_accept(const shared_ptr<tcp::socket>& socket, const boost::system::error_code& error) 
    {
        if (!acceptor_.is_open())
            return;

        if (!error)     
            connection_set_->insert(connection::create(socket, connection_set_));

        start_accept();
    }
};