Asio hangs on system call close (Mac OS X)

[ruslo]

This test:

#include <thread>
#include <asio.hpp>
#include <boost/chrono.hpp>

int main() {
  for (int i = 0; i < 10000; ++i) {
    std::cout << i << std::endl;

    asio::io_service ioService;
    asio::deadline_timer timer(ioService);

    ioService.post([&](){
      timer.expires_from_now(boost::posix_time::milliseconds(1));
      timer.async_wait([&](const asio::error_code &) {});
    });

    ioService.post([&](){
      ioService.post([](){});
    });

    // Run some threads
    std::vector<std::thread> threads;
    for (auto i=0; i<3; ++i) {
      threads.emplace_back([&](){ ioService.run(); });
    }
    for (auto& i: threads) {
      i.join();
    }
  }
  return EXIT_SUCCESS;
}

hangs on system call close:

• asio/asio/include/asio/detail/impl/pipe_select_interrupter.ipp

  Line 76 in 51bba37

  ::close(read_descriptor_);

Tested on: Mac OS X 10.9.1/10.8.5, clang -stdlib=libc++ -std=c++11, boost 1.55.0
master branch: e80030f
cpp11-only branch: f1cbf4f

Originally from: http://stackoverflow.com/questions/21835749/boost-asio-io-service-destructor-hangs-on-os-x

[chriskohlhoff]

Do you have any support channel to Apple? The OS is behaving very oddly, e.g. at the end of the test the process is still visible in top and is stuck in an "uninterruptible wait". The OS appears to be "leaking" something, although I am not sure exactly what.

This bug is also relevant: https://svn.boost.org/trac/boost/ticket/5021. A change was made under this bug that seems to have reduced the problem but not eliminated it.

[ruslo]

Do you have any support channel to Apple?

I've filed a bug to bugreport.apple.com, but I doubt this will have any effect (:

The OS is behaving very oddly

I think the root of the issue is race condition somewhere. I can reproduce bug on real 4 core machine, but not on virtual machine with 1 or 2 cores support (for >=3 deadlock occurs).

As @villintehaspam mention in SO question one of the possible workarounds is to
protect by mutex kevent system call inside kqueue_reactor::interrupt() method.

I can send PR if this is acceptable solution

• https://github.com/ruslo/asio/commit/5720cd7c3cd7792dbe9ae6dbe6c783e7c33fdc36

[chriskohlhoff]

It is hard to find a clear statement on kqueue thread safety, however the information I have come across seems to suggest that it is supposed to be.

I have just discovered that when changing from EV_ONESHOT to EV_CLEAR (as part of boost ticket 5021) I missed the opportunity to make the kqueue_reactor fully edge triggered. Let me try making a small change in this way to see if it makes a difference for you.

[chriskohlhoff]

Can you please try baee767.

[ruslo]

Can you please try baee767.

Example works nice!

[villintehaspam]

Hey! I wrote the original question on StackOverflow. I've added a support case with Apple, dunno how long it usually takes to get assistance though.

Just so that I understand, is baee767 intended as a fix for this, a workaround, or just another way of doing things in asio that might avoid the problem?

Cheers, V

[chriskohlhoff]

The latter, i.e. a different approach that hopefully will not trigger the problem. Although, to be honest, I should have made this change at the time I switched from EV_ONESHOT to EV_CLEAR.

I still suspect there is some sort of OS bug here, as processes really shouldn't get stuck in an uninterruptible wait in a close system call. I had to reboot after running the test program several times, as even kill -9 did nothing.

If you can give this change some thorough real-world testing it would be much appreciated.

[villintehaspam]

I agree, I don't think the OS should behave this way.

I made a quick test by applying the changes to the Boost 1.54 sources that I am using and was then unable to reproduce the issue. Might there be any issues with not applying to the latest version?

Otherwise I will apply these changes and run my actual application with it for a while.

Cheers, V

[chriskohlhoff]

That should be fine.

[villintehaspam]

I seem to get an issue, where it appears as if my asio-based programs stop communicating sometimes (very infrequently). It looks as if the program just doesn't recognise that there is something to read from the socket.

The basic structure of my setup is this: Process A starts process B and C. B listens for incoming connections, C connects to B. Based on commands from C, B launches D, that also connects back to B. B sends and receives messages to and from C and D, forwarding some between them and processing others itself. B, C, and D are using Asio and this has worked well in the past. I am trying the patch by running a set of system tests in A repeatedly, which will do the above once per test.

