tlbmc/hal/shared_mem/static_server_impl.cc

#include "tlbmc/hal/shared_mem/static_server_impl.h"

#include <cerrno>
#include <cstdint>
#include <filesystem>  // NOLINT
#include <fstream>
#include <limits>
#include <memory>
#include <new>
#include <string>
#include <string_view>
#include <system_error>  // NOLINT
#include <utility>

#include "absl/base/no_destructor.h"
#include "absl/container/flat_hash_map.h"
#include "absl/log/log.h"
#include "absl/status/status.h"
#include "absl/status/statusor.h"
#include "absl/strings/str_cat.h"
#include "absl/time/time.h"
#include "absl/types/span.h"
#include "boost/interprocess/exceptions.hpp"  // NOLINT
#include "boost/interprocess/managed_shared_memory.hpp"  //NOLINT
#include "g3/macros.h"
#include "tlbmc/hal/shared_mem/metrics.h"
#include "tlbmc/hal/shared_mem/sensors.h"
#include "tlbmc/hal/shared_mem/server_interface.h"
#include "tlbmc/hal/shared_mem/static_shm_common.h"

namespace milotic_tlbmc {
namespace {
bool PathExists(const std::filesystem::path& path) {
  std::error_code ec;
  bool exists = std::filesystem::exists(path, ec);
  if (ec) {
    LOG(ERROR) << "Filesystem error while checking for " << path << ": "
               << ec.message();
    return false;
  }
  return exists;
}
}  // namespace

static std::unique_ptr<TlbmcSharedMemoryType> FindExistingSegment(
    const std::string& initialized_file_path, const std::string& shm_name) {
  if (!PathExists(initialized_file_path)) {
    return nullptr;
  }
  try {
    return std::make_unique<TlbmcSharedMemoryType>(
        boost::interprocess::open_only, shm_name.c_str());
  } catch (const boost::interprocess::interprocess_exception& e) {
    LOG(ERROR) << "Failed to open shared memory: " << e.what();
    return nullptr;
  }

  return nullptr;
}

static void LoadDataFromSegment(
    TlbmcSharedMemoryType& segment,
    absl::flat_hash_map<std::string, IpcSensor*>& sensors_idx) {
  for (auto it = segment.named_begin(); it != segment.named_end(); ++it) {
    const std::string_view name(it->name(), it->name_length());
    if (name.empty()) {
      continue;
    }
    if (name.starts_with("__")) {
      continue;
    }

    auto val = segment.find<IpcSensor>(it->name());
    if (val.first != nullptr) {
      sensors_idx[name] = val.first;
    }
  }
}

static std::unique_ptr<StaticSharedMemoryServer> LoadServer(
    const std::string& initialized_file_path, const std::string& shm_name) {
  auto shared_memory = FindExistingSegment(initialized_file_path, shm_name);
  if (shared_memory == nullptr) {
    return nullptr;
  }
  absl::flat_hash_map<std::string, IpcSensor*> sensors_idx;
  LoadDataFromSegment(*shared_memory, sensors_idx);
  return std::make_unique<StaticSharedMemoryServer>(
      std::move(shared_memory), std::move(sensors_idx),
      StaticSharedMemoryMetrics::GetInProcessMetrics());
}
/*   These values are approximate, with some headroom. Precise calculation would
 * require factoring in memory alignment, size of internal data structures, and
 * length of sensor names
 */
constexpr uint64_t kBytesPerSensor = 128;
constexpr uint64_t kMinObjectOverhead = 64;
constexpr uint64_t kConstantOverhead = 256;
static std::unique_ptr<TlbmcSharedMemoryType> CreateNewSegment(
    const std::string& shm_name, uint64_t sensor_count) {
  try {
    // Creates the segment.
    if (boost::interprocess::shared_memory_object::remove(shm_name.c_str())) {
      LOG(WARNING) << "Shared memory already existed without a initialization "
                      "file. Cleared previous segment ";
    }
    uint64_t size_to_reserve =
        (sensor_count * (sizeof(IpcSensor) + kBytesPerSensor)) +
        kConstantOverhead;

    return std::make_unique<TlbmcSharedMemoryType>(
        boost::interprocess::create_only, shm_name.c_str(), size_to_reserve);
  } catch (...) {
    return nullptr;
  }
  return nullptr;
}

class FallbackServer : public IpcServer {
 public:
  absl::StatusOr<std::pair<float, uint64_t>> ReadSensorValue(
      const std::string& /*sensor_name*/) override {
    return absl::UnavailableError(
        "Using fallback server due to error initializing shared memory");
  }
  // copybara:strip_begin(g3-shared-libs)
  void UpdateMetricsRequestCount(bool is_tlbmc_request) override {
    StaticSharedMemoryMetrics::GetInProcessMetrics()->UpdateMetricsRequestCount(
        is_tlbmc_request);
  }
  void UpdateMetricsResponse(absl::Duration response_time, int status_code,
                             std::string_view resource_url) override {
    StaticSharedMemoryMetrics::GetInProcessMetrics()->UpdateMetricsResponse(
        response_time, status_code, resource_url);
  }
  void UpdateMetricsRps(absl::Duration time_delta) override {
    StaticSharedMemoryMetrics::GetInProcessMetrics()->UpdateMetricsRps(
        time_delta);
  }
  // copybara:strip_end
};
class StaticSharedMemoryInitializer {
  static absl::Status CreateInitFile(const std::string& initialized_file_path) {
    std::filesystem::path const directory =
        std::filesystem::path(initialized_file_path).parent_path();
    if (!PathExists(directory)) {
      std::error_code ec;
      std::filesystem::create_directories(directory, ec);
      if (ec) {
        LOG(ERROR) << "Failed to create directory: " << directory << ": "
                   << ec.message();
        return absl::InternalError("Failed to create directory");
      }
    }
    errno = 0;
    std::ofstream file(std::string{initialized_file_path});
    if (!file.is_open()) {
      int const error_num = errno;
      auto error_message =
          absl::StrCat("Failed to open file at: ", initialized_file_path);
      if (error_num != 0) {
        return absl::ErrnoToStatus(error_num, error_message);
      }
      return absl::UnavailableError(error_message);
    }
    // An error can happen during write/flush but we don't rely on the contents
    // of the file so we can ignore the error here
    file << 1;
    file.flush();
    return absl::OkStatus();
  }
  static absl::Status Allocate(TlbmcSharedMemoryType& segment,
                               absl::Span<const std::string> shm_sensors) {
    segment.reserve_named_objects(shm_sensors.size());
    for (const auto& sensor_name : shm_sensors) {
      if (segment.construct<IpcSensor>(sensor_name.c_str(), std::nothrow)() ==
          nullptr) {
        return absl::UnknownError("Cannot allocate sensor");
      }
    }
    return absl::OkStatus();
  }

