From 9594ebf72efdfff1a88aaf2206a90836ee9b9130 Mon Sep 17 00:00:00 2001 From: Mara Bos Date: Tue, 13 Nov 2018 20:31:22 +0100 Subject: [PATCH] Use unix sockets instead of pipes for posix daemon. (#10766) Unlike pipes, unix sockets provide bi-directional communication with each connected client. - No need to generate a unique uuid per client anymore. - The client doesn't have to create its own pipe anymore. - Since there is no risk of multiple client's writes getting mixed up, messages don't need to fit in a single write anymore, removing the limit on command length. - Since the server can detect a connection closing, the client no longer needs to handle signals. When the client is killed, the connection is automatically closed, which will cause the server to kill the related px4 thread. Since this does not rely on handling signals and the client sending an additional message, this is much more reliable. - Client is no longer a singleton. - The protocol is simplified. Standard output is directly written to the socket back to the client, without wrapping it in any protocol message. - Because of the simple protocol, one could now even use netcat to run a px4 command: $ echo hello | netcat -UN /tmp/px4-sock-0 Also removes a few race conditions. --- platforms/posix/src/main.cpp | 2 - platforms/posix/src/px4_daemon/CMakeLists.txt | 2 +- platforms/posix/src/px4_daemon/client.cpp | 322 ++++------------ platforms/posix/src/px4_daemon/client.h | 46 +-- platforms/posix/src/px4_daemon/server.cpp | 361 +++++++----------- platforms/posix/src/px4_daemon/server.h | 48 +-- platforms/posix/src/px4_daemon/server_io.cpp | 30 +- .../{pipe_protocol.cpp => sock_protocol.cpp} | 38 +- .../{pipe_protocol.h => sock_protocol.h} | 59 +-- 9 files changed, 270 insertions(+), 638 deletions(-) rename platforms/posix/src/px4_daemon/{pipe_protocol.cpp => sock_protocol.cpp} (62%) rename platforms/posix/src/px4_daemon/{pipe_protocol.h => sock_protocol.h} (52%) diff --git a/platforms/posix/src/main.cpp b/platforms/posix/src/main.cpp index 91739014cb..7cf1cae21a 100644 --- a/platforms/posix/src/main.cpp +++ b/platforms/posix/src/main.cpp @@ -173,8 +173,6 @@ int main(int argc, char **argv) argv[0] += path_length + strlen(prefix); px4_daemon::Client client(instance); - client.generate_uuid(); - client.register_sig_handler(); return client.process_args(argc, (const char **)argv); } else { diff --git a/platforms/posix/src/px4_daemon/CMakeLists.txt b/platforms/posix/src/px4_daemon/CMakeLists.txt index 9704af02c1..e6cf9fcc75 100644 --- a/platforms/posix/src/px4_daemon/CMakeLists.txt +++ b/platforms/posix/src/px4_daemon/CMakeLists.txt @@ -37,6 +37,6 @@ px4_add_library(px4_daemon client.cpp server.cpp server_io.cpp - pipe_protocol.cpp + sock_protocol.cpp ) diff --git a/platforms/posix/src/px4_daemon/client.cpp b/platforms/posix/src/px4_daemon/client.cpp index 5f76cf1a6b..194875aabe 100644 --- a/platforms/posix/src/px4_daemon/client.cpp +++ b/platforms/posix/src/px4_daemon/client.cpp @@ -1,6 +1,6 @@ /**************************************************************************** * - * Copyright (C) 2016 PX4 Development Team. All rights reserved. + * Copyright (C) 2016, 2018 PX4 Development Team. All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions @@ -35,13 +35,15 @@ * * @author Julian Oes * @author Beat Küng + * @author Mara Bos */ #include #include #include -#include +#include #include +#include #include #include @@ -53,60 +55,33 @@ namespace px4_daemon { - -namespace client -{ -static Client *_instance; -} - Client::Client(int instance_id) : - _uuid(0), - _client_send_pipe_fd(-1), + _fd(-1), _instance_id(instance_id) -{ - client::_instance = this; -} - -Client::~Client() -{ - client::_instance = nullptr; -} - -int -Client::generate_uuid() -{ - int rand_fd = open("/dev/urandom", O_RDONLY); - - if (rand_fd < 0) { - PX4_ERR("open urandom"); - return rand_fd; - } - - int ret = 0; - - int rand_read = read(rand_fd, &_uuid, sizeof(_uuid)); - - if (rand_read != sizeof(_uuid)) { - PX4_ERR("rand read fail"); - ret = -errno; - } - - close(rand_fd); - return ret; -} +{} int Client::process_args(const int argc, const char **argv) { - // Prepare return pipe first to avoid a race. - int ret = _prepare_recv_pipe(); + std::string sock_path = get_socket_path(_instance_id); - if (ret != 0) { - PX4_ERR("Could not prepare recv pipe"); - return -2; + _fd = socket(AF_UNIX, SOCK_STREAM, 0); + + if (_fd < 0) { + PX4_ERR("error creating socket"); + return -1; } - ret = _send_cmds(argc, argv); + sockaddr_un addr = {}; + addr.sun_family = AF_UNIX; + strncpy(addr.sun_path, sock_path.c_str(), sizeof(addr.sun_path) - 1); + + if (connect(_fd, (sockaddr *)&addr, sizeof(addr)) < 0) { + PX4_ERR("error connecting to socket"); + return -1; + } + + int ret = _send_cmds(argc, argv); if (ret != 0) { PX4_ERR("Could not send commands"); @@ -116,231 +91,102 @@ Client::process_args(const int argc, const char **argv) return _listen(); } - -int -Client::_prepare_recv_pipe() -{ - int ret = get_client_recv_pipe_path(_uuid, _recv_pipe_path, sizeof(_recv_pipe_path)); - - if (ret < 0) { - PX4_ERR("failed to assemble path"); - return ret; - } - - ret = mkfifo(_recv_pipe_path, 0666); - - if (ret < 0) { - PX4_ERR("pipe %s already exists, errno: %d, %s", _recv_pipe_path, errno, strerror(errno)); - return ret; - } - - return 0; -} - int Client::_send_cmds(const int argc, const char **argv) { - // Send the command to server. - client_send_packet_s packet; - packet.header.msg_id = client_send_packet_s::message_header_s::e_msg_id::EXECUTE; - packet.header.client_uuid = _uuid; - packet.payload.execute_msg.is_atty = isatty(STDOUT_FILENO); - - // Concat arguments to send them. std::string cmd_buf; for (int i = 0; i < argc; ++i) { cmd_buf += argv[i]; if (i + 1 != argc) { + // TODO: Use '\0' as argument separator (and parse this server-side as well), + // so (quoted) whitespace within arguments doesn't get lost. cmd_buf += " "; } } - if (cmd_buf.size() >= sizeof(packet.payload.execute_msg.cmd)) { - PX4_ERR("commmand too long"); + // Last byte is 'isatty'. + cmd_buf.push_back(isatty(STDOUT_FILENO)); + + size_t n = cmd_buf.size(); + const char *buf = cmd_buf.data(); + + while (n > 0) { + int n_sent = write(_fd, buf, n); + + if (n_sent < 0) { + PX4_ERR("write() failed: %s", strerror(errno)); + return -1; + } + + n -= n_sent; + buf += n_sent; + } + + // Let the server know we're done writing. + if (shutdown(_fd, SHUT_WR) < 0) { + PX4_ERR("shutdown() failed: %s", strerror(errno)); return -1; } - strcpy((char *)packet.payload.execute_msg.cmd, cmd_buf.c_str()); - - // The size is +1 because we want to include the null termination. - packet.header.payload_length = cmd_buf.size() + 1 + sizeof(packet.payload.execute_msg.is_atty); - - _client_send_pipe_fd = open(get_client_send_pipe_path(_instance_id).c_str(), O_WRONLY); - - if (_client_send_pipe_fd < 0) { - PX4_ERR("pipe open fail (%i)", errno); - return _client_send_pipe_fd; - } - - int bytes_to_send = get_client_send_packet_length(&packet); - int bytes_sent = write(_client_send_pipe_fd, &packet, bytes_to_send); - - if (bytes_sent != bytes_to_send) { - PX4_ERR("write fail (%i)", errno); - return bytes_sent; - } - return 0; } int Client::_listen() { - int client_recv_pipe_fd = open(_recv_pipe_path, O_RDONLY); + char buffer[1024]; + int n_buffer_used = 0; - if (client_recv_pipe_fd < 0) { - PX4_ERR("open failed, errno: %d, %s", errno, strerror(errno)); - } + // The response ends in {0, retval}. So when we detect a 0, or a 0 followed + // by another byte, we don't output it yet, until we know whether it was + // the end of the stream or not. + while (true) { + int n_read = read(_fd, buffer + n_buffer_used, sizeof buffer - n_buffer_used); - bool exit_loop = false; - int exit_arg = 0; + if (n_read < 0) { + PX4_ERR("unable to read from socket"); + return -1; - while (!exit_loop) { + } else if (n_read == 0) { + if (n_buffer_used == 2) { + return buffer[1]; - // We only read as much as we need, otherwise we might get out of - // sync with packets. - client_recv_packet_s packet_recv; - int bytes_read = read(client_recv_pipe_fd, &packet_recv, sizeof(client_recv_packet_s::header)); - - if (bytes_read > 0) { - - // Using the header we can determine how big the payload is. - int payload_to_read = sizeof(packet_recv) - - sizeof(packet_recv.header) - - sizeof(packet_recv.payload) - + packet_recv.header.payload_length; - - // Again, we only read as much as we need because otherwise we need - // hold a buffer and parse it. - bytes_read = read(client_recv_pipe_fd, ((uint8_t *)&packet_recv) + bytes_read, payload_to_read); - - if (bytes_read > 0) { - - int retval = 0; - bool should_exit = false; - - int parse_ret = _parse_client_recv_packet(packet_recv, retval, should_exit); - - if (parse_ret != 0) { - PX4_ERR("retval could not be parsed"); - exit_arg = -1; - - } else { - exit_arg = retval; - } - - exit_loop = should_exit; - - } else if (bytes_read == 0) { - exit_arg = 0; - exit_loop = true; + } else { + // Missing return value at end of stream. Stream was abruptly ended. + return -1; } - } else if (bytes_read == 0) { - // 0 means the pipe has been closed by all clients. - exit_arg = 0; - exit_loop = true; + } else { + n_read += n_buffer_used; + + if (n_read >= 2 && buffer[n_read - 2] == 0) { + // If the buffer ends in {0, retval}, keep it. + fwrite(buffer, n_read - 2, 1, stdout); + buffer[0] = 0; + buffer[1] = buffer[n_read - 1]; + n_buffer_used = 2; + + } else if (n_read >= 1 && buffer[n_read - 1] == 0) { + // If the buffer ends in a 0-byte, keep it. + fwrite(buffer, n_read - 1, 1, stdout); + buffer[0] = 0; + n_buffer_used = 1; + + } else { + fwrite(buffer, n_read, 1, stdout); + n_buffer_used = 0; + } } } - - close(_client_send_pipe_fd); - return exit_arg; } -int -Client::_parse_client_recv_packet(const client_recv_packet_s &packet, int &retval, bool &should_exit) +Client::~Client() { - switch (packet.header.msg_id) { - case client_recv_packet_s::message_header_s::e_msg_id::RETVAL: - - should_exit = true; - return _retval_cmd_packet(packet, retval); - - case client_recv_packet_s::message_header_s::e_msg_id::STDOUT: - - should_exit = false; - return _stdout_msg_packet(packet); - - default: - should_exit = true; - PX4_ERR("recv msg_id not handled: %d", (int)packet.header.msg_id); - return -1; - } -} - -int -Client::_retval_cmd_packet(const client_recv_packet_s &packet, int &retval) -{ - if (packet.header.payload_length == sizeof(packet.payload.retval_msg.retval)) { - retval = packet.payload.retval_msg.retval; - return 0; - - } else { - PX4_ERR("payload size wrong"); - return -1; - } -} - -int -Client::_stdout_msg_packet(const client_recv_packet_s &packet) -{ - if (packet.header.payload_length <= sizeof(packet.payload.stdout_msg.text)) { - - printf("%s", packet.payload.stdout_msg.text); - - return 0; - - } else { - PX4_ERR("payload size wrong (%i > %zu)", packet.header.payload_length, sizeof(packet.payload.stdout_msg.text)); - return -1; - } -} - -void -Client::register_sig_handler() -{ - // Register handlers for Ctrl+C to kill the thread if something hangs. - struct sigaction sig_int {}; - sig_int.sa_handler = Client::_static_sig_handler; - - // Without the flag SA_RESTART, we can't use open() after Ctrl+C has - // been pressed, and we can't wait for the return value from the - // cancelled command. - sig_int.sa_flags = SA_RESTART; - sigaction(SIGINT, &sig_int, nullptr); - sigaction(SIGTERM, &sig_int, nullptr); -} - -void -Client::_static_sig_handler(int sig_num) -{ - client::_instance->_sig_handler(sig_num); -} - -void -Client::_sig_handler(int sig_num) -{ - client_send_packet_s packet; - packet.header.msg_id = client_send_packet_s::message_header_s::e_msg_id::KILL; - packet.header.client_uuid = _uuid; - packet.payload.kill_msg.cmd_id = sig_num; - packet.header.payload_length = sizeof(packet.payload.kill_msg.cmd_id); - - if (_client_send_pipe_fd < 0) { - PX4_ERR("pipe open fail"); - system_exit(-1); - } - - int bytes_to_send = get_client_send_packet_length(&packet); - int bytes_sent = write(_client_send_pipe_fd, &packet, bytes_to_send); - - if (bytes_sent != bytes_to_send) { - PX4_ERR("write fail"); - system_exit(-1); + if (_fd >= 0) { + close(_fd); } } } // namespace px4_daemon - diff --git a/platforms/posix/src/px4_daemon/client.h b/platforms/posix/src/px4_daemon/client.h index 2e454c4cc2..41f4cb8381 100644 --- a/platforms/posix/src/px4_daemon/client.h +++ b/platforms/posix/src/px4_daemon/client.h @@ -1,6 +1,6 @@ /**************************************************************************** * - * Copyright (C) 2016 PX4 Development Team. All rights reserved. + * Copyright (C) 2016, 2018 PX4 Development Team. All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions @@ -33,21 +33,22 @@ /** * @file client.h * - * The client can write a command to the pipe that is supplied by the server. - * It will then open another pipe with its own UUID encoded and listen for - * stdout of the process that it started and the return value. + * The client can connect and write a command to the socket that is supplied by + * the server. It will then close its half of the connection, and read back the + * stdout stream of the process that it started, followed by its return value. * - * It the client receives a signal (e.g. Ctrl+C) it will catch it and send it - * as a message to the server in order to terminate the thread. + * It the client dies, the connection gets closed automatically and the corresponding + * thread in the server gets cancelled. * * @author Julian Oes * @author Beat Küng + * @author Mara Bos */ #pragma once #include -#include "pipe_protocol.h" +#include "sock_protocol.h" namespace px4_daemon { @@ -57,19 +58,14 @@ class Client { public: Client(int instance_id = 0); + ~Client(); - /** - * Initialize the unique ID of the client. - * - * @return 0 on success. - */ - int generate_uuid(); - - /** - * Make sure to catch signals in order to forward them to the server. - */ - void register_sig_handler(); + Client(Client &&other) : _fd(other._fd), _instance_id(other._instance_id) + { + // Steal the fd from the moved-from client. + other._fd = -1; + } /** * Process the supplied command line arguments and send them to server. @@ -81,22 +77,10 @@ public: int process_args(const int argc, const char **argv); private: - int _prepare_recv_pipe(); int _send_cmds(const int argc, const char **argv); int _listen(); - int _parse_client_recv_packet(const client_recv_packet_s &packet, int &retval, bool &should_exit); - - int _retval_cmd_packet(const client_recv_packet_s &packet, int &retval); - int _stdout_msg_packet(const client_recv_packet_s &packet); - - - static void _static_sig_handler(int sig_num); - void _sig_handler(int sig_num); - - uint64_t _uuid; - int _client_send_pipe_fd; - char _recv_pipe_path[RECV_PIPE_PATH_LEN]; + int _fd; int _instance_id; ///< instance ID for running multiple instances of the px4 server }; diff --git a/platforms/posix/src/px4_daemon/server.cpp b/platforms/posix/src/px4_daemon/server.cpp index 83fd4a0cae..f9f5b56854 100644 --- a/platforms/posix/src/px4_daemon/server.cpp +++ b/platforms/posix/src/px4_daemon/server.cpp @@ -1,6 +1,6 @@ /**************************************************************************** * - * Copyright (C) 2016 PX4 Development Team. All rights reserved. + * Copyright (C) 2016, 2018 PX4 Development Team. All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions @@ -35,29 +35,27 @@ * * @author Julian Oes * @author Beat Küng + * @author Mara Bos */ +#include #include #include +#include #include #include #include #include +#include #include +#include +#include #include #include "pxh.h" #include "server.h" -// In Cygwin opening and closing the same named pipe multiple times within one process doesn't work POSIX compliant. -// As a workaround we open the client send pipe file in read write mode such that we can keep it open all the time. -#if !defined(__PX4_CYGWIN) -#define CLIENT_SEND_PIPE_OFLAGS O_RDONLY -#else -#define CLIENT_SEND_PIPE_OFLAGS O_RDWR -#endif - namespace px4_daemon { @@ -78,22 +76,37 @@ Server::~Server() int Server::start() { - std::string client_send_pipe_path = get_client_send_pipe_path(_instance_id); + std::string sock_path = get_socket_path(_instance_id); - // Delete pipe in case it exists already. - unlink(client_send_pipe_path.c_str()); + // Delete socket in case it exists already. + unlink(sock_path.c_str()); - // Create new pipe to listen to clients. - // This needs to happen before we return from this method, so that the caller can launch clients. - mkfifo(client_send_pipe_path.c_str(), 0666); + _fd = socket(AF_UNIX, SOCK_STREAM, 0); + + if (_fd < 0) { + PX4_ERR("error creating socket"); + return -1; + } + + sockaddr_un addr = {}; + addr.sun_family = AF_UNIX; + strncpy(addr.sun_path, sock_path.c_str(), sizeof(addr.sun_path) - 1); + + if (bind(_fd, (sockaddr *)&addr, sizeof(addr)) < 0) { + PX4_ERR("error binding socket"); + return -1; + } + + if (listen(_fd, 10) < 0) { + PX4_ERR("error listing to socket"); + return -1; + } if (0 != pthread_create(&_server_main_pthread, nullptr, _server_main_trampoline, - nullptr)) { - + this)) { PX4_ERR("error creating client handler thread"); - return -1; } @@ -101,12 +114,9 @@ Server::start() } void * -Server::_server_main_trampoline(void *arg) +Server::_server_main_trampoline(void *self) { - if (_instance) { - _instance->_server_main(arg); - } - + ((Server *)self)->_server_main(); return nullptr; } @@ -116,9 +126,8 @@ void Server::_pthread_key_destructor(void *arg) } void -Server::_server_main(void *arg) +Server::_server_main() { - // Set thread specific pipe to supplied file descriptor. int ret = pthread_key_create(&_key, _pthread_key_destructor); if (ret != 0) { @@ -126,241 +135,151 @@ Server::_server_main(void *arg) return; } - std::string client_send_pipe_path = get_client_send_pipe_path(_instance_id); - int client_send_pipe_fd = open(client_send_pipe_path.c_str(), CLIENT_SEND_PIPE_OFLAGS); + // The list of file descriptors to watch. + std::vector poll_fds; + + // Watch the listening socket for incoming connections. + poll_fds.push_back(pollfd {_fd, POLLIN, 0}); while (true) { + int n_ready = poll(poll_fds.data(), poll_fds.size(), -1); - client_send_packet_s packet; + if (n_ready < 0) { + PX4_ERR("poll() failed: %s", strerror(errno)); + return; + } - // We only read as much as we need, otherwise we might get out of - // sync with packets. - int bytes_read = read(client_send_pipe_fd, &packet, sizeof(client_send_packet_s::header)); + _lock(); - if (bytes_read > 0) { + // Handle any new connections. + if (poll_fds[0].revents & POLLIN) { + --n_ready; + int client = accept(_fd, nullptr, nullptr); - // Using the header we can determine how big the payload is. - int payload_to_read = sizeof(packet) - - sizeof(packet.header) - - sizeof(packet.payload) - + packet.header.payload_length; + if (client == -1) { + PX4_ERR("failed to accept client: %s", strerror(errno)); + _unlock(); + return; + } - // Again, we only read as much as we need because otherwise we need - // hold a buffer and parse it. - bytes_read = read(client_send_pipe_fd, ((uint8_t *)&packet) + bytes_read, payload_to_read); + // Start a new thread to handle the client. + pthread_t *thread = &_fd_to_thread[client]; + ret = pthread_create(thread, nullptr, Server::_handle_client, (void *)(intptr_t)client); - if (bytes_read > 0) { + if (ret != 0) { + PX4_ERR("could not start pthread (%i)", ret); + _fd_to_thread.erase(client); + close(client); - _parse_client_send_packet(packet); + } else { + // We won't join the thread, so detach to automatically release resources at its end + pthread_detach(*thread); + } + // Start listening for the client hanging up. + poll_fds.push_back(pollfd {client, POLLHUP, 0}); + } + + // Handle any closed connections. + for (size_t i = 1; n_ready > 0 && i < poll_fds.size();) { + if (poll_fds[i].revents) { + --n_ready; + auto thread = _fd_to_thread.find(poll_fds[i].fd); + + if (thread != _fd_to_thread.end()) { + // Thread is still running, so we cancel it. + // TODO: use a more graceful exit method to avoid resource leaks + pthread_cancel(thread->second); + _fd_to_thread.erase(thread); + } + + close(poll_fds[i].fd); + poll_fds.erase(poll_fds.begin() + i); + + } else { + ++i; } } - if (bytes_read == 0) { - // 0 means the pipe has been closed by all clients - // and we need to re-open it. - close(client_send_pipe_fd); - client_send_pipe_fd = open(client_send_pipe_path.c_str(), O_RDONLY); + _unlock(); + } + + close(_fd); +} + +void +*Server::_handle_client(void *arg) +{ + int fd = (int)(intptr_t)arg; + + // Read until the end of the incoming stream. + std::string cmd; + + while (true) { + size_t n = cmd.size(); + cmd.resize(n + 1024); + ssize_t n_read = read(fd, &cmd[n], cmd.size() - n); + + if (n_read < 0) { + _cleanup(fd); + return nullptr; + } + + cmd.resize(n + n_read); + + if (n_read == 0) { + break; } } - close(client_send_pipe_fd); -} - -void -Server::_parse_client_send_packet(const client_send_packet_s &packet) -{ - switch (packet.header.msg_id) { - case client_send_packet_s::message_header_s::e_msg_id::EXECUTE: - _execute_cmd_packet(packet); - break; - - case client_send_packet_s::message_header_s::e_msg_id::KILL: - _kill_cmd_packet(packet); - break; - - default: - PX4_ERR("send msg_id not handled"); - break; - } -} - -void -Server::_execute_cmd_packet(const client_send_packet_s &packet) -{ - if (packet.header.payload_length == 0) { - PX4_ERR("command length 0"); - return; + if (cmd.size() < 2) { + _cleanup(fd); + return nullptr; } - // We open the client's specific pipe to write the return value and stdout back to. - // The pipe's path is created knowing the UUID of the client. - char path[RECV_PIPE_PATH_LEN]; - int ret = get_client_recv_pipe_path(packet.header.client_uuid, path, RECV_PIPE_PATH_LEN); - - if (ret < 0) { - PX4_ERR("failed to assemble path"); - return; - } - - int pipe_fd = open(path, O_WRONLY); - - if (pipe_fd < 0) { - PX4_ERR("pipe open fail"); - return; - } - - // To execute a command we start a new thread. - pthread_t new_pthread; - - // We need to copy everything that the new thread needs because we will go - // out of scope. - RunCmdArgs *args = new RunCmdArgs; - strncpy(args->cmd, (char *)packet.payload.execute_msg.cmd, sizeof(args->cmd)); - args->client_uuid = packet.header.client_uuid; - args->pipe_fd = pipe_fd; - args->is_atty = packet.payload.execute_msg.is_atty; - - _lock(); // need to lock, otherwise the thread could already exit before we insert into the map - ret = pthread_create(&new_pthread, nullptr, Server::_run_cmd, (void *)args); - - if (ret != 0) { - PX4_ERR("could not start pthread (%i)", ret); - delete args; - - } else { - // We won't join the thread, so detach to automatically release resources at its end - pthread_detach(new_pthread); - // We keep two maps for cleanup if a thread is finished or killed. - _client_uuid_to_pthread.insert(std::pair - (packet.header.client_uuid, new_pthread)); - _pthread_to_pipe_fd.insert(std::pair - (new_pthread, pipe_fd)); - } - - _unlock(); -} - - -void -Server::_kill_cmd_packet(const client_send_packet_s &packet) -{ - _lock(); - - // TODO: we currently ignore the signal type. - auto client_uuid_iter = _client_uuid_to_pthread.find(packet.header.client_uuid); - - if (client_uuid_iter == _client_uuid_to_pthread.end()) { - _unlock(); - return; - } - - pthread_t pthread_to_kill = client_uuid_iter->second; - - // TODO: use a more graceful exit method to avoid resource leaks - int ret = pthread_cancel(pthread_to_kill); - - __cleanup_thread(packet.header.client_uuid); - - _unlock(); - - if (ret != 0) { - PX4_ERR("failed to cancel thread"); - } - - // We don't send retval when we get killed. - // The client knows this and just exits without confirmation. -} - - - -void -*Server::_run_cmd(void *arg) -{ - RunCmdArgs *args = (RunCmdArgs *)arg; - - // Copy arguments so that we can cleanup the arg structure. - uint64_t client_uuid = args->client_uuid; - int pipe_fd = args->pipe_fd; - bool is_atty = args->is_atty; - std::string message_str(args->cmd); - - // Clean up the args from the heap in case the thread gets canceled - // from outside. - delete args; + // Last byte is 'isatty'. + uint8_t isatty = cmd.back(); + cmd.pop_back(); // We register thread specific data. This is used for PX4_INFO (etc.) log calls. CmdThreadSpecificData *thread_data_ptr; if ((thread_data_ptr = (CmdThreadSpecificData *)pthread_getspecific(_instance->_key)) == nullptr) { thread_data_ptr = new CmdThreadSpecificData; - thread_data_ptr->pipe_fd = pipe_fd; - thread_data_ptr->is_atty = is_atty; - thread_data_ptr->packet.header.msg_id = client_recv_packet_s::message_header_s::e_msg_id::STDOUT; + thread_data_ptr->fd = fd; + thread_data_ptr->is_atty = isatty; (void)pthread_setspecific(_instance->_key, (void *)thread_data_ptr); } // Run the actual command. - int retval = Pxh::process_line(message_str, true); + int retval = Pxh::process_line(cmd, true); // Report return value. - _send_retval(pipe_fd, retval, client_uuid); + char buf[2] = {0, (char)retval}; - // Clean up before returning. - _instance->_cleanup_thread(client_uuid); + if (write(fd, buf, sizeof buf) < 0) { + // Don't care it went wrong, as we're cleaning up anyway. + } + _cleanup(fd); return nullptr; } - void -Server::_send_retval(const int pipe_fd, const int retval, const uint64_t client_uuid) +Server::_cleanup(int fd) { - client_recv_packet_s packet; + _instance->_lock(); + _instance->_fd_to_thread.erase(fd); + _instance->_unlock(); - packet.header.msg_id = client_recv_packet_s::message_header_s::e_msg_id::RETVAL; - packet.header.payload_length = sizeof(packet.payload.retval_msg); - packet.payload.retval_msg.retval = retval; - - int bytes_to_send = get_client_recv_packet_length(&packet); - int bytes_sent = write(pipe_fd, &packet, bytes_to_send); - - if (bytes_sent != bytes_to_send) { - printf("write fail\n"); - return; - } -} - -void -Server::_cleanup_thread(const uint64_t client_uuid) -{ - _lock(); - __cleanup_thread(client_uuid); - _unlock(); -} - -void -Server::__cleanup_thread(const uint64_t client_uuid) -{ - pthread_t pthread_killed = _client_uuid_to_pthread[client_uuid]; - auto pipe_iter = _pthread_to_pipe_fd.find(pthread_killed); - - if (pipe_iter == _pthread_to_pipe_fd.end()) { - // can happen if the thread already exited and then got a kill packet - PX4_DEBUG("pipe fd already closed"); - return; - } - - int pipe_fd = pipe_iter->second; - close(pipe_fd); - - char path[RECV_PIPE_PATH_LEN] = {}; - get_client_recv_pipe_path(client_uuid, path, RECV_PIPE_PATH_LEN); - unlink(path); - - _client_uuid_to_pthread.erase(client_uuid); - _pthread_to_pipe_fd.erase(pthread_killed); + // We can't close() the fd here, since the main thread is probably + // polling for it: close()ing it causes a race condition. + // So, we only call shutdown(), which causes the main thread to register a + // 'POLLHUP', such that the main thread can close() it for us. + // We already removed this thread from _fd_to_thread, so there is no risk + // of the main thread trying to cancel this thread after it already exited. + shutdown(fd, SHUT_RDWR); } } //namespace px4_daemon diff --git a/platforms/posix/src/px4_daemon/server.h b/platforms/posix/src/px4_daemon/server.h index e1e0f8b845..1ce8d07675 100644 --- a/platforms/posix/src/px4_daemon/server.h +++ b/platforms/posix/src/px4_daemon/server.h @@ -1,6 +1,6 @@ /**************************************************************************** * - * Copyright (C) 2016 PX4 Development Team. All rights reserved. + * Copyright (C) 2016, 2018 PX4 Development Team. All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions @@ -33,12 +33,11 @@ /** * @file server.h * - * The server (also called daemon) opens a pipe for clients to write to. + * The server (also called daemon) opens a socket for clients to connect to. * - * Once a client connects it will send a command as well as a unique ID. + * Once a client connects it will send a command and close its side of the connection. * The server will return the stdout of the executing command, as well as the return - * value to the client on a client specific pipe. - * The client specific pipe are idenified by the unique ID of the client. + * value to the client. * * There should only every be one server running, therefore the static instance. * The Singleton implementation is not complete, but it should be obvious not @@ -46,6 +45,7 @@ * * @author Julian Oes * @author Beat Küng + * @author Mara Bos */ #pragma once @@ -54,7 +54,7 @@ #include #include -#include "pipe_protocol.h" +#include "sock_protocol.h" namespace px4_daemon @@ -74,11 +74,10 @@ public: */ int start(); - struct CmdThreadSpecificData { - int pipe_fd; // pipe fd to send data to descriptor + int fd; // fd to send stdout to bool is_atty; // whether file descriptor refers to a terminal - client_recv_packet_s packet; + char buffer[1024]; }; static bool is_running() @@ -92,17 +91,7 @@ public: } private: static void *_server_main_trampoline(void *arg); - void _server_main(void *arg); - - void _parse_client_send_packet(const client_send_packet_s &packet); - void _execute_cmd_packet(const client_send_packet_s &packet); - void _kill_cmd_packet(const client_send_packet_s &packet); - void _cleanup_thread(const uint64_t client_uuid); - - /** - * Like _cleanup_thread(), but does not take the mutex - */ - void __cleanup_thread(const uint64_t client_uuid); + void _server_main(); void _lock() { @@ -114,28 +103,19 @@ private: pthread_mutex_unlock(&_mutex); } - - static void _send_retval(const int pipe_fd, const int retval, const uint64_t client_uuid); - - struct RunCmdArgs { - char cmd[sizeof(client_send_packet_s::payload.execute_msg.cmd)]; - uint64_t client_uuid; - bool is_atty; - int pipe_fd; - }; - - static void *_run_cmd(void *arg); + static void *_handle_client(void *arg); + static void _cleanup(int fd); pthread_t _server_main_pthread; - std::map _pthread_to_pipe_fd; - std::map _client_uuid_to_pthread; - pthread_mutex_t _mutex; ///< protects access to _pthread_to_pipe_fd and _client_uuid_to_pthread + std::map _fd_to_thread; + pthread_mutex_t _mutex; ///< Protects _fd_to_thread. pthread_key_t _key; int _instance_id; ///< instance ID for running multiple instances of the px4 server + int _fd; static void _pthread_key_destructor(void *arg); diff --git a/platforms/posix/src/px4_daemon/server_io.cpp b/platforms/posix/src/px4_daemon/server_io.cpp index 72564b5c88..ba801053da 100644 --- a/platforms/posix/src/px4_daemon/server_io.cpp +++ b/platforms/posix/src/px4_daemon/server_io.cpp @@ -50,7 +50,7 @@ #include "server.h" #include -#include "pipe_protocol.h" +#include "sock_protocol.h" using namespace px4_daemon; @@ -84,10 +84,8 @@ int get_stdout_pipe_buffer(char **buffer, unsigned *max_length, bool *is_atty) #endif - client_recv_packet_s *packet = &thread_data_ptr->packet; - - *buffer = (char *)packet->payload.stdout_msg.text; - *max_length = sizeof(packet->payload.stdout_msg.text); + *buffer = thread_data_ptr->buffer; + *max_length = sizeof(thread_data_ptr->buffer); *is_atty = thread_data_ptr->is_atty; return 0; @@ -95,8 +93,6 @@ int get_stdout_pipe_buffer(char **buffer, unsigned *max_length, bool *is_atty) int send_stdout_pipe_buffer(unsigned buffer_length) { - assert(buffer_length <= sizeof(client_recv_packet_s::payload.stdout_msg.text)); - Server::CmdThreadSpecificData *thread_data_ptr; if (!Server::is_running()) { @@ -108,25 +104,9 @@ int send_stdout_pipe_buffer(unsigned buffer_length) return -1; } - client_recv_packet_s *packet = &thread_data_ptr->packet; - packet->header.payload_length = buffer_length; + int bytes_sent = write(thread_data_ptr->fd, thread_data_ptr->buffer, buffer_length); - int pipe_fd = thread_data_ptr->pipe_fd; - int bytes_to_send = get_client_recv_packet_length(packet); - - // Check if we can write first by writing 0 bytes. - // If we don't do this, we'll get SIGPIPE and be very unhappy - // because the whole process will go down. - int ret = write(pipe_fd, nullptr, 0); - - if (ret == 0 && errno == EPIPE) { - printf("Error: can't write to closed pipe, giving up.\n"); - pthread_exit(nullptr); - } - - int bytes_sent = write(pipe_fd, packet, bytes_to_send); - - if (bytes_sent != bytes_to_send) { + if (bytes_sent != (int)buffer_length) { printf("write fail\n"); return -1; } diff --git a/platforms/posix/src/px4_daemon/pipe_protocol.cpp b/platforms/posix/src/px4_daemon/sock_protocol.cpp similarity index 62% rename from platforms/posix/src/px4_daemon/pipe_protocol.cpp rename to platforms/posix/src/px4_daemon/sock_protocol.cpp index 57bc95ebfb..65b9e18000 100644 --- a/platforms/posix/src/px4_daemon/pipe_protocol.cpp +++ b/platforms/posix/src/px4_daemon/sock_protocol.cpp @@ -1,6 +1,6 @@ /**************************************************************************** * - * Copyright (C) 2016 PX4 Development Team. All rights reserved. + * Copyright (C) 2018 PX4 Development Team. All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions @@ -31,44 +31,20 @@ * ****************************************************************************/ /** - * @file pipe_protocol.cpp + * @file sock_protocol.cpp * - * @author Julian Oes - * @author Beat Küng + * @author Mara Bos */ -#include -#include -#include - -#include "pipe_protocol.h" - -static const char CLIENT_SEND_PIPE_PATH[] = "/tmp/px4_client_send_pipe-"; -static const char CLIENT_RECV_PIPE_PATH[] = "/tmp/px4_client_recv_pipe"; - +#include "sock_protocol.h" namespace px4_daemon { - -unsigned get_client_send_packet_length(const client_send_packet_s *packet) +std::string get_socket_path(int instance_id) { - return sizeof(client_send_packet_s) - sizeof(packet->payload) + packet->header.payload_length; -} - -unsigned get_client_recv_packet_length(const client_recv_packet_s *packet) -{ - return sizeof(client_recv_packet_s) - sizeof(packet->payload) + packet->header.payload_length; -} - -int get_client_recv_pipe_path(const uint64_t uuid, char *path, const size_t path_len) -{ - return snprintf(path, path_len, "%s-%016" PRIx64, CLIENT_RECV_PIPE_PATH, uuid); -} - -std::string get_client_send_pipe_path(int instance_id) -{ - return std::string(CLIENT_SEND_PIPE_PATH) + std::to_string(instance_id); + // TODO: Use /var/run/px4/$instance/sock (or /var/run/user/$UID/... for non-root). + return "/tmp/px4-sock-" + std::to_string(instance_id); } } // namespace px4_daemon diff --git a/platforms/posix/src/px4_daemon/pipe_protocol.h b/platforms/posix/src/px4_daemon/sock_protocol.h similarity index 52% rename from platforms/posix/src/px4_daemon/pipe_protocol.h rename to platforms/posix/src/px4_daemon/sock_protocol.h index 0f8012cdfd..f158a29b66 100644 --- a/platforms/posix/src/px4_daemon/pipe_protocol.h +++ b/platforms/posix/src/px4_daemon/sock_protocol.h @@ -1,6 +1,6 @@ /**************************************************************************** * - * Copyright (C) 2016 PX4 Development Team. All rights reserved. + * Copyright (C) 2018 PX4 Development Team. All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions @@ -31,69 +31,18 @@ * ****************************************************************************/ /** - * @file pipe_protocol.h + * @file sock_protocol.h * - * @author Julian Oes - * @author Beat Küng + * @author Mara Bos */ #pragma once -#include #include namespace px4_daemon { -static const unsigned RECV_PIPE_PATH_LEN = 64; - -// The packet size is no more than 512 bytes, because that is the minimum guaranteed size -// for a pipe to avoid interleaving of messages when multiple clients write at the same time -// (atomic writes). -struct client_send_packet_s { - struct message_header_s { - uint64_t client_uuid; - enum class e_msg_id : int { - EXECUTE, - KILL - } msg_id; - unsigned payload_length; - } header; - - union { - struct execute_msg_s { - uint8_t is_atty; - uint8_t cmd[512 - sizeof(message_header_s) - sizeof(uint8_t)]; - } execute_msg; - struct kill_msg_s { - int cmd_id; - } kill_msg; - } payload; -}; - -// We have per client receiver a pipe with the uuid in its file path. -struct client_recv_packet_s { - struct message_header_s { - enum class e_msg_id : int { - RETVAL, - STDOUT - } msg_id; - unsigned payload_length; - } header; - - union { - struct retval_msg_s { - int retval; - } retval_msg; - struct stdout_msg_s { - uint8_t text[512 - sizeof(message_header_s)]; ///< null-terminated string (payload_length includes the null) - } stdout_msg; - } payload; -}; - -unsigned get_client_send_packet_length(const client_send_packet_s *packet); -unsigned get_client_recv_packet_length(const client_recv_packet_s *packet); -int get_client_recv_pipe_path(const uint64_t uuid, char *path, const size_t path_len); -std::string get_client_send_pipe_path(int instance_id); +std::string get_socket_path(int instance_id); } // namespace px4_daemon