absl::MutexLock l(&block->mu);
  if (block->state != FetchState::FINISHED) {
    return true;  // No need to check for staleness.
  }
  if (max_staleness_ == 0) return true;  // Not enforcing staleness.
  return timer_seconds_() - block->timestamp <= max_staleness_;
}

std::shared_ptr<RamFileBlockCache::Block> RamFileBlockCache::Lookup(
    const Key& key) {
  absl::MutexLock lock(&mu_);

  env->SetNowSeconds(6);
  EXPECT_FALSE(cache.Lookup(key, &value));
}

TEST(ExpiringLRUCacheTest, MaxEntries) {
  // max_age of 0 means nothing will be cached.
  tf_gcs_filesystem::ExpiringLRUCache<int> cache1(0, 4);
  cache1.Insert("a", 1);
  int value = 0;
  EXPECT_FALSE(cache1.Lookup("a", &value));
  // Now set max_age = 1 and verify the LRU eviction logic.
  tf_gcs_filesystem::ExpiringLRUCache<int> cache2(1, 4);
  cache2.Insert("a", 1);

    EXPECT_EQ(got_n, want_n);
    calls++;
    memset(buffer, 'x', got_n);
    TF_SetStatus(status, TF_OK, "");
    return got_n;
  };
  // If block_size, max_bytes, or both are zero, or want_n is larger than
  // max_bytes the cache is a pass-through.
  tf_gcs_filesystem::RamFileBlockCache cache1(1, 0, 0, fetcher);
  tf_gcs_filesystem::RamFileBlockCache cache2(0, 1, 0, fetcher);

    CHECK_EQ(TF_OK, TF_GetCode(status)) << TF_Message(status);
    TFE_Execute(matmul, &retvals[0], &num_retvals, status);
    CHECK_EQ(TF_OK, TF_GetCode(status)) << TF_Message(status);
    if (state.iterations() >= state.max_iterations && async) {
      TFE_Executor* executor = TFE_ContextGetExecutorForThread(ctx);
      TFE_ExecutorWaitForAllPendingNodes(executor, status);
      ASSERT_EQ(TF_OK, TF_GetCode(status)) << TF_Message(status);

                   std::function<uint64_t()> timer_seconds = TF_NowSeconds)
      : max_age_(max_age),
        max_entries_(max_entries),
        timer_seconds_(timer_seconds) {}

  /// Insert `value` with key `key`. This will replace any previous entry with
  /// the same key.
  void Insert(const std::string& key, const T& value) {
    if (max_age_ == 0) {
      return;
    }
    absl::MutexLock lock(&mu_);

  }
  const char* max_bytes_env = std::getenv(kMaxCacheSize);
  if (max_bytes_env && absl::SimpleAtoi(max_bytes_env, &value)) {
    max_bytes = static_cast<size_t>(value * 1024 * 1024);
  }
  const char* max_staleness_env = std::getenv(kMaxStaleness);
  if (max_staleness_env && absl::SimpleAtoi(max_staleness_env, &value)) {
    max_staleness = value;
  }

      BlockFetcher;

  RamFileBlockCache(size_t block_size, size_t max_bytes, uint64_t max_staleness,
                    BlockFetcher block_fetcher,
                    std::function<uint64_t()> timer_seconds = TF_NowSeconds)
      : block_size_(block_size),
        max_bytes_(max_bytes),
        max_staleness_(max_staleness),
        block_fetcher_(block_fetcher),
        timer_seconds_(timer_seconds),

  int rc = 1;
  // When `sendfile` returns 0 we stop copying and let callers handle this.
  while (offset < size && rc > 0) {
    // Use uint64 for safe compare SSIZE_MAX
    uint64_t chunk = size - offset;
    if (chunk > SSIZE_MAX) chunk = SSIZE_MAX;

    rc = sendfile(dst_fd, src_fd, &offset, static_cast<size_t>(chunk));
    if (rc < 0) return -1;
    bytes_transferred += rc;
  }

  return bytes_transferred;
}

  return {ToOperation(edge->src()), edge->src_output()};
}

void TF_OperationAllInputs(TF_Operation* oper, TF_Output* inputs,
                           int max_inputs) {
  for (auto* edge : oper->node.in_edges()) {
    if (edge->dst_input() >= 0 && edge->dst_input() < max_inputs) {
      inputs[edge->dst_input()] = {ToOperation(edge->src()),
                                   edge->src_output()};
    }
  }
}

                          /*expected_max_outputs=*/1, status.get());
  ASSERT_FALSE(TF_GetCode(status.get()) == TF_OK);
  TF_SetStatus(status.get(), TF_OK, "");

  // Check that ops still run successfully on the device.
  parallel_device.Execute(context.get(), std::vector<ParallelTensor*>(),
                          "VarHandleOp", TFE_OpGetAttrs(handle_op.get()),
                          /*expected_max_outputs=*/1, status.get());