simpleperf/UnixSocket.h

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