diff --git a/src/anbox/cmds/session_manager.cpp b/src/anbox/cmds/session_manager.cpp index dcf0592..a408fdd 100644 --- a/src/anbox/cmds/session_manager.cpp +++ b/src/anbox/cmds/session_manager.cpp @@ -195,7 +195,7 @@ anbox::cmds::SessionManager::SessionManager(const BusFactory &bus_factory) {"/dev/fuse", "/dev/fuse"}, }; - dispatcher->dispatch([&]() { container.start_container(container_configuration); }); + dispatcher->dispatch([&]() { container.start(container_configuration); }); auto bus = bus_factory_(); bus->install_executor(core::dbus::asio::make_executor(bus, rt->service())); @@ -204,6 +204,11 @@ anbox::cmds::SessionManager::SessionManager(const BusFactory &bus_factory) rt->start(); trap->run(); + + // Stop the container which should close all open connections we have on + // our side and should terminate all services. + container.stop(); + rt->stop(); return EXIT_SUCCESS; diff --git a/src/anbox/container/client.cpp b/src/anbox/container/client.cpp index f2c3a40..bfbbf3e 100644 --- a/src/anbox/container/client.cpp +++ b/src/anbox/container/client.cpp @@ -42,7 +42,7 @@ Client::Client(const std::shared_ptr &rt) Client::~Client() {} -void Client::start_container(const Configuration &configuration) { +void Client::start(const Configuration &configuration) { try { management_api_->start_container(configuration); } catch (const std::exception &e) { @@ -52,6 +52,10 @@ void Client::start_container(const Configuration &configuration) { } } +void Client::stop() { + management_api_->stop_container(); +} + void Client::register_terminate_handler(const TerminateCallback &callback) { terminate_callback_ = callback; } diff --git a/src/anbox/container/client.h b/src/anbox/container/client.h index bccf4d1..fdd9d18 100644 --- a/src/anbox/container/client.h +++ b/src/anbox/container/client.h @@ -39,7 +39,8 @@ class Client { Client(const std::shared_ptr &rt); ~Client(); - void start_container(const Configuration &configuration); + void start(const Configuration &configuration); + void stop(); void register_terminate_handler(const TerminateCallback &callback); diff --git a/src/anbox/container/management_api_message_processor.cpp b/src/anbox/container/management_api_message_processor.cpp index 51254cb..1ef4e76 100644 --- a/src/anbox/container/management_api_message_processor.cpp +++ b/src/anbox/container/management_api_message_processor.cpp @@ -32,11 +32,11 @@ ManagementApiMessageProcessor::ManagementApiMessageProcessor( ManagementApiMessageProcessor::~ManagementApiMessageProcessor() {} -void ManagementApiMessageProcessor::dispatch( - rpc::Invocation const &invocation) { +void ManagementApiMessageProcessor::dispatch(rpc::Invocation const &invocation) { if (invocation.method_name() == "start_container") - invoke(this, server_.get(), &ManagementApiSkeleton::start_container, - invocation); + invoke(this, server_.get(), &ManagementApiSkeleton::start_container, invocation); + else if (invocation.method_name() == "stop_container") + invoke(this, server_.get(), &ManagementApiSkeleton::stop_container, invocation); } void ManagementApiMessageProcessor::process_event_sequence( diff --git a/src/anbox/container/management_api_skeleton.cpp b/src/anbox/container/management_api_skeleton.cpp index 6c669c1..bc2fa43 100644 --- a/src/anbox/container/management_api_skeleton.cpp +++ b/src/anbox/container/management_api_skeleton.cpp @@ -55,8 +55,28 @@ void ManagementApiSkeleton::start_container( try { container_->start(container_configuration); } catch (std::exception &err) { - response->set_error( - utils::string_format("Failed to start container: %s", err.what())); + response->set_error(utils::string_format("Failed to start container: %s", err.what())); + } + + done->Run(); +} + +void ManagementApiSkeleton::stop_container( + anbox::protobuf::container::StopContainer const *request, + anbox::protobuf::rpc::Void *response, google::protobuf::Closure *done) { + + (void)request; + + if (container_->state() != Container::State::running) { + response->set_error("Container is not running"); + done->Run(); + return; + } + + try { + container_->stop(); + } catch (std::exception &err) { + response->set_error(utils::string_format("Failed to stop container: %s", err.what())); } done->Run(); diff --git a/src/anbox/container/management_api_skeleton.h b/src/anbox/container/management_api_skeleton.h index 6296151..0cfe99e 100644 --- a/src/anbox/container/management_api_skeleton.h +++ b/src/anbox/container/management_api_skeleton.h @@ -33,6 +33,7 @@ class Void; } // namespace rpc namespace container { class StartContainer; +class StopContainer; } // namespace container } // namespace protobuf namespace rpc { @@ -51,6 +52,10 @@ class ManagementApiSkeleton { anbox::protobuf::container::StartContainer const *request, anbox::protobuf::rpc::Void *response, google::protobuf::Closure *done); + void stop_container( + anbox::protobuf::container::StopContainer const *request, + anbox::protobuf::rpc::Void *response, google::protobuf::Closure *done); + private: std::shared_ptr pending_calls_; std::shared_ptr container_; diff --git a/src/anbox/container/management_api_stub.cpp b/src/anbox/container/management_api_stub.cpp index 87ffd50..1806daf 100644 --- a/src/anbox/container/management_api_stub.cpp +++ b/src/anbox/container/management_api_stub.cpp @@ -36,7 +36,7 @@ void ManagementApiStub::start_container(const Configuration &configuration) { protobuf::container::StartContainer message; auto message_configuration = new protobuf::container::Configuration; - for (const auto item : configuration.bind_mounts) { + for (const auto &item : configuration.bind_mounts) { auto bind_mount_message = message_configuration->add_bind_mounts(); bind_mount_message->set_source(item.first); bind_mount_message->set_target(item.second); @@ -46,24 +46,43 @@ void ManagementApiStub::start_container(const Configuration &configuration) { { std::lock_guard lock(mutex_); - start_wait_handle_.expect_result(); + c->wh.expect_result(); } - channel_->call_method( - "start_container", &message, c->response.get(), - google::protobuf::NewCallback(this, &ManagementApiStub::container_started, - c.get())); + channel_->call_method("start_container", &message, c->response.get(), + google::protobuf::NewCallback(this, &ManagementApiStub::container_started, c.get())); - start_wait_handle_.wait_for_all(); + c->wh.wait_for_all(); if (c->response->has_error()) throw std::runtime_error(c->response->error()); } -void ManagementApiStub::container_started( - Request *request) { - (void)request; - DEBUG(""); - start_wait_handle_.result_received(); +void ManagementApiStub::container_started(Request *request) { + request->wh.result_received(); } + +void ManagementApiStub::stop_container() { + auto c = std::make_shared>(); + + protobuf::container::StopContainer message; + message.set_force(false); + + { + std::lock_guard lock(mutex_); + c->wh.expect_result(); + } + + channel_->call_method("stop_container", &message, c->response.get(), + google::protobuf::NewCallback(this, &ManagementApiStub::container_stopped, c.get())); + + c->wh.wait_for_all(); + + if (c->response->has_error()) throw std::runtime_error(c->response->error()); +} + +void ManagementApiStub::container_stopped(Request *request) { + request->wh.result_received(); +} + } // namespace container } // namespace anbox diff --git a/src/anbox/container/management_api_stub.h b/src/anbox/container/management_api_stub.h index ac7fd6c..11ebeeb 100644 --- a/src/anbox/container/management_api_stub.h +++ b/src/anbox/container/management_api_stub.h @@ -40,6 +40,7 @@ class ManagementApiStub : public DoNotCopyOrMove { ~ManagementApiStub(); void start_container(const Configuration &configuration); + void stop_container(); private: template @@ -47,13 +48,14 @@ class ManagementApiStub : public DoNotCopyOrMove { Request() : response(std::make_shared()), success(true) {} std::shared_ptr response; bool success; + common::WaitHandle wh; }; void container_started(Request *request); + void container_stopped(Request *request); mutable std::mutex mutex_; std::shared_ptr channel_; - common::WaitHandle start_wait_handle_; }; } // namespace container } // namespace anbox diff --git a/src/anbox/container/service.cpp b/src/anbox/container/service.cpp index 667d1bb..15d90e2 100644 --- a/src/anbox/container/service.cpp +++ b/src/anbox/container/service.cpp @@ -33,10 +33,12 @@ namespace container { std::shared_ptr Service::create(const std::shared_ptr &rt, bool privileged) { auto sp = std::shared_ptr(new Service(rt, privileged)); - auto delegate_connector = std::make_shared< - network::DelegateConnectionCreator>( - [sp](std::shared_ptr const - &socket) { sp->new_client(socket); }); + auto wp = std::weak_ptr(sp); + auto delegate_connector = std::make_shared>( + [wp](std::shared_ptr const &socket) { + if (auto service = wp.lock()) + service->new_client(socket); + }); const auto container_socket_path = SystemConfiguration::instance().container_socket_path(); sp->connector_ = std::make_shared(container_socket_path, rt, delegate_connector); @@ -56,7 +58,9 @@ Service::Service(const std::shared_ptr &rt, bool privileged) privileged_(privileged) { } -Service::~Service() {} +Service::~Service() { + connections_->clear(); +} int Service::next_id() { return next_connection_id_++; } diff --git a/src/anbox/graphics/emugl/RenderThread.cpp b/src/anbox/graphics/emugl/RenderThread.cpp index 750d6de..9dbc569 100644 --- a/src/anbox/graphics/emugl/RenderThread.cpp +++ b/src/anbox/graphics/emugl/RenderThread.cpp @@ -33,7 +33,9 @@ RenderThread::RenderThread(const std::shared_ptr &renderer, IOStream *stream, emugl::Mutex *lock) : emugl::Thread(), renderer_(renderer), m_lock(lock), m_stream(stream) {} -RenderThread::~RenderThread() {} +RenderThread::~RenderThread() { + forceStop(); +} RenderThread *RenderThread::create(const std::shared_ptr &renderer, IOStream *stream, emugl::Mutex *lock) { return new RenderThread(renderer, stream, lock); diff --git a/src/anbox/network/base_socket_messenger.cpp b/src/anbox/network/base_socket_messenger.cpp index b5aae10..fb093f3 100644 --- a/src/anbox/network/base_socket_messenger.cpp +++ b/src/anbox/network/base_socket_messenger.cpp @@ -80,7 +80,7 @@ ssize_t BaseSocketMessenger::send_raw(char const* data, std::copy(data, data + length, whole_message.data()); std::unique_lock lg(message_lock); - return ::send(socket_fd, data, length, 0); + return ::send(socket_fd, data, length, MSG_NOSIGNAL); } template @@ -95,8 +95,8 @@ void BaseSocketMessenger::send(char const* data, ba::write(*socket, ba::buffer(whole_message.data(), whole_message.size()), boost::asio::transfer_all()); } catch (const boost::system::system_error& err) { - DEBUG("Got error: %s", err.what()); if (err.code() == boost::asio::error::try_again) continue; + throw; } break; } diff --git a/src/anbox/network/published_socket_connector.cpp b/src/anbox/network/published_socket_connector.cpp index b1ae7f6..5e1a9b6 100644 --- a/src/anbox/network/published_socket_connector.cpp +++ b/src/anbox/network/published_socket_connector.cpp @@ -18,6 +18,7 @@ #include "anbox/network/published_socket_connector.h" #include "anbox/network/connection_context.h" #include "anbox/network/socket_helper.h" +#include "anbox/logger.h" namespace anbox { namespace network { @@ -35,8 +36,7 @@ PublishedSocketConnector::PublishedSocketConnector( PublishedSocketConnector::~PublishedSocketConnector() {} void PublishedSocketConnector::start_accept() { - auto socket = std::make_shared( - runtime_->service()); + auto socket = std::make_shared(runtime_->service()); acceptor_.async_accept(*socket, [this, socket](boost::system::error_code const& err) { @@ -44,10 +44,13 @@ void PublishedSocketConnector::start_accept() { }); } -void PublishedSocketConnector::on_new_connection( - std::shared_ptr const& socket, - boost::system::error_code const& err) { - if (!err) connection_creator_->create_connection_for(socket); +void PublishedSocketConnector::on_new_connection(std::shared_ptr const& socket, + boost::system::error_code const& err) { + if (!err) + connection_creator_->create_connection_for(socket); + + if (err.value() == boost::asio::error::operation_aborted) + return; start_accept(); } diff --git a/src/anbox/network/socket_connection.cpp b/src/anbox/network/socket_connection.cpp index d2c1014..02c079d 100644 --- a/src/anbox/network/socket_connection.cpp +++ b/src/anbox/network/socket_connection.cpp @@ -53,16 +53,13 @@ void SocketConnection::send(char const* data, size_t length) { } void SocketConnection::read_next_message() { - auto callback = std::bind(&SocketConnection::on_read_size, this, - std::placeholders::_1, std::placeholders::_2); + auto callback = std::bind(&SocketConnection::on_read_size, this, std::placeholders::_1, std::placeholders::_2); message_receiver_->async_receive_msg(callback, ba::buffer(buffer_)); } -void SocketConnection::on_read_size(const boost::system::error_code& error, - std::size_t bytes_read) { +void SocketConnection::on_read_size(const boost::system::error_code& error, std::size_t bytes_read) { if (error) { - if (connections_) connections_->remove(id()); - + connections_->remove(id()); return; } @@ -71,8 +68,8 @@ void SocketConnection::on_read_size(const boost::system::error_code& error, if (processor_->process_data(data)) read_next_message(); - else if (connections_) - connections_->remove(id()); + else + connections_->remove(id()); } } // namespace anbox } // namespace network diff --git a/src/anbox/network/socket_connection.h b/src/anbox/network/socket_connection.h index a0618e0..cf23c3b 100644 --- a/src/anbox/network/socket_connection.h +++ b/src/anbox/network/socket_connection.h @@ -42,10 +42,6 @@ class SocketConnection { void set_name(const std::string& name) { name_ = name; } - std::shared_ptr message_sender() const { - return message_sender_; - } - int id() const { return id_; } void send(char const* data, size_t length); @@ -55,10 +51,10 @@ class SocketConnection { void on_read_size(const boost::system::error_code& ec, std::size_t bytes_read); - std::shared_ptr message_receiver_; - std::shared_ptr message_sender_; + std::shared_ptr const message_receiver_; + std::shared_ptr const message_sender_; int id_; - std::shared_ptr> connections_; + std::shared_ptr> const connections_; std::shared_ptr processor_; std::array buffer_; std::string name_; diff --git a/src/anbox/network/tcp_socket_connector.cpp b/src/anbox/network/tcp_socket_connector.cpp index 36ee840..00d33c2 100644 --- a/src/anbox/network/tcp_socket_connector.cpp +++ b/src/anbox/network/tcp_socket_connector.cpp @@ -53,12 +53,10 @@ void TcpSocketConnector::on_new_connection( case boost::system::errc::success: connection_creator_->create_connection_for(socket); break; - case boost::system::errc::operation_canceled: + default: // Socket was closed so don't listen for any further incoming // connection attempts. return; - default: - break; } start_accept(); diff --git a/src/anbox/network/tcp_socket_messenger.h b/src/anbox/network/tcp_socket_messenger.h index 2712b52..e97e597 100644 --- a/src/anbox/network/tcp_socket_messenger.h +++ b/src/anbox/network/tcp_socket_messenger.h @@ -33,7 +33,7 @@ class TcpSocketMessenger : public BaseSocketMessenger { std::shared_ptr const &socket); ~TcpSocketMessenger(); - unsigned short local_port() const; + unsigned short local_port() const override; private: unsigned short local_port_; diff --git a/src/anbox/protobuf/anbox_container.proto b/src/anbox/protobuf/anbox_container.proto index 07f8e0c..05cbee9 100644 --- a/src/anbox/protobuf/anbox_container.proto +++ b/src/anbox/protobuf/anbox_container.proto @@ -13,3 +13,7 @@ message Configuration { message StartContainer { required Configuration configuration = 1; } + +message StopContainer { + optional bool force = 1; +} diff --git a/src/anbox/qemu/adb_message_processor.cpp b/src/anbox/qemu/adb_message_processor.cpp index 921b921..f41b9de 100644 --- a/src/anbox/qemu/adb_message_processor.cpp +++ b/src/anbox/qemu/adb_message_processor.cpp @@ -49,6 +49,7 @@ AdbMessageProcessor::AdbMessageProcessor( host_notify_timer_(rt->service()) {} AdbMessageProcessor::~AdbMessageProcessor() { + state_ = closed_by_host; host_connector_.reset(); active_instance.unlock(); } @@ -62,6 +63,11 @@ void AdbMessageProcessor::advance_state() { // one is established but will not use it until the active one is closed. active_instance.lock(); + if (state_ == closed_by_host) { + host_connector_.reset(); + return; + } + wait_for_host_connection(); break; case waiting_for_host_connection: @@ -122,9 +128,7 @@ void AdbMessageProcessor::wait_for_host_connection() { } } -void AdbMessageProcessor::on_host_connection( - std::shared_ptr< - boost::asio::basic_stream_socket> const &socket) { +void AdbMessageProcessor::on_host_connection(std::shared_ptr> const &socket) { host_messenger_ = std::make_shared(socket); // set_no_delay() reduces the latency of sending data, at the cost @@ -136,16 +140,14 @@ void AdbMessageProcessor::on_host_connection( // Let adb inside the container know that we have a connection to // the adb host instance - messenger_->send(reinterpret_cast(ok_command.data()), - ok_command.size()); + messenger_->send(reinterpret_cast(ok_command.data()), ok_command.size()); state_ = waiting_for_guest_start_command; expected_command_ = start_command; } void AdbMessageProcessor::read_next_host_message() { - auto callback = - std::bind(&AdbMessageProcessor::on_host_read_size, this, _1, _2); + auto callback = std::bind(&AdbMessageProcessor::on_host_read_size, this, _1, _2); host_messenger_->async_receive_msg(callback, boost::asio::buffer(host_buffer_)); } @@ -153,15 +155,11 @@ void AdbMessageProcessor::read_next_host_message() { void AdbMessageProcessor::on_host_read_size( const boost::system::error_code &error, std::size_t bytes_read) { if (error) { - // If messenger is still alive then close the connection which will - // trigger the terminate of our processor instance too. - if (messenger_) messenger_->close(); - - return; + state_ = closed_by_host; + BOOST_THROW_EXCEPTION(std::runtime_error(error.message())); } - messenger_->send(reinterpret_cast(host_buffer_.data()), - bytes_read); + messenger_->send(reinterpret_cast(host_buffer_.data()), bytes_read); read_next_host_message(); } diff --git a/src/anbox/qemu/adb_message_processor.h b/src/anbox/qemu/adb_message_processor.h index 88f0cd6..b98fa7a 100644 --- a/src/anbox/qemu/adb_message_processor.h +++ b/src/anbox/qemu/adb_message_processor.h @@ -60,7 +60,7 @@ class AdbMessageProcessor : public network::MessageProcessor { std::shared_ptr runtime_; State state_ = waiting_for_guest_accept_command; std::string expected_command_; - std::shared_ptr messenger_; + std::shared_ptr const messenger_; std::vector buffer_; std::shared_ptr host_connector_; std::shared_ptr host_messenger_; diff --git a/src/anbox/qemu/pipe_connection_creator.cpp b/src/anbox/qemu/pipe_connection_creator.cpp index b2005c3..78bb258 100644 --- a/src/anbox/qemu/pipe_connection_creator.cpp +++ b/src/anbox/qemu/pipe_connection_creator.cpp @@ -73,13 +73,14 @@ PipeConnectionCreator::PipeConnectionCreator(const std::shared_ptr &re std::make_shared>()) { } -PipeConnectionCreator::~PipeConnectionCreator() {} +PipeConnectionCreator::~PipeConnectionCreator() { + connections_->clear(); +} void PipeConnectionCreator::create_connection_for( std::shared_ptr const &socket) { - auto const messenger = - std::make_shared(socket); + auto const messenger = std::make_shared(socket); const auto type = identify_client(messenger); auto const processor = create_processor(type, messenger); if (!processor) diff --git a/src/anbox/qemu/pipe_connection_creator.h b/src/anbox/qemu/pipe_connection_creator.h index cbfadd3..2dff9b9 100644 --- a/src/anbox/qemu/pipe_connection_creator.h +++ b/src/anbox/qemu/pipe_connection_creator.h @@ -68,8 +68,7 @@ class PipeConnectionCreator std::shared_ptr renderer_; std::shared_ptr runtime_; std::atomic next_connection_id_; - std::shared_ptr> const - connections_; + std::shared_ptr> const connections_; }; } // namespace qemu } // namespace anbox diff --git a/src/anbox/runtime.cpp b/src/anbox/runtime.cpp index 896a0c1..1be2d1c 100644 --- a/src/anbox/runtime.cpp +++ b/src/anbox/runtime.cpp @@ -38,10 +38,9 @@ void exception_safe_run(boost::asio::io_service& service) { // a service::work instance). break; } catch (const std::exception& e) { - std::cerr << e.what() << std::endl; + ERROR("%s", e.what()); } catch (...) { - std::cerr - << "Unknown exception caught while executing boost::asio::io_service"; + ERROR("Unknown exception caught while executing boost::asio::io_service"); } } } @@ -74,7 +73,9 @@ void Runtime::start() { void Runtime::stop() { service_.stop(); - for (auto& worker : workers_) pthread_kill(worker.native_handle(), SIGTERM); + for (auto& worker : workers_) + if (worker.joinable()) + worker.join(); } std::function)> Runtime::to_dispatcher_functional() {