StarPilot/common/params.cc

#include "common/params.h"

#include <dirent.h>
#include <sys/file.h>

#include <algorithm>
#include <cassert>
#include <csignal>
#include <unordered_map>

#include "common/params_keys.h"
#include "common/queue.h"
#include "common/swaglog.h"
#include "common/util.h"
#include "system/hardware/hw.h"

namespace {

volatile sig_atomic_t params_do_exit = 0;
void params_sig_handler(int signal) {
  params_do_exit = 1;
}

int fsync_dir(const std::string &path) {
  int result = -1;
  int fd = HANDLE_EINTR(open(path.c_str(), O_RDONLY, 0755));
  if (fd >= 0) {
    result = HANDLE_EINTR(fsync(fd));
    HANDLE_EINTR(close(fd));
  }
  return result;
}

bool create_params_path(const std::string &param_path, const std::string &key_path) {
  // Make sure params path exists
  if (!util::file_exists(param_path) && !util::create_directories(param_path, 0775)) {
    return false;
  }

  // See if the symlink exists, otherwise create it
  if (!util::file_exists(key_path)) {
    // 1) Create temp folder
    // 2) Symlink it to temp link
    // 3) Move symlink to <params>/d

    std::string tmp_path = param_path + "/.tmp_XXXXXX";
    // this should be OK since mkdtemp just replaces characters in place
    char *tmp_dir = mkdtemp((char *)tmp_path.c_str());
    if (tmp_dir == NULL) {
      return false;
    }

    std::string link_path = std::string(tmp_dir) + ".link";
    if (symlink(tmp_dir, link_path.c_str()) != 0) {
      return false;
    }

    // don't return false if it has been created by other
    if (rename(link_path.c_str(), key_path.c_str()) != 0 && errno != EEXIST) {
      return false;
    }
  }

  return true;
}

std::string ensure_params_path(const std::string &prefix, const std::string &path = {}) {
  std::string params_path = path.empty() ? Path::params() : path;
  if (!create_params_path(params_path, params_path + prefix)) {
    throw std::runtime_error(util::string_format(
        "Failed to ensure params path, errno=%d, path=%s, param_prefix=%s",
        errno, params_path.c_str(), prefix.c_str()));
  }
  return params_path;
}

class FileLock {
public:
  FileLock(const std::string &fn) {
    fd_ = HANDLE_EINTR(open(fn.c_str(), O_CREAT, 0775));
    if (fd_ < 0 || HANDLE_EINTR(flock(fd_, LOCK_EX)) < 0) {
      LOGE("Failed to lock file %s, errno=%d", fn.c_str(), errno);
    }
  }
  ~FileLock() { close(fd_); }

private:
  int fd_ = -1;
};

} // namespace


Params::Params(const std::string &path, bool memory) {
  params_prefix = "/" + util::getenv("OPENPILOT_PREFIX", "d");

  // StarPilot variables
  std::string params_folder;
  if (memory) {
    params_folder = Path::shm_path() + "/params";
  } else {
    cache_path = "/cache/params" + params_prefix + "/";
    params_folder = path;
  }
  params_path = ensure_params_path(params_prefix, params_folder);
}

Params::~Params() {
  if (future.valid()) {
    future.wait();
  }
  std::scoped_lock lk(pending_writes_lock);
  assert(queue.empty());
  assert(pending_writes.empty());
  assert(!writer_running);
}

std::vector<std::string> Params::allKeys() const {
  std::vector<std::string> ret;
  for (auto &p : keys) {
    ret.push_back(p.first);
  }
  return ret;
}

bool Params::checkKey(const std::string &key) {
  return keys.find(key) != keys.end();
}

ParamKeyFlag Params::getKeyFlag(const std::string &key) {
  return static_cast<ParamKeyFlag>(keys[key].flags);
}

ParamKeyType Params::getKeyType(const std::string &key) {
  return keys[key].type;
}

std::optional<std::string> Params::getKeyDefaultValue(const std::string &key) {
  return keys[key].default_value;
}

