Index: lldb/trunk/include/lldb/Host/MainLoop.h =================================================================== --- lldb/trunk/include/lldb/Host/MainLoop.h +++ lldb/trunk/include/lldb/Host/MainLoop.h @@ -42,6 +42,7 @@ public: typedef std::unique_ptr SignalHandleUP; + MainLoop(); ~MainLoop() override; ReadHandleUP RegisterReadObject(const lldb::IOObjectSP &object_sp, @@ -71,6 +72,9 @@ void UnregisterSignal(int signo); private: + void ProcessReadObject(IOObject::WaitableHandle handle); + void ProcessSignal(int signo); + class SignalHandle { public: ~SignalHandle() { m_mainloop.UnregisterSignal(m_signo); } @@ -97,6 +101,9 @@ llvm::DenseMap m_read_fds; llvm::DenseMap m_signals; +#if HAVE_SYS_EVENT_H + int m_kqueue; +#endif bool m_terminate_request : 1; }; Index: lldb/trunk/source/Host/common/MainLoop.cpp =================================================================== --- lldb/trunk/source/Host/common/MainLoop.cpp +++ lldb/trunk/source/Host/common/MainLoop.cpp @@ -18,6 +18,11 @@ #include #include +// Multiplexing is implemented using kqueue on systems that support it (BSD +// variants including OSX). On linux we use ppoll, while android uses pselect +// (ppoll is present but not implemented properly). On windows we use WSApoll +// (which does not support signals). + #if HAVE_SYS_EVENT_H #include #elif defined(LLVM_ON_WIN32) @@ -65,92 +70,72 @@ class MainLoop::RunImpl { public: - // TODO: Use llvm::Expected - static std::unique_ptr Create(MainLoop &loop, Error &error); - ~RunImpl(); + RunImpl(MainLoop &loop); + ~RunImpl() = default; Error Poll(); - - template void ForEachReadFD(F &&f); - template void ForEachSignal(F &&f); + void ProcessEvents(); private: MainLoop &loop; #if HAVE_SYS_EVENT_H - int queue_id; std::vector in_events; struct kevent out_events[4]; int num_events = -1; - RunImpl(MainLoop &loop, int queue_id) : loop(loop), queue_id(queue_id) { - in_events.reserve(loop.m_read_fds.size() + loop.m_signals.size()); - } #else - std::vector signals; #ifdef FORCE_PSELECT fd_set read_fd_set; #else std::vector read_fds; #endif - RunImpl(MainLoop &loop) : loop(loop) { - signals.reserve(loop.m_signals.size()); - } - sigset_t get_sigmask(); #endif }; #if HAVE_SYS_EVENT_H -MainLoop::RunImpl::~RunImpl() { - int r = close(queue_id); - assert(r == 0); - (void)r; -} -std::unique_ptr MainLoop::RunImpl::Create(MainLoop &loop, Error &error) -{ - error.Clear(); - int queue_id = kqueue(); - if(queue_id < 0) { - error = Error(errno, eErrorTypePOSIX); - return nullptr; - } - return std::unique_ptr(new RunImpl(loop, queue_id)); +MainLoop::RunImpl::RunImpl(MainLoop &loop) : loop(loop) { + in_events.reserve(loop.m_read_fds.size()); } Error MainLoop::RunImpl::Poll() { - in_events.resize(loop.m_read_fds.size() + loop.m_signals.size()); + in_events.resize(loop.m_read_fds.size()); unsigned i = 0; for (auto &fd : loop.m_read_fds) EV_SET(&in_events[i++], fd.first, EVFILT_READ, EV_ADD, 0, 0, 0); - for (const auto &sig : loop.m_signals) - EV_SET(&in_events[i++], sig.first, EVFILT_SIGNAL, EV_ADD, 0, 0, 0); - - num_events = kevent(queue_id, in_events.data(), in_events.size(), out_events, - llvm::array_lengthof(out_events), nullptr); + num_events = kevent(loop.m_kqueue, in_events.data(), in_events.size(), + out_events, llvm::array_lengthof(out_events), nullptr); if (num_events < 0) return Error("kevent() failed with error %d\n", num_events); return Error(); } -template void MainLoop::RunImpl::ForEachReadFD(F &&f) { +void MainLoop::RunImpl::ProcessEvents() { assert(num_events >= 0); for (int i = 0; i < num_events; ++i) { - f(out_events[i].ident); if (loop.m_terminate_request) return; + switch (out_events[i].filter) { + case EVFILT_READ: + loop.ProcessReadObject(out_events[i].ident); + break; + case EVFILT_SIGNAL: + loop.ProcessSignal(out_events[i].ident); + break; + default: + llvm_unreachable("Unknown event"); + } } } -template void MainLoop::RunImpl::ForEachSignal(F && f) {} #else -MainLoop::RunImpl::~RunImpl() {} -std::unique_ptr MainLoop::RunImpl::Create(MainLoop &loop, Error &error) -{ - error.Clear(); - return std::unique_ptr(new RunImpl(loop)); +MainLoop::RunImpl::RunImpl(MainLoop &loop) : loop(loop) { +#ifndef FORCE_PSELECT + read_fds.reserve(loop.m_read_fds.size()); +#endif } sigset_t MainLoop::RunImpl::get_sigmask() { @@ -162,18 +147,14 @@ assert(ret == 0); (void) ret; - for (const auto &sig : loop.m_signals) { - signals.push_back(sig.first); + for (const auto &sig : loop.m_signals) sigdelset(&sigmask, sig.first); - } return sigmask; #endif } #ifdef FORCE_PSELECT Error MainLoop::RunImpl::Poll() { - signals.clear(); - FD_ZERO(&read_fd_set); int nfds = 0; for (const auto &fd : loop.m_read_fds) { @@ -188,20 +169,8 @@ return Error(); } - -template void MainLoop::RunImpl::ForEachReadFD(F &&f) { - for (const auto &fd : loop.m_read_fds) { - if(!FD_ISSET(fd.first, &read_fd_set)) - continue; - - f(fd.first); - if (loop.m_terminate_request) - return; - } -} #else Error MainLoop::RunImpl::Poll() { - signals.clear(); read_fds.clear(); sigset_t sigmask = get_sigmask(); @@ -220,33 +189,47 @@ return Error(); } +#endif -template void MainLoop::RunImpl::ForEachReadFD(F &&f) { +void MainLoop::RunImpl::ProcessEvents() { +#ifdef FORCE_PSELECT + for (const auto &fd : loop.m_read_fds) { + if (!FD_ISSET(fd.first, &read_fd_set)) + continue; + IOObject::WaitableHandle handle = fd.first; +#else for (const auto &fd : read_fds) { if ((fd.revents & POLLIN) == 0) continue; - - f(fd.fd); + IOObject::WaitableHandle handle = fd.fd; +#endif if (loop.m_terminate_request) return; - } -} -#endif -template void MainLoop::RunImpl::ForEachSignal(F &&f) { - for (int sig : signals) { - if (g_signal_flags[sig] == 0) - continue; // No signal - g_signal_flags[sig] = 0; - f(sig); + loop.ProcessReadObject(handle); + } + for (const auto &entry : loop.m_signals) { if (loop.m_terminate_request) return; + if (g_signal_flags[entry.first] == 0) + continue; // No signal + g_signal_flags[entry.first] = 0; + loop.ProcessSignal(entry.first); } } #endif +MainLoop::MainLoop() { +#if HAVE_SYS_EVENT_H + m_kqueue = kqueue(); + assert(m_kqueue >= 0); +#endif +} MainLoop::~MainLoop() { +#if HAVE_SYS_EVENT_H + close(m_kqueue); +#endif assert(m_read_fds.size() == 0); assert(m_signals.size() == 0); } @@ -298,24 +281,30 @@ new_action.sa_flags = SA_SIGINFO; sigemptyset(&new_action.sa_mask); sigaddset(&new_action.sa_mask, signo); - sigset_t old_set; - if (int ret = pthread_sigmask(SIG_BLOCK, &new_action.sa_mask, &old_set)) { - error.SetErrorStringWithFormat("pthread_sigmask failed with error %d\n", - ret); - return nullptr; - } - info.was_blocked = sigismember(&old_set, signo); - if (sigaction(signo, &new_action, &info.old_action) == -1) { - error.SetErrorToErrno(); - if (!info.was_blocked) - pthread_sigmask(SIG_UNBLOCK, &new_action.sa_mask, nullptr); - return nullptr; - } + g_signal_flags[signo] = 0; + + // Even if using kqueue, the signal handler will still be invoked, so it's + // important to replace it with our "bening" handler. + int ret = sigaction(signo, &new_action, &info.old_action); + assert(ret == 0 && "sigaction failed"); + +#if HAVE_SYS_EVENT_H + struct kevent ev; + EV_SET(&ev, signo, EVFILT_SIGNAL, EV_ADD, 0, 0, 0); + ret = kevent(m_kqueue, &ev, 1, nullptr, 0, nullptr); + assert(ret == 0); +#endif + // If we're using kqueue, the signal needs to be unblocked in order to recieve + // it. If using pselect/ppoll, we need to block it, and later unblock it as a + // part of the system call. + ret = pthread_sigmask(HAVE_SYS_EVENT_H ? SIG_UNBLOCK : SIG_BLOCK, + &new_action.sa_mask, &old_set); + assert(ret == 0 && "pthread_sigmask failed"); + info.was_blocked = sigismember(&old_set, signo); m_signals.insert({signo, info}); - g_signal_flags[signo] = 0; return SignalHandleUP(new SignalHandle(*this, signo)); #endif @@ -331,7 +320,6 @@ #if SIGNAL_POLLING_UNSUPPORTED Error("Signal polling is not supported on this platform."); #else - // We undo the actions of RegisterSignal on a best-effort basis. auto it = m_signals.find(signo); assert(it != m_signals.end()); @@ -340,8 +328,17 @@ sigset_t set; sigemptyset(&set); sigaddset(&set, signo); - pthread_sigmask(it->second.was_blocked ? SIG_BLOCK : SIG_UNBLOCK, &set, - nullptr); + int ret = pthread_sigmask(it->second.was_blocked ? SIG_BLOCK : SIG_UNBLOCK, + &set, nullptr); + assert(ret == 0); + (void)ret; + +#if HAVE_SYS_EVENT_H + struct kevent ev; + EV_SET(&ev, signo, EVFILT_SIGNAL, EV_DELETE, 0, 0, 0); + ret = kevent(m_kqueue, &ev, 1, nullptr, 0, nullptr); + assert(ret == 0); +#endif m_signals.erase(it); #endif @@ -351,32 +348,31 @@ m_terminate_request = false; Error error; - auto impl = RunImpl::Create(*this, error); - if (!impl) - return error; + RunImpl impl(*this); // run until termination or until we run out of things to listen to while (!m_terminate_request && (!m_read_fds.empty() || !m_signals.empty())) { - error = impl->Poll(); + error = impl.Poll(); if (error.Fail()) return error; - impl->ForEachSignal([&](int sig) { - auto it = m_signals.find(sig); - if (it != m_signals.end()) - it->second.callback(*this); // Do the work - }); - if (m_terminate_request) - return Error(); + impl.ProcessEvents(); - impl->ForEachReadFD([&](int fd) { - auto it = m_read_fds.find(fd); - if (it != m_read_fds.end()) - it->second(*this); // Do the work - }); if (m_terminate_request) return Error(); } return Error(); } + +void MainLoop::ProcessSignal(int signo) { + auto it = m_signals.find(signo); + if (it != m_signals.end()) + it->second.callback(*this); // Do the work +} + +void MainLoop::ProcessReadObject(IOObject::WaitableHandle handle) { + auto it = m_read_fds.find(handle); + if (it != m_read_fds.end()) + it->second(*this); // Do the work +} Index: lldb/trunk/unittests/Host/CMakeLists.txt =================================================================== --- lldb/trunk/unittests/Host/CMakeLists.txt +++ lldb/trunk/unittests/Host/CMakeLists.txt @@ -1,6 +1,7 @@ set (FILES FileSpecTest.cpp FileSystemTest.cpp + MainLoopTest.cpp SocketAddressTest.cpp SocketTest.cpp SymbolsTest.cpp Index: lldb/trunk/unittests/Host/MainLoopTest.cpp =================================================================== --- lldb/trunk/unittests/Host/MainLoopTest.cpp +++ lldb/trunk/unittests/Host/MainLoopTest.cpp @@ -0,0 +1,120 @@ +//===-- MainLoopTest.cpp ----------------------------------------*- C++ -*-===// +// +// The LLVM Compiler Infrastructure +// +// This file is distributed under the University of Illinois Open Source +// License. See LICENSE.TXT for details. +// +//===----------------------------------------------------------------------===// + +#include "lldb/Host/MainLoop.h" +#include "lldb/Host/common/TCPSocket.h" +#include "gtest/gtest.h" +#include + +using namespace lldb_private; + +namespace { +class MainLoopTest : public testing::Test { +public: + static void SetUpTestCase() { +#ifdef _MSC_VER + WSADATA data; + ASSERT_EQ(0, WSAStartup(MAKEWORD(2, 2), &data)); +#endif + } + + static void TearDownTestCase() { +#ifdef _MSC_VER + ASSERT_EQ(0, WSACleanup()); +#endif + } + + void SetUp() override { + bool child_processes_inherit = false; + Error error; + std::unique_ptr listen_socket_up( + new TCPSocket(true, child_processes_inherit)); + ASSERT_TRUE(error.Success()); + error = listen_socket_up->Listen("localhost:0", 5); + ASSERT_TRUE(error.Success()); + + Socket *accept_socket; + std::future accept_error = std::async(std::launch::async, [&] { + return listen_socket_up->Accept(accept_socket); + }); + + std::unique_ptr connect_socket_up( + new TCPSocket(true, child_processes_inherit)); + error = connect_socket_up->Connect( + llvm::formatv("localhost:{0}", listen_socket_up->GetLocalPortNumber()) + .str()); + ASSERT_TRUE(error.Success()); + ASSERT_TRUE(accept_error.get().Success()); + + callback_count = 0; + socketpair[0] = std::move(connect_socket_up); + socketpair[1].reset(accept_socket); + } + + void TearDown() override { + socketpair[0].reset(); + socketpair[1].reset(); + } + +protected: + MainLoop::Callback make_callback() { + return [&](MainLoopBase &loop) { + ++callback_count; + loop.RequestTermination(); + }; + } + std::shared_ptr socketpair[2]; + unsigned callback_count; +}; +} // namespace + +TEST_F(MainLoopTest, ReadObject) { + char X = 'X'; + size_t len = sizeof(X); + ASSERT_TRUE(socketpair[0]->Write(&X, len).Success()); + + MainLoop loop; + + Error error; + auto handle = loop.RegisterReadObject(socketpair[1], make_callback(), error); + ASSERT_TRUE(error.Success()); + ASSERT_TRUE(handle); + ASSERT_TRUE(loop.Run().Success()); + ASSERT_EQ(1u, callback_count); +} + +TEST_F(MainLoopTest, TerminatesImmediately) { + char X = 'X'; + size_t len = sizeof(X); + ASSERT_TRUE(socketpair[0]->Write(&X, len).Success()); + ASSERT_TRUE(socketpair[1]->Write(&X, len).Success()); + + MainLoop loop; + Error error; + auto handle0 = loop.RegisterReadObject(socketpair[0], make_callback(), error); + ASSERT_TRUE(error.Success()); + auto handle1 = loop.RegisterReadObject(socketpair[1], make_callback(), error); + ASSERT_TRUE(error.Success()); + + ASSERT_TRUE(loop.Run().Success()); + ASSERT_EQ(1u, callback_count); +} + +#ifdef LLVM_ON_UNIX +TEST_F(MainLoopTest, Signal) { + MainLoop loop; + Error error; + + auto handle = loop.RegisterSignal(SIGUSR1, make_callback(), error); + ASSERT_TRUE(error.Success()); + kill(getpid(), SIGUSR1); + ASSERT_TRUE(loop.Run().Success()); + ASSERT_EQ(1u, callback_count); +} +#endif