Method TestWriteError

be/src/runtime/bufferpool/buffer-pool-test.cc:1614–1665

Test that the buffer pool handles a write error correctly. Delete the scratch directory before an operation that would cause a write and test that subsequent API calls return errors as expected.

// directory before an operation that would cause a write and test that subsequent API
// calls return errors as expected.
void BufferPoolTest::TestWriteError(
    int write_delay_ms, const string& compression, bool punch_holes) {
  InitTmpFileMgr(1, compression, punch_holes);
  int MAX_NUM_BUFFERS = 2;
  int64_t TOTAL_MEM = MAX_NUM_BUFFERS * TEST_BUFFER_LEN;
  BufferPool pool(test_env_->metrics(), TEST_BUFFER_LEN, TOTAL_MEM, TOTAL_MEM);
  global_reservations_.InitRootTracker(NewProfile(), TOTAL_MEM);
  ClientHandle client;
  ASSERT_OK(pool.RegisterClient("test client", NewFileGroup(), &global_reservations_,
      nullptr, TOTAL_MEM, NewProfile(), &client));
  ASSERT_TRUE(client.IncreaseReservation(TOTAL_MEM));
  client.impl_->set_debug_write_delay_ms(write_delay_ms);

  vector<PageHandle> pages;
  CreatePages(&pool, &client, TEST_BUFFER_LEN, MAX_NUM_BUFFERS, &pages);
  // Unpin two pages here, to ensure that backing storage is allocated in tmp file.
  UnpinAll(&pool, &client, &pages);
  WaitForAllWrites(&client);
  // Repin the pages
  ASSERT_OK(PinAll(&pool, &client, &pages));
  // Remove permissions to the backing storage so that future writes will fail
  ASSERT_GT(RemoveScratchPerms(test_env_->tmp_file_mgr()->GetTmpDirPath(0)), 0);
  // Give the first write a chance to fail before the second write starts.
  const int INTERVAL_MS = 10;
  UnpinAll(&pool, &client, &pages, INTERVAL_MS);
  WaitForAllWrites(&client);

  // Subsequent calls to APIs that require allocating memory should fail: the write error
  // is picked up asynchronously.
  BufferHandle tmp_buffer;
  PageHandle tmp_page;
  Status error = pool.AllocateBuffer(&client, TEST_BUFFER_LEN, &tmp_buffer);
  EXPECT_EQ(TErrorCode::SCRATCH_ALLOCATION_FAILED, error.code());
  ASSERT_NE(string::npos, error.msg().msg().find(GetBackendString()));
  EXPECT_FALSE(tmp_buffer.is_open());
  error = pool.CreatePage(&client, TEST_BUFFER_LEN, &tmp_page);
  EXPECT_EQ(TErrorCode::SCRATCH_ALLOCATION_FAILED, error.code());
  EXPECT_FALSE(tmp_page.is_open());
  error = pool.Pin(&client, pages.data());
  EXPECT_EQ(TErrorCode::SCRATCH_ALLOCATION_FAILED, error.code());
  EXPECT_FALSE(pages[0].is_pinned());

  // Transferring reservation does not result in an error.
  bool transferred;
  EXPECT_OK(
      client.TransferReservationTo(&global_reservations_, TEST_BUFFER_LEN, &transferred));
  EXPECT_TRUE(transferred);

  DestroyAll(&pool, &client, &pages);
  pool.DeregisterClient(&client);
  global_reservations_.Close();
}

TEST_F(BufferPoolTest, WriteError) {
  TestWriteError(0, "", false);