/* * Copyright (C) 2016 The Android Open Source Project * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ #ifndef SIMPLE_PERF_UNIX_SOCKET_H_ #define SIMPLE_PERF_UNIX_SOCKET_H_ #include <unistd.h> #include <functional> #include <memory> #include <mutex> #include <string> #include <vector> #include <android-base/logging.h> #include "IOEventLoop.h" #include "utils.h" // Class wrappers for unix socket communication operations. class UnixSocketConnection; // UnixSocketMessage is the message structure used for communication. struct UnixSocketMessage { uint32_t len; uint32_t type; char data[0]; }; // We want to avoid memory copy by being able to cast from char array // to UnixSocketMessage* directly (See the implementation in // UnixSocketConnection::ConsumeDataInReadBuffer()). To access members // of UnixSocketMessage and its extensions without causing alignment problems // (On arm, some instructions (like LDRD) don't support unaligned address), // we make sure all messages are stored at 8-bytes aligned addresses. Namely, // each message will be padded to 8-bytes aligned size. static constexpr uint32_t UnixSocketMessageAlignment = 8u; // UnixSocketMessageBuffer is a circular buffer used to store // UnixSocketMessages. class UnixSocketMessageBuffer { public: explicit UnixSocketMessageBuffer(size_t capacity) : data_(capacity), read_head_(0), valid_bytes_(0) {} bool Empty() const { return valid_bytes_ == 0; } bool HalfFull() const { return valid_bytes_ * 2 >= data_.size(); } bool StoreMessage(const UnixSocketMessage& message) { uint32_t aligned_len = Align(message.len, UnixSocketMessageAlignment); if (data_.size() - valid_bytes_ < aligned_len) { return false; } uint32_t write_head = (read_head_ + valid_bytes_) % data_.size(); if (message.len <= data_.size() - write_head) { memcpy(data_.data() + write_head, &message, message.len); } else { uint32_t len1 = data_.size() - write_head; memcpy(data_.data() + write_head, &message, len1); memcpy(data_.data(), reinterpret_cast<const char*>(&message) + len1, message.len - len1); } valid_bytes_ += aligned_len; return true; } size_t PeekData(const char** pdata) { *pdata = &data_[read_head_]; if (read_head_ + valid_bytes_ <= data_.size()) { return valid_bytes_; } return data_.size() - read_head_; } void CommitData(size_t size) { CHECK_GE(valid_bytes_, size); read_head_ = (read_head_ + size) % data_.size(); valid_bytes_ -= size; } private: std::vector<char> data_; uint32_t read_head_; uint32_t valid_bytes_; }; // UnixSocketServer creates a unix socket server listening on a unix file path. class UnixSocketServer { public: static std::unique_ptr<UnixSocketServer> Create( const std::string& server_path, bool is_abstract); ~UnixSocketServer(); const std::string& GetPath() const { return path_; } std::unique_ptr<UnixSocketConnection> AcceptConnection(); private: UnixSocketServer(int server_fd, const std::string& path) : server_fd_(server_fd), path_(path) {} const int server_fd_; const std::string path_; }; // UnixSocketConnection is used to communicate between server and client. // It is either created by accepting a connection in UnixSocketServer, or by // connecting to a UnixSocketServer. // UnixSocketConnection binds to a IOEventLoop, so it writes messages to fd // when it is writable, and read messages from fd when it is readable. To send // messages, UnixSocketConnection uses a buffer to store to-be-sent messages. // And whenever it receives a complete message from fd, it calls the callback // function. // In UnixSocketConnection, although user can send messages concurrently from // different threads, only the thread running IOEventLoop::RunLoop() can // do IO operations, calling WriteData() and ReadData(). To make it work // properly, the thread creating/destroying UnixSocketConnection should be // the same thread running IOEventLoop::RunLoop(). class UnixSocketConnection { private: static constexpr size_t SEND_BUFFER_SIZE = 512 * 1024; static constexpr size_t READ_BUFFER_SIZE = 16 * 1024; public: explicit UnixSocketConnection(int fd) : fd_(fd), read_buffer_(READ_BUFFER_SIZE), read_buffer_size_(0), read_event_(nullptr), send_buffer_(SEND_BUFFER_SIZE), write_event_enabled_(true), write_event_(nullptr), no_more_message_(false) {} static std::unique_ptr<UnixSocketConnection> Connect( const std::string& server_path, bool is_abstract); ~UnixSocketConnection(); bool IsClosed() { return fd_ == -1; } bool PrepareForIO(IOEventLoop& loop, const std::function<bool(const UnixSocketMessage&)>& receive_message_callback, const std::function<bool()>& close_connection_callback); // Thread-safe function, can be called from signal handler. // The message is put into the send buffer. If [undelayed] is true, messages // in the send buffer are sent immediately, otherwise they will be sent // when the buffer is half full. bool SendMessage(const UnixSocketMessage& message, bool undelayed) { std::lock_guard<std::mutex> lock(send_buffer_and_write_event_mtx_); if (no_more_message_ || !send_buffer_.StoreMessage(message)) { return false; } // By buffering messages, we can effectively decrease context-switch times. if (undelayed || send_buffer_.HalfFull()) { return EnableWriteEventWithLock(); } return true; } // Thread-safe function. // After NoMoreMessage(), the connection will not accept more messages // in SendMessage(), and it will be closed after sending existing messages // in send buffer. bool NoMoreMessage() { std::lock_guard<std::mutex> lock(send_buffer_and_write_event_mtx_); if (!no_more_message_) { no_more_message_ = true; return EnableWriteEventWithLock(); } return true; } private: // The caller should have send_buffer_and_write_event_mtx_ locked. bool EnableWriteEventWithLock() { if (!write_event_enabled_) { if (!IOEventLoop::EnableEvent(write_event_)) { return false; } write_event_enabled_ = true; } return true; } // The caller should have send_buffer_and_write_event_mtx_ locked. bool DisableWriteEventWithLock() { if (write_event_enabled_) { if (!IOEventLoop::DisableEvent(write_event_)) { return false; } write_event_enabled_ = false; } return true; } // Below functions are only called in the thread running IO operations. bool WriteData(); bool GetDataFromSendBuffer(const char** pdata, size_t* pdata_size); bool ReadData(); bool ConsumeDataInReadBuffer(); bool CloseConnection(); // Below members can only be accessed in the thread running IO operations. int fd_; std::function<bool(const UnixSocketMessage&)> read_callback_; std::function<bool()> close_callback_; // read_buffer_ is used to cache data read from the other end. // read_buffer_size_ is the number of valid bytes in read_buffer_. std::vector<char> read_buffer_; size_t read_buffer_size_; IOEventRef read_event_; // send_buffer_and_write_event_mtx_ protects following members, which can be // accessed in multiple threads. std::mutex send_buffer_and_write_event_mtx_; UnixSocketMessageBuffer send_buffer_; bool write_event_enabled_; IOEventRef write_event_; bool no_more_message_; }; #endif // SIMPLE_PERF_UNIX_SOCKET_H_