int Params::put(const char* key, const char* value, size_t value_size) {
  // Information about safely and atomically writing a file: https://lwn.net/Articles/457667/
  // 1) Create temp file
  // 2) Write data to temp file
  // 3) fsync() the temp file
  // 4) rename the temp file to the real name
  // 5) fsync() the containing directory
  std::string tmp_path = params_path + "/.tmp_value_XXXXXX";
  int tmp_fd = mkstemp((char*)tmp_path.c_str());
  if (tmp_fd < 0) return -1;

  int result = -1;
  do {
    // Write value to temp.
    ssize_t bytes_written = HANDLE_EINTR(write(tmp_fd, value, value_size));
    if (bytes_written < 0 || (size_t)bytes_written != value_size) {
      result = -20;
      break;
    }

    // fsync to force persist the changes.
    if ((result = HANDLE_EINTR(fsync(tmp_fd))) < 0) break;

    FileLock file_lock(params_path + "/.lock");

    // Move temp into place.
    if ((result = rename(tmp_path.c_str(), getParamPath(key).c_str())) < 0) break;

    // fsync parent directory
    result = fsync_dir(getParamPath());
  } while (false);

  close(tmp_fd);
  if (result != 0) {
    ::unlink(tmp_path.c_str());
  }
  return result;
}

int Params::remove(const std::string &key) {
  FileLock file_lock(params_path + "/.lock");
  int result = unlink(getParamPath(key).c_str());

  // StarPilot variables
  if (!cache_path.empty()) {
    unlink((cache_path + key).c_str());
  }

  if (result != 0) {
    return result;
  }
  return fsync_dir(getParamPath());
}

std::string Params::get(const std::string &key, bool block) {
  if (!block) {
    return util::read_file(getParamPath(key));
  } else {
    // blocking read until successful
    params_do_exit = 0;
    void (*prev_handler_sigint)(int) = std::signal(SIGINT, params_sig_handler);
    void (*prev_handler_sigterm)(int) = std::signal(SIGTERM, params_sig_handler);

    std::string value;
    while (!params_do_exit) {
      if (value = util::read_file(getParamPath(key)); !value.empty()) {
        break;
      }
      util::sleep_for(100);  // 0.1 s
    }

    std::signal(SIGINT, prev_handler_sigint);
    std::signal(SIGTERM, prev_handler_sigterm);
    return value;
  }
}

std::map<std::string, std::string> Params::readAll() {
  FileLock file_lock(params_path + "/.lock");
  return util::read_files_in_dir(getParamPath());
}

void Params::clearAll(ParamKeyFlag key_flag) {
  FileLock file_lock(params_path + "/.lock");

  // 1) delete params of key_flag
  // 2) delete files that are not defined in the keys.
  if (DIR *d = opendir(getParamPath().c_str())) {
    struct dirent *de = NULL;
    while ((de = readdir(d))) {
      if (de->d_type != DT_DIR) {
        auto it = keys.find(de->d_name);
        if (it == keys.end() || (it->second.flags & key_flag)) {
          unlink(getParamPath(de->d_name).c_str());

          // StarPilot variables
          if (!cache_path.empty()) {
            unlink((cache_path + de->d_name).c_str());
          }
        }
      }
    }
    closedir(d);
  }

  fsync_dir(getParamPath());
}

void Params::putNonBlocking(const std::string &key, const std::string &val) {
  bool should_enqueue = false;
  bool should_start_thread = false;

  {
    std::scoped_lock lk(pending_writes_lock);
    auto it = pending_writes.find(key);
    if (it == pending_writes.end()) {
      pending_writes.emplace(key, val);
      should_enqueue = true;
    } else {
      it->second = val;
    }

    if (!writer_running) {
      writer_running = true;
      should_start_thread = true;
    }
  }

  if (should_enqueue) {
    queue.push(key);
  }

  if (should_start_thread) {
    future = std::async(std::launch::async, &Params::asyncWriteThread, this);
  }
}

void Params::asyncWriteThread() {
  std::string key;
  while (true) {
    if (!queue.try_pop(key, 0)) {
      std::scoped_lock lk(pending_writes_lock);
      if (queue.empty() && pending_writes.empty()) {
        writer_running = false;
        return;
      }
      continue;
    }

    std::string val;
    {
      std::scoped_lock lk(pending_writes_lock);
      auto it = pending_writes.find(key);
      if (it == pending_writes.end()) {
        continue;
      }
      val = std::move(it->second);
      pending_writes.erase(it);
    }

    // Params::put is Thread-Safe
    put(key, val);
  }
}

// StarPilot variables
int Params::getTuningLevel(const std::string &key) {
  return keys[key].tuning_level;
}

std::optional<std::string> Params::getStockValue(const std::string &key) {
  ParamKeyAttributes &attributes = keys[key];
  if (attributes.stock_value) {
    return attributes.stock_value;
  }
  return attributes.default_value;
}