 public:
  static absl::Status Initialize(
      const std::string& initialized_file_path, const std::string& shm_name,
      const absl::Span<const std::string> shm_sensors) {
    if (FindExistingSegment(initialized_file_path, shm_name) != nullptr) {
      return absl::OkStatus();
    }
    auto new_segment = CreateNewSegment(shm_name, shm_sensors.size());
    if (new_segment == nullptr) {
      return absl::InternalError("Failed to create new shared memory");
    }
    ECCLESIA_RETURN_IF_ERROR(Allocate(*new_segment, shm_sensors));
    auto status = CreateInitFile(initialized_file_path);
    if (!status.ok()) {
      LOG(ERROR) << "Failed to create init file" << status;
      return status;
    }
    return absl::OkStatus();
  }

  static void TearDownForTesting(const std::string& initialized_file_path) {
    std::error_code ec;
    std::filesystem::remove(initialized_file_path, ec);
    if (ec) {
      LOG(ERROR) << "Error calling remove for path '" << initialized_file_path
                 << "': " << ec.message();
    }
  }
};
absl::Status StaticSharedMemoryServer::InitializeSharedMemoryForUnitTest(
    absl::Span<const std::string> shm_sensors) {
  StaticSharedMemoryInitializer::TearDownForTesting(
      "/tmp/tlbmc/static_shm_initialized");
  return InitializeSharedMemory("/tmp/tlbmc/static_shm_initialized",
                                kStaticShmName, shm_sensors);
}
void StaticSharedMemoryServer::SetupInstanceForUnitTest() {
  GetInstance("/tmp/tlbmc/static_shm_initialized", kStaticShmName, true);
}

absl::Status StaticSharedMemoryServer::InitializeSharedMemory(
    absl::Span<const std::string> shm_sensors) {
  return InitializeSharedMemory(kStaticShmInitializedFile, kStaticShmName,
                                shm_sensors);
}
absl::Status StaticSharedMemoryServer::InitializeSharedMemory(
    const std::string& initialized_file_path, const std::string& shm_name,
    absl::Span<const std::string> shm_sensors) {
  return StaticSharedMemoryInitializer::Initialize(initialized_file_path,
                                                   shm_name, shm_sensors);
}
IpcServer& StaticSharedMemoryServer::GetInstance() {
  return GetInstance(kStaticShmInitializedFile, kStaticShmName);
}
std::unique_ptr<IpcServer> StaticSharedMemoryServer::PerformServerInit(
    const std::string& initialized_file_path, const std::string& shm_name) {
  std::unique_ptr<StaticSharedMemoryServer> server =
      LoadServer(initialized_file_path, shm_name);
  if (server == nullptr) {
    LOG(ERROR) << "Failed to initialize shared memory server , falling "
                  "back to dummy ";
    return std::make_unique<FallbackServer>();
  }
  return server;
}
IpcServer& StaticSharedMemoryServer::GetInstance(
    const std::string& initialized_file_path, const std::string& shm_name,
    bool force_reset_for_testing) {
  static absl::NoDestructor instance(
      (PerformServerInit(initialized_file_path, shm_name)));
  if (force_reset_for_testing) {
    *instance = PerformServerInit(initialized_file_path, shm_name);
  }
  return **instance;
}

absl::StatusOr<std::pair<float, uint64_t>>
StaticSharedMemoryServer::ReadSensorValue(const std::string& sensor_name) {
  auto sensor_entry = name_to_sensor_.find(sensor_name);
  if (sensor_entry == name_to_sensor_.end()) {
    return absl::UnavailableError("Requested sensor does not exist");
  }
  std::pair<float, uint64_t> value = sensor_entry->second->GetValue();

  // If the sensor value is not updated by the client, the value will be set
  // to infinity and the timestamp will be 0.
  if (value.first == std::numeric_limits<float>::infinity() ||
      value.second == 0) {
    return absl::UnavailableError(
        "Sensor value is still invalid due to no update from client.");
  }
  return value;
}

// copybara:strip_begin(g3-shared-libs)
void StaticSharedMemoryServer::UpdateMetricsRequestCount(
    bool is_tlbmc_request) {
  if (metrics_ == nullptr) {
    return;
  }
  metrics_->UpdateMetricsRequestCount(is_tlbmc_request);
}

void StaticSharedMemoryServer::UpdateMetricsResponse(
    absl::Duration response_time, int status_code,
    std::string_view resource_url) {
  if (metrics_ == nullptr) {
    return;
  }
  metrics_->UpdateMetricsResponse(response_time, status_code, resource_url);
}

void StaticSharedMemoryServer::UpdateMetricsRps(absl::Duration time_delta) {
  if (metrics_ == nullptr) {
    return;
  }
  metrics_->UpdateMetricsRps(time_delta);
}
// copybara:strip_end
StaticSharedMemoryServer::StaticSharedMemoryServer(
    std::unique_ptr<TlbmcSharedMemoryType> memory,
    absl::flat_hash_map<std::string, IpcSensor*> name_to_sensor,
    TlbmcMetrics* metrics)
    : memory_(std::move(memory)),
      name_to_sensor_(std::move(name_to_sensor)),
      metrics_(metrics) {}

}  // namespace milotic_tlbmc