// Copyright 2019 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "src/storage/blobfs/pager/user-pager.h"

#include <fuchsia/scheduler/cpp/fidl.h>
#include <lib/fdio/directory.h>
#include <lib/fit/defer.h>
#include <lib/fzl/owned-vmo-mapper.h>
#include <lib/syslog/cpp/macros.h>
#include <lib/zx/thread.h>
#include <limits.h>
#include <zircon/status.h>
#include <zircon/threads.h>

#include <algorithm>
#include <memory>

#include <fbl/auto_call.h>
#include <fs/trace.h>

#include "src/storage/blobfs/format.h"
#include "src/storage/blobfs/metrics.h"
#include "src/storage/lib/watchdog/include/lib/watchdog/operations.h"

namespace blobfs {
namespace pager {

UserPager::UserPager(size_t decompression_buffer_size, BlobfsMetrics* metrics)
    : decompression_buffer_size_(decompression_buffer_size), metrics_(metrics) {}

zx::status<std::unique_ptr<UserPager>> UserPager::Create(
    std::unique_ptr<TransferBuffer> uncompressed_buffer,
    std::unique_ptr<TransferBuffer> compressed_buffer, size_t decompression_buffer_size,
    BlobfsMetrics* metrics, bool sandbox_decompression) {
  ZX_DEBUG_ASSERT(metrics != nullptr && uncompressed_buffer != nullptr &&
                  uncompressed_buffer->vmo().is_valid() && compressed_buffer != nullptr &&
                  compressed_buffer->vmo().is_valid());

  if (uncompressed_buffer->size() % kBlobfsBlockSize ||
      compressed_buffer->size() % kBlobfsBlockSize ||
      decompression_buffer_size % kBlobfsBlockSize) {
    return zx::error(ZX_ERR_INVALID_ARGS);
  }
  if (compressed_buffer->size() < decompression_buffer_size) {
    return zx::error(ZX_ERR_INVALID_ARGS);
  }

  TRACE_DURATION("blobfs", "UserPager::Create");

  auto pager = std::unique_ptr<UserPager>(new UserPager(decompression_buffer_size, metrics));
  pager->uncompressed_transfer_buffer_ = std::move(uncompressed_buffer);
  pager->compressed_transfer_buffer_ = std::move(compressed_buffer);

  zx_status_t status =
      pager->compressed_mapper_.Map(pager->compressed_transfer_buffer_->vmo(), 0,
                                    pager->compressed_transfer_buffer_->size(), ZX_VM_PERM_READ);
  if (status != ZX_OK) {
    FX_LOGS(ERROR) << "Failed to map the compressed TransferBuffer: "
                   << zx_status_get_string(status);
    return zx::error(status);
  }

  status = zx::vmo::create(pager->decompression_buffer_size_, 0, &pager->decompression_buffer_);
  if (status != ZX_OK) {
    FX_LOGS(ERROR) << "Failed to create decompression buffer: " << zx_status_get_string(status);
    return zx::error(status);
  }

  if (sandbox_decompression) {
    status = zx::vmo::create(kDecompressionBufferSize, 0, &pager->sandbox_buffer_);
    if (status != ZX_OK) {
      FX_LOGS(ERROR) << "blobfs: Failed to create sandbox buffer: %s\n"
                     << zx_status_get_string(status);
      return zx::error(status);
    }

    auto client_or = ExternalDecompressorClient::Create(pager->sandbox_buffer_,
                                                        pager->compressed_transfer_buffer_->vmo());
    if (!client_or.is_ok()) {
      return zx::error(client_or.status_value());
    }
    pager->decompressor_client_ = std::move(client_or.value());
  }

  // Create the pager object.
  status = zx::pager::create(0, &pager->pager_);
  if (status != ZX_OK) {
    FX_LOGS(ERROR) << "Cannot initialize pager";
    return zx::error(status);
  }

  // Start the pager thread.
  thrd_t thread;
  status = pager->pager_loop_.StartThread("blobfs-pager-thread", &thread);
  if (status != ZX_OK) {
    FX_LOGS(ERROR) << "Could not start pager thread";
    return zx::error(status);
  }

  // Set a scheduling deadline profile for the blobfs-pager-thread. This is purely a performance
  // optimization, and failure to do so is not fatal. So in the case of an error encountered
  // in any of the steps within |SetDeadlineProfile|, we log a warning, and successfully return the
  // UserPager instance.
  SetDeadlineProfile(thread);

  // Initialize and start the watchdog.
  pager->watchdog_ = fs_watchdog::CreateWatchdog();
  zx::status<> watchdog_status = pager->watchdog_->Start();
  if (!watchdog_status.is_ok()) {
    FX_LOGS(ERROR) << "Could not start pager watchdog";
    return zx::error(watchdog_status.status_value());
  }

  return zx::ok(std::move(pager));
}

void UserPager::SetDeadlineProfile(thrd_t thread) {
  zx::channel channel0, channel1;
  zx_status_t status = zx::channel::create(0u, &channel0, &channel1);
  if (status != ZX_OK) {
    FX_LOGS(WARNING) << "Could not create channel pair: " << zx_status_get_string(status);
    return;
  }

  // Connect to the scheduler profile provider service.
  status = fdio_service_connect(
      (std::string("/svc_blobfs/") + fuchsia::scheduler::ProfileProvider::Name_).c_str(),
      channel0.release());
  if (status != ZX_OK) {
    FX_LOGS(WARNING) << "Could not connect to scheduler profile provider: "
                     << zx_status_get_string(status);
    return;
  }

  fuchsia::scheduler::ProfileProvider_SyncProxy provider(std::move(channel1));

  zx_status_t fidl_status = ZX_OK;
  zx::profile profile;

  // Deadline profile parameters for the pager thread.
  // Details on the performance analysis to arrive at these numbers can be found in fxbug.dev/56291.
  //
  // TODO(fxbug.dev/40858): Migrate to the role-based API when available, instead of hard
  // coding parameters.
  const zx_duration_t capacity = ZX_USEC(1800);
  const zx_duration_t deadline = ZX_USEC(2800);
  const zx_duration_t period = deadline;

  status = provider.GetDeadlineProfile(
      capacity, deadline, period, "/boot/bin/blobfs:blobfs-pager-thread", &fidl_status, &profile);

  if (status != ZX_OK || fidl_status != ZX_OK) {
    FX_LOGS(WARNING) << "Failed to get deadline profile: " << zx_status_get_string(status) << ", "
                     << zx_status_get_string(fidl_status);
  } else {
    auto pager_thread = zx::unowned_thread(thrd_get_zx_handle(thread));
    // Set the deadline profile.
    status = pager_thread->set_profile(profile, 0);
    if (status != ZX_OK) {
      FX_LOGS(WARNING) << "Failed to set deadline profile: " << zx_status_get_string(status);
    }
  }
}

UserPager::ReadRange UserPager::GetBlockAlignedReadRange(const UserPagerInfo& info, uint64_t offset,
                                                         uint64_t length) const {
  ZX_DEBUG_ASSERT(offset < info.data_length_bytes);
  // Clamp the range to the size of the blob.
  length = std::min(length, info.data_length_bytes - offset);

  // Align to the block size for verification. (In practice this means alignment to 8k).
  zx_status_t status = info.verifier->Align(&offset, &length);
  // This only happens if the info.verifier thinks that [offset,length) is out of range, which
  // will only happen if |verifier| was initialized with a different length than the rest of |info|
  // (which is a programming error).
  ZX_DEBUG_ASSERT(status == ZX_OK);

  ZX_DEBUG_ASSERT(offset % kBlobfsBlockSize == 0);
  ZX_DEBUG_ASSERT(length % kBlobfsBlockSize == 0 || offset + length == info.data_length_bytes);

  return {.offset = offset, .length = length};
}

UserPager::ReadRange UserPager::GetBlockAlignedExtendedRange(const UserPagerInfo& info,
                                                             uint64_t offset,
                                                             uint64_t length) const {
  // TODO(rashaeqbal): Consider making the cluster size dynamic once we have prefetch read
  // efficiency metrics from the kernel - i.e. what percentage of prefetched pages are actually
  // used. Note that dynamic prefetch sizing might not play well with compression, since we
  // always need to read in entire compressed frames.
  //
  // TODO(rashaeqbal): Consider extending the range backwards as well. Will need some way to track
  // populated ranges.
  //
  // Read in at least 32KB at a time. This gives us the best performance numbers w.r.t. memory
  // savings and observed latencies. Detailed results from experiments to tune this can be found in
  // fxbug.dev/48519.
  constexpr uint64_t kReadAheadClusterSize = (32 * (1 << 10));

  size_t read_ahead_offset = offset;
  size_t read_ahead_length = std::max(kReadAheadClusterSize, length);
  read_ahead_length = std::min(read_ahead_length, info.data_length_bytes - read_ahead_offset);

  // Align to the block size for verification. (In practice this means alignment to 8k).
  return GetBlockAlignedReadRange(info, read_ahead_offset, read_ahead_length);
}

PagerErrorStatus UserPager::TransferPagesToVmo(uint64_t offset, uint64_t length, const zx::vmo& vmo,
                                               const UserPagerInfo& info) {
  size_t end;
  if (add_overflow(offset, length, &end)) {
    FX_LOGS(ERROR) << "pager transfer range would overflow (off=" << offset << ", len=" << length
                   << ")";
    return PagerErrorStatus::kErrBadState;
  }

  static const fs_watchdog::FsOperationType kOperation(
      fs_watchdog::FsOperationType::CommonFsOperation::PageFault, std::chrono::seconds(60));
  [[maybe_unused]] fs_watchdog::FsOperationTracker tracker(&kOperation, watchdog_.get());

  if (info.decompressor != nullptr) {
    return TransferChunkedPagesToVmo(offset, length, vmo, info);
  } else {
    return TransferUncompressedPagesToVmo(offset, length, vmo, info);
  }
}

// The requested range is aligned in multiple steps as follows:
// 1. The range is extended to speculatively read in 32k at a time.
// 2. The extended range is further aligned for Merkle tree verification later.
// 3. This range is read in chunks equal to the size of the uncompressed_transfer_buffer_. Each
// chunk is verified as it is read in, and spliced into the destination VMO with supply_pages().
//
// The assumption here is that the transfer buffer is sized per the alignment requirements for
// Merkle tree verification. We have static asserts in place to check this assumption - the transfer
// buffer (256MB) is 8k block aligned.
PagerErrorStatus UserPager::TransferUncompressedPagesToVmo(uint64_t requested_offset,
                                                           uint64_t requested_length,
                                                           const zx::vmo& vmo,
                                                           const UserPagerInfo& info) {
  ZX_DEBUG_ASSERT(!info.decompressor);

  const auto [start_offset, total_length] =
      GetBlockAlignedExtendedRange(info, requested_offset, requested_length);

  TRACE_DURATION("blobfs", "UserPager::TransferUncompressedPagesToVmo", "offset", start_offset,
                 "length", total_length);

  uint64_t offset = start_offset;
  uint64_t length_remaining = total_length;
  uint64_t length;

  // Read in multiples of the transfer buffer size. In practice we should only require one iteration
  // for the majority of cases, since the transfer buffer is 256MB.
  while (length_remaining > 0) {
    length = std::min(uncompressed_transfer_buffer_->size(), length_remaining);

    auto decommit = fit::defer([this, length = length]() {
      // Decommit pages in the transfer buffer that might have been populated. All blobs share the
      // same transfer buffer - this prevents data leaks between different blobs.
      uncompressed_transfer_buffer_->vmo().op_range(
          ZX_VMO_OP_DECOMMIT, 0, fbl::round_up(length, kBlobfsBlockSize), nullptr, 0);
    });

    // Read from storage into the transfer buffer.
    auto populate_status = uncompressed_transfer_buffer_->Populate(offset, length, info);
    if (!populate_status.is_ok()) {
      FX_LOGS(ERROR) << "TransferUncompressed: Failed to populate transfer vmo: "
                     << populate_status.status_string();
      return ToPagerErrorStatus(populate_status.status_value());
    }

    const uint64_t rounded_length = fbl::round_up<uint64_t, uint64_t>(length, PAGE_SIZE);

    // The block size is a multiple of the page size and |length| has already been block aligned. If
    // |rounded_length| is greater than |length| then |length| isn't block aligned because it's at
    // the end of the blob.  In the compact layout the Merkle tree can share the last block of the
    // data and may have been read into the transfer buffer.  The Merkle tree needs to be removed
    // before transfering the pages to the destination VMO.
    static_assert(kBlobfsBlockSize % PAGE_SIZE == 0);
    if (rounded_length > length) {
      zx_status_t status = uncompressed_transfer_buffer_->vmo().op_range(
          ZX_VMO_OP_ZERO, length, rounded_length - length, nullptr, 0);
      if (status != ZX_OK) {
        FX_LOGS(ERROR) << "TransferUncompressed: Failed to remove Merkle tree from "
                          "transfer buffer: "
                       << zx_status_get_string(status);
        return ToPagerErrorStatus(status);
      }
    }

    // Verify the pages read in.
    {
      fzl::VmoMapper mapping;
      // We need to unmap the transfer VMO before its pages can be transferred to the destination
      // VMO, via |zx_pager_supply_pages|.
      auto unmap = fit::defer([&]() { mapping.Unmap(); });

      // Map the transfer VMO in order to pass the verifier a pointer to the data.
      zx_status_t status =
          mapping.Map(uncompressed_transfer_buffer_->vmo(), 0, rounded_length, ZX_VM_PERM_READ);
      if (status != ZX_OK) {
        FX_LOGS(ERROR) << "TransferUncompressed: Failed to map transfer buffer: "
                       << zx_status_get_string(status);
        return ToPagerErrorStatus(status);
      }

      status = info.verifier->VerifyPartial(mapping.start(), length, offset, rounded_length);
      if (status != ZX_OK) {
        FX_LOGS(ERROR) << "TransferUncompressed: Failed to verify data: "
                       << zx_status_get_string(status);
        return ToPagerErrorStatus(status);
      }
    }

    ZX_DEBUG_ASSERT(offset % PAGE_SIZE == 0);
    // Move the pages from the transfer buffer to the destination VMO.
    zx_status_t status =
        pager_.supply_pages(vmo, offset, rounded_length, uncompressed_transfer_buffer_->vmo(), 0);
    if (status != ZX_OK) {
      FX_LOGS(ERROR) << "TransferUncompressed: Failed to supply pages to paged VMO: "
                     << zx_status_get_string(status);
      return ToPagerErrorStatus(status);
    }

    length_remaining -= length;
    offset += length;
  }

  fbl::String merkle_root_hash = info.verifier->digest().ToString();
  metrics_->IncrementPageIn(merkle_root_hash, start_offset, total_length);

  return PagerErrorStatus::kOK;
}

// The requested range is aligned in multiple steps as follows:
// 1. The desired uncompressed range is aligned for Merkle tree verification.
// 2. This range is extended to span complete compression frames / chunks, since that is the
// granularity we can decompress data in. The result of this alignment produces a
// CompressionMapping, which contains the mapping of the requested uncompressed range to the
// compressed range that needs to be read in from disk.
// 3. The uncompressed range is processed in chunks equal to the decompression_buffer_size_. For
// each chunk, we compute the CompressionMapping to determine the compressed range that needs to be
// read in. Each chunk is uncompressed and verified as it is read in, and spliced into the
// destination VMO with supply_pages().
//
// There are two assumptions we make here: First that the decompression buffer is sized per the
// alignment requirements for Merkle tree verification. And second that the transfer buffer is sized
// such that it can accommodate all the compressed data for the decompression buffer, i.e. the
// transfer buffer should work with the worst case compression ratio of 1. We have static asserts in
// place to check both these assumptions - the transfer buffer is the same size as the decompression
// buffer (256MB), and both these buffers are 8k block aligned.
PagerErrorStatus UserPager::TransferChunkedPagesToVmo(uint64_t requested_offset,
                                                      uint64_t requested_length, const zx::vmo& vmo,
                                                      const UserPagerInfo& info) {
  ZX_DEBUG_ASSERT(info.decompressor);

  const auto [offset, length] = GetBlockAlignedReadRange(info, requested_offset, requested_length);

  TRACE_DURATION("blobfs", "UserPager::TransferChunkedPagesToVmo", "offset", offset, "length",
                 length);

  fbl::String merkle_root_hash = info.verifier->digest().ToString();

  size_t current_decompressed_offset = offset;
  size_t desired_decompressed_end = offset + length;

  // Read in multiples of the decompression buffer size. In practice we should only require one
  // iteration for the majority of cases, since the decompression buffer is 256MB.
  while (current_decompressed_offset < desired_decompressed_end) {
    size_t current_decompressed_length = desired_decompressed_end - current_decompressed_offset;
    zx::status<CompressionMapping> mapping_status = info.decompressor->MappingForDecompressedRange(
        current_decompressed_offset, current_decompressed_length, decompression_buffer_size_);

    if (!mapping_status.is_ok()) {
      FX_LOGS(ERROR) << "TransferChunked: Failed to find range for [" << offset << ", "
                     << current_decompressed_offset + current_decompressed_length
                     << "): " << mapping_status.status_string();
      return ToPagerErrorStatus(mapping_status.status_value());
    }
    CompressionMapping mapping = mapping_status.value();

    // The compressed frame may not fall at a block aligned address, but we read in block aligned
    // chunks. This offset will be applied to the buffer we pass to decompression.
    // TODO(jfsulliv): Caching blocks which span frames may be useful for performance.
    size_t offset_of_compressed_data = mapping.compressed_offset % kBlobfsBlockSize;

    // Read from storage into the transfer buffer.
    size_t read_offset = fbl::round_down(mapping.compressed_offset, kBlobfsBlockSize);
    size_t read_len = (mapping.compressed_length + offset_of_compressed_data);

    auto decommit_compressed = fit::defer([this, length = read_len]() {
      // Decommit pages in the transfer buffer that might have been populated. All blobs share the
      // same transfer buffer - this prevents data leaks between different blobs.
      compressed_transfer_buffer_->vmo().op_range(
          ZX_VMO_OP_DECOMMIT, 0, fbl::round_up(length, kBlobfsBlockSize), nullptr, 0);
    });

    auto populate_status = compressed_transfer_buffer_->Populate(read_offset, read_len, info);
    if (!populate_status.is_ok()) {
      FX_LOGS(ERROR) << "TransferChunked: Failed to populate transfer vmo: "
                     << populate_status.status_string();
      return ToPagerErrorStatus(populate_status.status_value());
    }

    auto decommit_decompressed = fit::defer([this, length = mapping.decompressed_length]() {
      // Decommit pages in the decompression buffer that might have been populated. All blobs share
      // the same transfer buffer - this prevents data leaks between different blobs.
      decompression_buffer_.op_range(ZX_VMO_OP_DECOMMIT, 0, fbl::round_up(length, kBlobfsBlockSize),
                                     nullptr, 0);
    });

    // Map the decompression VMO.
    fzl::VmoMapper decompressed_mapper;
    if (zx_status_t status =
            decompressed_mapper.Map(decompression_buffer_, 0, mapping.decompressed_length,
                                    ZX_VM_PERM_READ | ZX_VM_PERM_WRITE) != ZX_OK) {
      FX_LOGS(ERROR) << "TransferChunked: Failed to map decompress buffer: "
                     << zx_status_get_string(status);
      return ToPagerErrorStatus(status);
    }
    auto unmap_decompression = fit::defer([&]() { decompressed_mapper.Unmap(); });

    fs::Ticker ticker(metrics_->Collecting());
    size_t decompressed_size = mapping.decompressed_length;
    zx_status_t decompress_status;
    if (decompressor_client_) {
      ZX_DEBUG_ASSERT(sandbox_buffer_.is_valid());
      auto decommit_sandbox = fit::defer([this, length = mapping.decompressed_length]() {
        // Decommit pages in the sandbox buffer that might have been populated. All blobs share
        // the same sandbox buffer - this prevents data leaks between different blobs.
        sandbox_buffer_.op_range(ZX_VMO_OP_DECOMMIT, 0, fbl::round_up(length, kBlobfsBlockSize),
                                 nullptr, 0);
      });
      ExternalSeekableDecompressor decompressor(decompressor_client_.get(),
                                                info.decompressor.get());
      decompress_status = decompressor.DecompressRange(
          offset_of_compressed_data, mapping.compressed_length, mapping.decompressed_length);
      if (decompress_status == ZX_OK) {
        zx_status_t read_status =
            sandbox_buffer_.read(decompressed_mapper.start(), 0, mapping.decompressed_length);
        if (read_status != ZX_OK) {
          FX_LOGS(ERROR) << "TransferChunked: Failed to copy from sandbox buffer: "
                         << zx_status_get_string(read_status);
          return ToPagerErrorStatus(read_status);
        }
      }
    } else {
      // Decompress the data
      uint8_t* src = static_cast<uint8_t*>(compressed_mapper_.start()) + offset_of_compressed_data;
      decompress_status = info.decompressor->DecompressRange(
          decompressed_mapper.start(), &decompressed_size, src, mapping.compressed_length,
          mapping.decompressed_offset);
    }
    if (decompress_status != ZX_OK) {
      FX_LOGS(ERROR) << "TransferChunked: Failed to decompress: "
                     << zx_status_get_string(decompress_status);
      return ToPagerErrorStatus(decompress_status);
    }
    metrics_->paged_read_metrics().IncrementDecompression(CompressionAlgorithm::CHUNKED,
                                                          decompressed_size, ticker.End(),
                                                          decompressor_client_ != nullptr);

    // Verify the decompressed pages.
    const uint64_t rounded_length =
        fbl::round_up<uint64_t, uint64_t>(mapping.decompressed_length, PAGE_SIZE);
    zx_status_t status =
        info.verifier->VerifyPartial(decompressed_mapper.start(), mapping.decompressed_length,
                                     mapping.decompressed_offset, rounded_length);
    if (status != ZX_OK) {
      FX_LOGS(ERROR) << "TransferChunked: Failed to verify data: " << zx_status_get_string(status);
      return ToPagerErrorStatus(status);
    }

    decompressed_mapper.Unmap();

    // Move the pages from the decompression buffer to the destination VMO.
    status = pager_.supply_pages(vmo, mapping.decompressed_offset, rounded_length,
                                 decompression_buffer_, 0);
    if (status != ZX_OK) {
      FX_LOGS(ERROR) << "TransferChunked: Failed to supply pages to paged VMO: "
                     << zx_status_get_string(status);
      return ToPagerErrorStatus(status);
    }
    metrics_->IncrementPageIn(merkle_root_hash, read_offset, read_len);

    // Advance the required decompressed offset based on how much has already been populated.
    current_decompressed_offset = mapping.decompressed_offset + mapping.decompressed_length;
  }

  return PagerErrorStatus::kOK;
}

}  // namespace pager
}  // namespace blobfs