/*
* 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_