I am now getting an issue about once per day where the programs stop communicating (process C believes it has sent a message to D through B and awaits a response, D never gets the message, and at the moment I do not know whether B has gotten the message and forwarded it or not). I can see that the processes do not seem to be deadlocked, but rather just wait for things to happen with asio.

At the moment I cannot pinpoint the exact problem (the code is multithreaded and somewhat complicated), so I cannot say for sure that this is related to the patch. The source has been running fine on Windows, but the port to OS X is new. Previously I was unable to run the tests longer than about an hour on OS X due to the original issue.

Note: the code in B uses fork() + execve() to launch D. I am currently not using the notify_fork() function as it seems to be difficult to use this properly in a multithreaded environment (see http://stackoverflow.com/q/21529540/227322 ). Can this have a negative impact?

I will continue to investigate and see if I can pinpoint whether this is asio-related or not.

Cheers,
V

[villintehaspam]

Ignore my last comment. I have found an unrelated multithreading issue that caused the above problem.

Cheers,
V

[villintehaspam]

I do think I have found an issue with the patch though. I have created a small repro case based on the async tcp echo server example.

When I run this example using plain Boost 1.55, I typically am able to run around 8190 iterations before connections start to fail with EADDRNOTAVAIL, which I assume means that there simply are no more temporary sockets available for the system to use.

However, if I run with Boost 1.55 + the patch from here though, the test typically hangs in the first iteration directly.

The test is really simple. It basically consists of the async tcp server example from http://www.boost.org/doc/libs/1_55_0/doc/html/boost_asio/example/cpp11/echo/async_tcp_echo_server.cpp, just slightly modified. I have added a client class that does a synchronous resolve, followed by an asynchronous connect and a couple of async write/read cycles.

The test starts a server, then in a loop creates a client ("c2") and lets it do it's thing.

The hang occurs if I create an extra client instance "c1" before c2, which is only allowed to do the synchronous resolve + asynchronous connect. But io_service::run() is never called for that instance.

To see a bit more about what is going on, the test simply uses put-debugging to indicate what is happening. A theory would be that if the server is able to call async_accept() before the second client calls async_connect(), everything works, but not if things happen in the other order.

I posted the entire test case below. Apologies for the length and for any errors or sillyness on my part.

Cheers,
V

#include <cstdlib>
#include <iostream>
#include <memory>
#include <mutex>
#include <thread>
#include <utility>
#include <boost/asio.hpp>

using boost::asio::ip::tcp;
using namespace boost;
using namespace boost::system;

static std::mutex global_log_mutex;

class Log {
public:
  ~Log() {
    auto tid = (pid_t)(uintptr_t)pthread_self();
    std::lock_guard<std::mutex> lock(global_log_mutex);
    printf("%d: %s\n", tid, m_str.str().c_str());
  }

  std::ostream & stream() { return m_str; }
private:
  std::stringstream m_str;
};

#define LOG() Log().stream()

class session : public std::enable_shared_from_this<session> {
public:
  session(tcp::socket socket)
    : socket_(std::move(socket))
  {}

  void start() {
    do_read();
  }

private:
  void do_read() {
    auto self(shared_from_this());
    LOG() << "Server: starting to read on connection.";
    socket_.async_read_some(asio::buffer(data_, max_length),
        [this, self](error_code ec, std::size_t length)
        {
          LOG() << "Server: read completed with e: " << ec << ", " << length << " bytes.";
          if (!ec) {
            do_write(length);
          }
        });
  }

  void do_write(std::size_t length) {
    auto self(shared_from_this());
    LOG() << "Server: starting to write on connection.";
    asio::async_write(socket_, asio::buffer(data_, length),
        [this, self](error_code ec, std::size_t length)
        {
          LOG() << "Server: write completed with e: " << ec << ", " << length << " bytes.";
          if (!ec) {
            do_read();
          }
        });
  }

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

class server {
public:
  enum { NUM_THREADS = 4 };

  server(short port)
    : acceptor_(io_service_, tcp::endpoint(tcp::v4(), port)),
      socket_(io_service_)
  {
    LOG() << "Server: started.";
    do_accept();
    for (int i = 0; i < NUM_THREADS; ++i) {
      std::thread tmp([&]() {
        io_service_.run();
      });
      threads_.push_back(std::move(tmp));
    }
  }

  ~server() {
    for (auto &t : threads_) {
        if (t.joinable()) {
            t.join();
      }
    }
  }

  void stop() {
    io_service_.post([&]() {
      std::lock_guard<std::mutex> lock(acceptor_mutex_);
        acceptor_.cancel();
    });
  }

private:
  void do_accept() {
    LOG() << "Server: accepting connection...";
    std::lock_guard<std::mutex> lock(acceptor_mutex_);
    acceptor_.async_accept(socket_,
        [this](error_code ec)
        {
          LOG() << "Server: accept completed with e: " << ec;
          if (!ec) {
            std::make_shared<session>(std::move(socket_))->start();
          }

          if (ec != asio::error::operation_aborted) {
              do_accept();
            }
        });
  }

  asio::io_service io_service_;
  std::mutex acceptor_mutex_;
  tcp::acceptor acceptor_;
  tcp::socket socket_;
  std::vector<std::thread> threads_;
};

class client {
public:
  client(short port)
  : socket_(io_service_)
  , count_(0)
  {
    LOG() << "Client: resolving address...";
    boost::asio::ip::tcp::resolver resolver(io_service_);
    boost::asio::ip::tcp::resolver::query query("127.0.0.1", std::to_string(port));
    boost::asio::ip::tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
    boost::asio::ip::tcp::resolver::iterator end;
    LOG() << "Client: address resolved. Connecting...";

    do_connect(endpoint_iterator);
  }

  error_code run() {
    io_service_.run();
    return err_;
  }

private:
  enum {
    BUFFER_SIZE = 50,
    MAX_ITERATIONS = 3,
  };

  void do_connect(tcp::resolver::iterator i) {
    tcp::resolver::iterator end;
    asio::async_connect(socket_, i, end, [&](const error_code &ec, tcp::resolver::iterator j) {
      LOG() << "Client: connection attempt completed with e: " << ec;
      if (!ec) {
        do_write();
      } else {
        err_ = ec;
      }
    });
  }

  void do_write() {
    if (++count_ == MAX_ITERATIONS) {
      LOG() << "Client: done.";
      return;
    }
    LOG() << "Client: initiating write.";
    asio::async_write(socket_, asio::buffer(buffer_, BUFFER_SIZE), [&](const error_code &ec, size_t bytes) {
      LOG() << "Client: write completed with e: " << ec << ", " << bytes << " bytes.";
      if (!ec) {
        do_read();
      } else {
        err_ = ec;
      }
    });
  }

  void do_read() {
    LOG() << "Client: initiating read.";
    asio::async_read(socket_, asio::buffer(buffer_, BUFFER_SIZE), [&](const error_code &ec, size_t bytes) {
      LOG() << "Client: read completed with e: " << ec << ", " << bytes << " bytes.";
      if (!ec) {
        do_write();
      } else {
        err_ = ec;
      }
    });
  }

  asio::io_service io_service_;
  tcp::socket socket_;

  char buffer_[BUFFER_SIZE];
  int count_;
  error_code err_;
};

int main(int argc, char* argv[])
{
  try {
    if (argc != 2) {
      std::cerr << "Usage: testasio <port>" << std::endl;
      return 1;
    }

    auto port = std::atoi(argv[1]);
    server s(port);
    for (int i = 0; i < 10000; ++i) {
        LOG() << "Iteration: " << i;
      client c1(port);
      client c2(port);
      auto ec = c2.run();
      if (ec) {
        LOG() << "Failed: " << ec;
        break;
      }
    }
    s.stop();
  } catch (std::exception& e) {
    std::cerr << "Exception: " << e.what() << std::endl;
  }

  return 0;
}

[villintehaspam]

Hi,

Do you have any ideas on what could cause this? As it currently stands, asio is not reliable for us on OS X - either there is a hang on exit or connections don't work properly, neither of which are great… :) If you cannot reproduce it, maybe I've done something wrong, or applied the patch incorrectly?

Cheers,
V

[chriskohlhoff]

Can you please try 39d2e4b as well.

N.B. I have not yet checked to see if this causes the original problem to reoccur.

[villintehaspam]

Thanks! This seems to do the trick for me - I haven't been able to reproduce any of the problems using this patch (so far).

Cheers,
Ragnar