diff --git a/include/storage/mem_index_page.h b/include/storage/mem_index_page.h index 423fbfb3..795b36c9 100644 --- a/include/storage/mem_index_page.h +++ b/include/storage/mem_index_page.h @@ -171,6 +171,16 @@ class MemIndexPage return page_.IsRegistered(); } + void SetError(KvError err) + { + err_ = err; + } + + KvError Error() const + { + return err_; + } + private: /** * @brief The page ID is 0, if the page is newly created in memory and has @@ -188,6 +198,15 @@ class MemIndexPage FilePageId file_page_id_{MaxFilePageId}; Page page_; + /** + * @brief The error of the page. + * If the page is not loaded, the error is KvError::NoError. + * If the page is loaded, the error is the error of loading the page. + * If the page is not KvError::NoError, the page will be freed on no + * reference to it. + */ + KvError err_{KvError::NoError}; + WaitingZone waiting_; /** diff --git a/src/storage/index_page_manager.cpp b/src/storage/index_page_manager.cpp index 185eb63b..1c6eff5a 100644 --- a/src/storage/index_page_manager.cpp +++ b/src/storage/index_page_manager.cpp @@ -119,6 +119,7 @@ MemIndexPage *IndexPageManager::AllocIndexPage() } assert(next_free->IsDetached()); assert(!next_free->IsPinned()); + assert(next_free->Error() == KvError::NoError); next_free->in_free_list_ = false; return next_free; } @@ -127,6 +128,7 @@ void IndexPageManager::FreeIndexPage(MemIndexPage *page) { assert(page->IsDetached()); assert(!page->IsPinned()); + page->SetError(KvError::NoError); page->in_free_list_ = true; free_head_.EnqueNext(page); } @@ -492,9 +494,17 @@ std::pair IndexPageManager::FindPage( new_page->page_ = std::move(page); if (err != KvError::NoError) { - new_page->waiting_.WakeAll(); mapping->Unswizzling(new_page); - FreeIndexPage(new_page); + if (new_page->IsPinned()) + { + new_page->SetError(err); + new_page->waiting_.WakeAll(); + } + else + { + CHECK(new_page->waiting_.Empty()); + FreeIndexPage(new_page); + } return {MemIndexPage::Handle(), err}; } FinishIo(mapping, new_page); @@ -505,6 +515,18 @@ std::pair IndexPageManager::FindPage( { // This page is not loaded yet. handle->waiting_.Wait(ThdTask()); + if (handle->Error() != KvError::NoError) + { + MemIndexPage *page = handle.Get(); + KvError err = page->Error(); + handle.Reset(); + if (!page->IsPinned()) + { + FreeIndexPage(page); + } + return {MemIndexPage::Handle(), err}; + } + assert(handle->IsPinned()); } else {