1386 lines · c
1// SPDX-License-Identifier: GPL-2.0-only2/*3 * Copyright 2023 Red Hat4 */5 6 7#include "index.h"8 9#include "logger.h"10#include "memory-alloc.h"11 12#include "funnel-requestqueue.h"13#include "hash-utils.h"14#include "sparse-cache.h"15 16static const u64 NO_LAST_SAVE = U64_MAX;17 18/*19 * When searching for deduplication records, the index first searches the volume index, and then20 * searches the chapter index for the relevant chapter. If the chapter has been fully committed to21 * storage, the chapter pages are loaded into the page cache. If the chapter has not yet been22 * committed (either the open chapter or a recently closed one), the index searches the in-memory23 * representation of the chapter. Finally, if the volume index does not find a record and the index24 * is sparse, the index will search the sparse cache.25 *26 * The index send two kinds of messages to coordinate between zones: chapter close messages for the27 * chapter writer, and sparse cache barrier messages for the sparse cache.28 *29 * The chapter writer is responsible for committing chapters of records to storage. Since zones can30 * get different numbers of records, some zones may fall behind others. Each time a zone fills up31 * its available space in a chapter, it informs the chapter writer that the chapter is complete,32 * and also informs all other zones that it has closed the chapter. Each other zone will then close33 * the chapter immediately, regardless of how full it is, in order to minimize skew between zones.34 * Once every zone has closed the chapter, the chapter writer will commit that chapter to storage.35 *36 * The last zone to close the chapter also removes the oldest chapter from the volume index.37 * Although that chapter is invalid for zones that have moved on, the existence of the open chapter38 * means that those zones will never ask the volume index about it. No zone is allowed to get more39 * than one chapter ahead of any other. If a zone is so far ahead that it tries to close another40 * chapter before the previous one has been closed by all zones, it is forced to wait.41 *42 * The sparse cache relies on having the same set of chapter indexes available to all zones. When a43 * request wants to add a chapter to the sparse cache, it sends a barrier message to each zone44 * during the triage stage that acts as a rendezvous. Once every zone has reached the barrier and45 * paused its operations, the cache membership is changed and each zone is then informed that it46 * can proceed. More details can be found in the sparse cache documentation.47 *48 * If a sparse cache has only one zone, it will not create a triage queue, but it still needs the49 * barrier message to change the sparse cache membership, so the index simulates the message by50 * invoking the handler directly.51 */52 53struct chapter_writer {54 /* The index to which we belong */55 struct uds_index *index;56 /* The thread to do the writing */57 struct thread *thread;58 /* The lock protecting the following fields */59 struct mutex mutex;60 /* The condition signalled on state changes */61 struct cond_var cond;62 /* Set to true to stop the thread */63 bool stop;64 /* The result from the most recent write */65 int result;66 /* The number of bytes allocated by the chapter writer */67 size_t memory_size;68 /* The number of zones which have submitted a chapter for writing */69 unsigned int zones_to_write;70 /* Open chapter index used by uds_close_open_chapter() */71 struct open_chapter_index *open_chapter_index;72 /* Collated records used by uds_close_open_chapter() */73 struct uds_volume_record *collated_records;74 /* The chapters to write (one per zone) */75 struct open_chapter_zone *chapters[];76};77 78static bool is_zone_chapter_sparse(const struct index_zone *zone, u64 virtual_chapter)79{80 return uds_is_chapter_sparse(zone->index->volume->geometry,81 zone->oldest_virtual_chapter,82 zone->newest_virtual_chapter, virtual_chapter);83}84 85static int launch_zone_message(struct uds_zone_message message, unsigned int zone,86 struct uds_index *index)87{88 int result;89 struct uds_request *request;90 91 result = vdo_allocate(1, struct uds_request, __func__, &request);92 if (result != VDO_SUCCESS)93 return result;94 95 request->index = index;96 request->unbatched = true;97 request->zone_number = zone;98 request->zone_message = message;99 100 uds_enqueue_request(request, STAGE_MESSAGE);101 return UDS_SUCCESS;102}103 104static void enqueue_barrier_messages(struct uds_index *index, u64 virtual_chapter)105{106 struct uds_zone_message message = {107 .type = UDS_MESSAGE_SPARSE_CACHE_BARRIER,108 .virtual_chapter = virtual_chapter,109 };110 unsigned int zone;111 112 for (zone = 0; zone < index->zone_count; zone++) {113 int result = launch_zone_message(message, zone, index);114 115 VDO_ASSERT_LOG_ONLY((result == UDS_SUCCESS), "barrier message allocation");116 }117}118 119/*120 * Determine whether this request should trigger a sparse cache barrier message to change the121 * membership of the sparse cache. If a change in membership is desired, the function returns the122 * chapter number to add.123 */124static u64 triage_index_request(struct uds_index *index, struct uds_request *request)125{126 u64 virtual_chapter;127 struct index_zone *zone;128 129 virtual_chapter = uds_lookup_volume_index_name(index->volume_index,130 &request->record_name);131 if (virtual_chapter == NO_CHAPTER)132 return NO_CHAPTER;133 134 zone = index->zones[request->zone_number];135 if (!is_zone_chapter_sparse(zone, virtual_chapter))136 return NO_CHAPTER;137 138 /*139 * FIXME: Optimize for a common case by remembering the chapter from the most recent140 * barrier message and skipping this chapter if is it the same.141 */142 143 return virtual_chapter;144}145 146/*147 * Simulate a message to change the sparse cache membership for a single-zone sparse index. This148 * allows us to forgo the complicated locking required by a multi-zone sparse index. Any other kind149 * of index does nothing here.150 */151static int simulate_index_zone_barrier_message(struct index_zone *zone,152 struct uds_request *request)153{154 u64 sparse_virtual_chapter;155 156 if ((zone->index->zone_count > 1) ||157 !uds_is_sparse_index_geometry(zone->index->volume->geometry))158 return UDS_SUCCESS;159 160 sparse_virtual_chapter = triage_index_request(zone->index, request);161 if (sparse_virtual_chapter == NO_CHAPTER)162 return UDS_SUCCESS;163 164 return uds_update_sparse_cache(zone, sparse_virtual_chapter);165}166 167/* This is the request processing function for the triage queue. */168static void triage_request(struct uds_request *request)169{170 struct uds_index *index = request->index;171 u64 sparse_virtual_chapter = triage_index_request(index, request);172 173 if (sparse_virtual_chapter != NO_CHAPTER)174 enqueue_barrier_messages(index, sparse_virtual_chapter);175 176 uds_enqueue_request(request, STAGE_INDEX);177}178 179static int finish_previous_chapter(struct uds_index *index, u64 current_chapter_number)180{181 int result;182 struct chapter_writer *writer = index->chapter_writer;183 184 mutex_lock(&writer->mutex);185 while (index->newest_virtual_chapter < current_chapter_number)186 uds_wait_cond(&writer->cond, &writer->mutex);187 result = writer->result;188 mutex_unlock(&writer->mutex);189 190 if (result != UDS_SUCCESS)191 return vdo_log_error_strerror(result,192 "Writing of previous open chapter failed");193 194 return UDS_SUCCESS;195}196 197static int swap_open_chapter(struct index_zone *zone)198{199 int result;200 201 result = finish_previous_chapter(zone->index, zone->newest_virtual_chapter);202 if (result != UDS_SUCCESS)203 return result;204 205 swap(zone->open_chapter, zone->writing_chapter);206 return UDS_SUCCESS;207}208 209/*210 * Inform the chapter writer that this zone is done with this chapter. The chapter won't start211 * writing until all zones have closed it.212 */213static unsigned int start_closing_chapter(struct uds_index *index,214 unsigned int zone_number,215 struct open_chapter_zone *chapter)216{217 unsigned int finished_zones;218 struct chapter_writer *writer = index->chapter_writer;219 220 mutex_lock(&writer->mutex);221 finished_zones = ++writer->zones_to_write;222 writer->chapters[zone_number] = chapter;223 uds_broadcast_cond(&writer->cond);224 mutex_unlock(&writer->mutex);225 226 return finished_zones;227}228 229static int announce_chapter_closed(struct index_zone *zone, u64 closed_chapter)230{231 int result;232 unsigned int i;233 struct uds_zone_message zone_message = {234 .type = UDS_MESSAGE_ANNOUNCE_CHAPTER_CLOSED,235 .virtual_chapter = closed_chapter,236 };237 238 for (i = 0; i < zone->index->zone_count; i++) {239 if (zone->id == i)240 continue;241 242 result = launch_zone_message(zone_message, i, zone->index);243 if (result != UDS_SUCCESS)244 return result;245 }246 247 return UDS_SUCCESS;248}249 250static int open_next_chapter(struct index_zone *zone)251{252 int result;253 u64 closed_chapter;254 u64 expiring;255 unsigned int finished_zones;256 u32 expire_chapters;257 258 vdo_log_debug("closing chapter %llu of zone %u after %u entries (%u short)",259 (unsigned long long) zone->newest_virtual_chapter, zone->id,260 zone->open_chapter->size,261 zone->open_chapter->capacity - zone->open_chapter->size);262 263 result = swap_open_chapter(zone);264 if (result != UDS_SUCCESS)265 return result;266 267 closed_chapter = zone->newest_virtual_chapter++;268 uds_set_volume_index_zone_open_chapter(zone->index->volume_index, zone->id,269 zone->newest_virtual_chapter);270 uds_reset_open_chapter(zone->open_chapter);271 272 finished_zones = start_closing_chapter(zone->index, zone->id,273 zone->writing_chapter);274 if ((finished_zones == 1) && (zone->index->zone_count > 1)) {275 result = announce_chapter_closed(zone, closed_chapter);276 if (result != UDS_SUCCESS)277 return result;278 }279 280 expiring = zone->oldest_virtual_chapter;281 expire_chapters = uds_chapters_to_expire(zone->index->volume->geometry,282 zone->newest_virtual_chapter);283 zone->oldest_virtual_chapter += expire_chapters;284 285 if (finished_zones < zone->index->zone_count)286 return UDS_SUCCESS;287 288 while (expire_chapters-- > 0)289 uds_forget_chapter(zone->index->volume, expiring++);290 291 return UDS_SUCCESS;292}293 294static int handle_chapter_closed(struct index_zone *zone, u64 virtual_chapter)295{296 if (zone->newest_virtual_chapter == virtual_chapter)297 return open_next_chapter(zone);298 299 return UDS_SUCCESS;300}301 302static int dispatch_index_zone_control_request(struct uds_request *request)303{304 struct uds_zone_message *message = &request->zone_message;305 struct index_zone *zone = request->index->zones[request->zone_number];306 307 switch (message->type) {308 case UDS_MESSAGE_SPARSE_CACHE_BARRIER:309 return uds_update_sparse_cache(zone, message->virtual_chapter);310 311 case UDS_MESSAGE_ANNOUNCE_CHAPTER_CLOSED:312 return handle_chapter_closed(zone, message->virtual_chapter);313 314 default:315 vdo_log_error("invalid message type: %d", message->type);316 return UDS_INVALID_ARGUMENT;317 }318}319 320static void set_request_location(struct uds_request *request,321 enum uds_index_region new_location)322{323 request->location = new_location;324 request->found = ((new_location == UDS_LOCATION_IN_OPEN_CHAPTER) ||325 (new_location == UDS_LOCATION_IN_DENSE) ||326 (new_location == UDS_LOCATION_IN_SPARSE));327}328 329static void set_chapter_location(struct uds_request *request,330 const struct index_zone *zone, u64 virtual_chapter)331{332 request->found = true;333 if (virtual_chapter == zone->newest_virtual_chapter)334 request->location = UDS_LOCATION_IN_OPEN_CHAPTER;335 else if (is_zone_chapter_sparse(zone, virtual_chapter))336 request->location = UDS_LOCATION_IN_SPARSE;337 else338 request->location = UDS_LOCATION_IN_DENSE;339}340 341static int search_sparse_cache_in_zone(struct index_zone *zone, struct uds_request *request,342 u64 virtual_chapter, bool *found)343{344 int result;345 struct volume *volume;346 u16 record_page_number;347 u32 chapter;348 349 result = uds_search_sparse_cache(zone, &request->record_name, &virtual_chapter,350 &record_page_number);351 if ((result != UDS_SUCCESS) || (virtual_chapter == NO_CHAPTER))352 return result;353 354 request->virtual_chapter = virtual_chapter;355 volume = zone->index->volume;356 chapter = uds_map_to_physical_chapter(volume->geometry, virtual_chapter);357 return uds_search_cached_record_page(volume, request, chapter,358 record_page_number, found);359}360 361static int get_record_from_zone(struct index_zone *zone, struct uds_request *request,362 bool *found)363{364 struct volume *volume;365 366 if (request->location == UDS_LOCATION_RECORD_PAGE_LOOKUP) {367 *found = true;368 return UDS_SUCCESS;369 } else if (request->location == UDS_LOCATION_UNAVAILABLE) {370 *found = false;371 return UDS_SUCCESS;372 }373 374 if (request->virtual_chapter == zone->newest_virtual_chapter) {375 uds_search_open_chapter(zone->open_chapter, &request->record_name,376 &request->old_metadata, found);377 return UDS_SUCCESS;378 }379 380 if ((zone->newest_virtual_chapter > 0) &&381 (request->virtual_chapter == (zone->newest_virtual_chapter - 1)) &&382 (zone->writing_chapter->size > 0)) {383 uds_search_open_chapter(zone->writing_chapter, &request->record_name,384 &request->old_metadata, found);385 return UDS_SUCCESS;386 }387 388 volume = zone->index->volume;389 if (is_zone_chapter_sparse(zone, request->virtual_chapter) &&390 uds_sparse_cache_contains(volume->sparse_cache, request->virtual_chapter,391 request->zone_number))392 return search_sparse_cache_in_zone(zone, request,393 request->virtual_chapter, found);394 395 return uds_search_volume_page_cache(volume, request, found);396}397 398static int put_record_in_zone(struct index_zone *zone, struct uds_request *request,399 const struct uds_record_data *metadata)400{401 unsigned int remaining;402 403 remaining = uds_put_open_chapter(zone->open_chapter, &request->record_name,404 metadata);405 if (remaining == 0)406 return open_next_chapter(zone);407 408 return UDS_SUCCESS;409}410 411static int search_index_zone(struct index_zone *zone, struct uds_request *request)412{413 int result;414 struct volume_index_record record;415 bool overflow_record, found = false;416 struct uds_record_data *metadata;417 u64 chapter;418 419 result = uds_get_volume_index_record(zone->index->volume_index,420 &request->record_name, &record);421 if (result != UDS_SUCCESS)422 return result;423 424 if (record.is_found) {425 if (request->requeued && request->virtual_chapter != record.virtual_chapter)426 set_request_location(request, UDS_LOCATION_UNKNOWN);427 428 request->virtual_chapter = record.virtual_chapter;429 result = get_record_from_zone(zone, request, &found);430 if (result != UDS_SUCCESS)431 return result;432 }433 434 if (found)435 set_chapter_location(request, zone, record.virtual_chapter);436 437 /*438 * If a record has overflowed a chapter index in more than one chapter (or overflowed in439 * one chapter and collided with an existing record), it will exist as a collision record440 * in the volume index, but we won't find it in the volume. This case needs special441 * handling.442 */443 overflow_record = (record.is_found && record.is_collision && !found);444 chapter = zone->newest_virtual_chapter;445 if (found || overflow_record) {446 if ((request->type == UDS_QUERY_NO_UPDATE) ||447 ((request->type == UDS_QUERY) && overflow_record)) {448 /* There is nothing left to do. */449 return UDS_SUCCESS;450 }451 452 if (record.virtual_chapter != chapter) {453 /*454 * Update the volume index to reference the new chapter for the block. If455 * the record had been deleted or dropped from the chapter index, it will456 * be back.457 */458 result = uds_set_volume_index_record_chapter(&record, chapter);459 } else if (request->type != UDS_UPDATE) {460 /* The record is already in the open chapter. */461 return UDS_SUCCESS;462 }463 } else {464 /*465 * The record wasn't in the volume index, so check whether the466 * name is in a cached sparse chapter. If we found the name on467 * a previous search, use that result instead.468 */469 if (request->location == UDS_LOCATION_RECORD_PAGE_LOOKUP) {470 found = true;471 } else if (request->location == UDS_LOCATION_UNAVAILABLE) {472 found = false;473 } else if (uds_is_sparse_index_geometry(zone->index->volume->geometry) &&474 !uds_is_volume_index_sample(zone->index->volume_index,475 &request->record_name)) {476 result = search_sparse_cache_in_zone(zone, request, NO_CHAPTER,477 &found);478 if (result != UDS_SUCCESS)479 return result;480 }481 482 if (found)483 set_request_location(request, UDS_LOCATION_IN_SPARSE);484 485 if ((request->type == UDS_QUERY_NO_UPDATE) ||486 ((request->type == UDS_QUERY) && !found)) {487 /* There is nothing left to do. */488 return UDS_SUCCESS;489 }490 491 /*492 * Add a new entry to the volume index referencing the open chapter. This needs to493 * be done both for new records, and for records from cached sparse chapters.494 */495 result = uds_put_volume_index_record(&record, chapter);496 }497 498 if (result == UDS_OVERFLOW) {499 /*500 * The volume index encountered a delta list overflow. The condition was already501 * logged. We will go on without adding the record to the open chapter.502 */503 return UDS_SUCCESS;504 }505 506 if (result != UDS_SUCCESS)507 return result;508 509 if (!found || (request->type == UDS_UPDATE)) {510 /* This is a new record or we're updating an existing record. */511 metadata = &request->new_metadata;512 } else {513 /* Move the existing record to the open chapter. */514 metadata = &request->old_metadata;515 }516 517 return put_record_in_zone(zone, request, metadata);518}519 520static int remove_from_index_zone(struct index_zone *zone, struct uds_request *request)521{522 int result;523 struct volume_index_record record;524 525 result = uds_get_volume_index_record(zone->index->volume_index,526 &request->record_name, &record);527 if (result != UDS_SUCCESS)528 return result;529 530 if (!record.is_found)531 return UDS_SUCCESS;532 533 /* If the request was requeued, check whether the saved state is still valid. */534 535 if (record.is_collision) {536 set_chapter_location(request, zone, record.virtual_chapter);537 } else {538 /* Non-collision records are hints, so resolve the name in the chapter. */539 bool found;540 541 if (request->requeued && request->virtual_chapter != record.virtual_chapter)542 set_request_location(request, UDS_LOCATION_UNKNOWN);543 544 request->virtual_chapter = record.virtual_chapter;545 result = get_record_from_zone(zone, request, &found);546 if (result != UDS_SUCCESS)547 return result;548 549 if (!found) {550 /* There is no record to remove. */551 return UDS_SUCCESS;552 }553 }554 555 set_chapter_location(request, zone, record.virtual_chapter);556 557 /*558 * Delete the volume index entry for the named record only. Note that a later search might559 * later return stale advice if there is a colliding name in the same chapter, but it's a560 * very rare case (1 in 2^21).561 */562 result = uds_remove_volume_index_record(&record);563 if (result != UDS_SUCCESS)564 return result;565 566 /*567 * If the record is in the open chapter, we must remove it or mark it deleted to avoid568 * trouble if the record is added again later.569 */570 if (request->location == UDS_LOCATION_IN_OPEN_CHAPTER)571 uds_remove_from_open_chapter(zone->open_chapter, &request->record_name);572 573 return UDS_SUCCESS;574}575 576static int dispatch_index_request(struct uds_index *index, struct uds_request *request)577{578 int result;579 struct index_zone *zone = index->zones[request->zone_number];580 581 if (!request->requeued) {582 result = simulate_index_zone_barrier_message(zone, request);583 if (result != UDS_SUCCESS)584 return result;585 }586 587 switch (request->type) {588 case UDS_POST:589 case UDS_UPDATE:590 case UDS_QUERY:591 case UDS_QUERY_NO_UPDATE:592 result = search_index_zone(zone, request);593 break;594 595 case UDS_DELETE:596 result = remove_from_index_zone(zone, request);597 break;598 599 default:600 result = vdo_log_warning_strerror(UDS_INVALID_ARGUMENT,601 "invalid request type: %d",602 request->type);603 break;604 }605 606 return result;607}608 609/* This is the request processing function invoked by each zone's thread. */610static void execute_zone_request(struct uds_request *request)611{612 int result;613 struct uds_index *index = request->index;614 615 if (request->zone_message.type != UDS_MESSAGE_NONE) {616 result = dispatch_index_zone_control_request(request);617 if (result != UDS_SUCCESS) {618 vdo_log_error_strerror(result, "error executing message: %d",619 request->zone_message.type);620 }621 622 /* Once the message is processed it can be freed. */623 vdo_free(vdo_forget(request));624 return;625 }626 627 index->need_to_save = true;628 if (request->requeued && (request->status != UDS_SUCCESS)) {629 set_request_location(request, UDS_LOCATION_UNAVAILABLE);630 index->callback(request);631 return;632 }633 634 result = dispatch_index_request(index, request);635 if (result == UDS_QUEUED) {636 /* The request has been requeued so don't let it complete. */637 return;638 }639 640 if (!request->found)641 set_request_location(request, UDS_LOCATION_UNAVAILABLE);642 643 request->status = result;644 index->callback(request);645}646 647static int initialize_index_queues(struct uds_index *index,648 const struct index_geometry *geometry)649{650 int result;651 unsigned int i;652 653 for (i = 0; i < index->zone_count; i++) {654 result = uds_make_request_queue("indexW", &execute_zone_request,655 &index->zone_queues[i]);656 if (result != UDS_SUCCESS)657 return result;658 }659 660 /* The triage queue is only needed for sparse multi-zone indexes. */661 if ((index->zone_count > 1) && uds_is_sparse_index_geometry(geometry)) {662 result = uds_make_request_queue("triageW", &triage_request,663 &index->triage_queue);664 if (result != UDS_SUCCESS)665 return result;666 }667 668 return UDS_SUCCESS;669}670 671/* This is the driver function for the chapter writer thread. */672static void close_chapters(void *arg)673{674 int result;675 struct chapter_writer *writer = arg;676 struct uds_index *index = writer->index;677 678 vdo_log_debug("chapter writer starting");679 mutex_lock(&writer->mutex);680 for (;;) {681 while (writer->zones_to_write < index->zone_count) {682 if (writer->stop && (writer->zones_to_write == 0)) {683 /*684 * We've been told to stop, and all of the zones are in the same685 * open chapter, so we can exit now.686 */687 mutex_unlock(&writer->mutex);688 vdo_log_debug("chapter writer stopping");689 return;690 }691 uds_wait_cond(&writer->cond, &writer->mutex);692 }693 694 /*695 * Release the lock while closing a chapter. We probably don't need to do this, but696 * it seems safer in principle. It's OK to access the chapter and chapter_number697 * fields without the lock since those aren't allowed to change until we're done.698 */699 mutex_unlock(&writer->mutex);700 701 if (index->has_saved_open_chapter) {702 /*703 * Remove the saved open chapter the first time we close an open chapter704 * after loading from a clean shutdown, or after doing a clean save. The705 * lack of the saved open chapter will indicate that a recovery is706 * necessary.707 */708 index->has_saved_open_chapter = false;709 result = uds_discard_open_chapter(index->layout);710 if (result == UDS_SUCCESS)711 vdo_log_debug("Discarding saved open chapter");712 }713 714 result = uds_close_open_chapter(writer->chapters, index->zone_count,715 index->volume,716 writer->open_chapter_index,717 writer->collated_records,718 index->newest_virtual_chapter);719 720 mutex_lock(&writer->mutex);721 index->newest_virtual_chapter++;722 index->oldest_virtual_chapter +=723 uds_chapters_to_expire(index->volume->geometry,724 index->newest_virtual_chapter);725 writer->result = result;726 writer->zones_to_write = 0;727 uds_broadcast_cond(&writer->cond);728 }729}730 731static void stop_chapter_writer(struct chapter_writer *writer)732{733 struct thread *writer_thread = NULL;734 735 mutex_lock(&writer->mutex);736 if (writer->thread != NULL) {737 writer_thread = writer->thread;738 writer->thread = NULL;739 writer->stop = true;740 uds_broadcast_cond(&writer->cond);741 }742 mutex_unlock(&writer->mutex);743 744 if (writer_thread != NULL)745 vdo_join_threads(writer_thread);746}747 748static void free_chapter_writer(struct chapter_writer *writer)749{750 if (writer == NULL)751 return;752 753 stop_chapter_writer(writer);754 uds_free_open_chapter_index(writer->open_chapter_index);755 vdo_free(writer->collated_records);756 vdo_free(writer);757}758 759static int make_chapter_writer(struct uds_index *index,760 struct chapter_writer **writer_ptr)761{762 int result;763 struct chapter_writer *writer;764 size_t collated_records_size =765 (sizeof(struct uds_volume_record) * index->volume->geometry->records_per_chapter);766 767 result = vdo_allocate_extended(struct chapter_writer, index->zone_count,768 struct open_chapter_zone *, "Chapter Writer",769 &writer);770 if (result != VDO_SUCCESS)771 return result;772 773 writer->index = index;774 mutex_init(&writer->mutex);775 uds_init_cond(&writer->cond);776 777 result = vdo_allocate_cache_aligned(collated_records_size, "collated records",778 &writer->collated_records);779 if (result != VDO_SUCCESS) {780 free_chapter_writer(writer);781 return result;782 }783 784 result = uds_make_open_chapter_index(&writer->open_chapter_index,785 index->volume->geometry,786 index->volume->nonce);787 if (result != UDS_SUCCESS) {788 free_chapter_writer(writer);789 return result;790 }791 792 writer->memory_size = (sizeof(struct chapter_writer) +793 index->zone_count * sizeof(struct open_chapter_zone *) +794 collated_records_size +795 writer->open_chapter_index->memory_size);796 797 result = vdo_create_thread(close_chapters, writer, "writer", &writer->thread);798 if (result != VDO_SUCCESS) {799 free_chapter_writer(writer);800 return result;801 }802 803 *writer_ptr = writer;804 return UDS_SUCCESS;805}806 807static int load_index(struct uds_index *index)808{809 int result;810 u64 last_save_chapter;811 812 result = uds_load_index_state(index->layout, index);813 if (result != UDS_SUCCESS)814 return UDS_INDEX_NOT_SAVED_CLEANLY;815 816 last_save_chapter = ((index->last_save != NO_LAST_SAVE) ? index->last_save : 0);817 818 vdo_log_info("loaded index from chapter %llu through chapter %llu",819 (unsigned long long) index->oldest_virtual_chapter,820 (unsigned long long) last_save_chapter);821 822 return UDS_SUCCESS;823}824 825static int rebuild_index_page_map(struct uds_index *index, u64 vcn)826{827 int result;828 struct delta_index_page *chapter_index_page;829 struct index_geometry *geometry = index->volume->geometry;830 u32 chapter = uds_map_to_physical_chapter(geometry, vcn);831 u32 expected_list_number = 0;832 u32 index_page_number;833 u32 lowest_delta_list;834 u32 highest_delta_list;835 836 for (index_page_number = 0;837 index_page_number < geometry->index_pages_per_chapter;838 index_page_number++) {839 result = uds_get_volume_index_page(index->volume, chapter,840 index_page_number,841 &chapter_index_page);842 if (result != UDS_SUCCESS) {843 return vdo_log_error_strerror(result,844 "failed to read index page %u in chapter %u",845 index_page_number, chapter);846 }847 848 lowest_delta_list = chapter_index_page->lowest_list_number;849 highest_delta_list = chapter_index_page->highest_list_number;850 if (lowest_delta_list != expected_list_number) {851 return vdo_log_error_strerror(UDS_CORRUPT_DATA,852 "chapter %u index page %u is corrupt",853 chapter, index_page_number);854 }855 856 uds_update_index_page_map(index->volume->index_page_map, vcn, chapter,857 index_page_number, highest_delta_list);858 expected_list_number = highest_delta_list + 1;859 }860 861 return UDS_SUCCESS;862}863 864static int replay_record(struct uds_index *index, const struct uds_record_name *name,865 u64 virtual_chapter, bool will_be_sparse_chapter)866{867 int result;868 struct volume_index_record record;869 bool update_record;870 871 if (will_be_sparse_chapter &&872 !uds_is_volume_index_sample(index->volume_index, name)) {873 /*874 * This entry will be in a sparse chapter after the rebuild completes, and it is875 * not a sample, so just skip over it.876 */877 return UDS_SUCCESS;878 }879 880 result = uds_get_volume_index_record(index->volume_index, name, &record);881 if (result != UDS_SUCCESS)882 return result;883 884 if (record.is_found) {885 if (record.is_collision) {886 if (record.virtual_chapter == virtual_chapter) {887 /* The record is already correct. */888 return UDS_SUCCESS;889 }890 891 update_record = true;892 } else if (record.virtual_chapter == virtual_chapter) {893 /*894 * There is a volume index entry pointing to the current chapter, but we895 * don't know if it is for the same name as the one we are currently896 * working on or not. For now, we're just going to assume that it isn't.897 * This will create one extra collision record if there was a deleted898 * record in the current chapter.899 */900 update_record = false;901 } else {902 /*903 * If we're rebuilding, we don't normally want to go to disk to see if the904 * record exists, since we will likely have just read the record from disk905 * (i.e. we know it's there). The exception to this is when we find an906 * entry in the volume index that has a different chapter. In this case, we907 * need to search that chapter to determine if the volume index entry was908 * for the same record or a different one.909 */910 result = uds_search_volume_page_cache_for_rebuild(index->volume,911 name,912 record.virtual_chapter,913 &update_record);914 if (result != UDS_SUCCESS)915 return result;916 }917 } else {918 update_record = false;919 }920 921 if (update_record) {922 /*923 * Update the volume index to reference the new chapter for the block. If the924 * record had been deleted or dropped from the chapter index, it will be back.925 */926 result = uds_set_volume_index_record_chapter(&record, virtual_chapter);927 } else {928 /*929 * Add a new entry to the volume index referencing the open chapter. This should be930 * done regardless of whether we are a brand new record or a sparse record, i.e.931 * one that doesn't exist in the index but does on disk, since for a sparse record,932 * we would want to un-sparsify if it did exist.933 */934 result = uds_put_volume_index_record(&record, virtual_chapter);935 }936 937 if ((result == UDS_DUPLICATE_NAME) || (result == UDS_OVERFLOW)) {938 /* The rebuilt index will lose these records. */939 return UDS_SUCCESS;940 }941 942 return result;943}944 945static bool check_for_suspend(struct uds_index *index)946{947 bool closing;948 949 if (index->load_context == NULL)950 return false;951 952 mutex_lock(&index->load_context->mutex);953 if (index->load_context->status != INDEX_SUSPENDING) {954 mutex_unlock(&index->load_context->mutex);955 return false;956 }957 958 /* Notify that we are suspended and wait for the resume. */959 index->load_context->status = INDEX_SUSPENDED;960 uds_broadcast_cond(&index->load_context->cond);961 962 while ((index->load_context->status != INDEX_OPENING) &&963 (index->load_context->status != INDEX_FREEING))964 uds_wait_cond(&index->load_context->cond, &index->load_context->mutex);965 966 closing = (index->load_context->status == INDEX_FREEING);967 mutex_unlock(&index->load_context->mutex);968 return closing;969}970 971static int replay_chapter(struct uds_index *index, u64 virtual, bool sparse)972{973 int result;974 u32 i;975 u32 j;976 const struct index_geometry *geometry;977 u32 physical_chapter;978 979 if (check_for_suspend(index)) {980 vdo_log_info("Replay interrupted by index shutdown at chapter %llu",981 (unsigned long long) virtual);982 return -EBUSY;983 }984 985 geometry = index->volume->geometry;986 physical_chapter = uds_map_to_physical_chapter(geometry, virtual);987 uds_prefetch_volume_chapter(index->volume, physical_chapter);988 uds_set_volume_index_open_chapter(index->volume_index, virtual);989 990 result = rebuild_index_page_map(index, virtual);991 if (result != UDS_SUCCESS) {992 return vdo_log_error_strerror(result,993 "could not rebuild index page map for chapter %u",994 physical_chapter);995 }996 997 for (i = 0; i < geometry->record_pages_per_chapter; i++) {998 u8 *record_page;999 u32 record_page_number;1000 1001 record_page_number = geometry->index_pages_per_chapter + i;1002 result = uds_get_volume_record_page(index->volume, physical_chapter,1003 record_page_number, &record_page);1004 if (result != UDS_SUCCESS) {1005 return vdo_log_error_strerror(result, "could not get page %d",1006 record_page_number);1007 }1008 1009 for (j = 0; j < geometry->records_per_page; j++) {1010 const u8 *name_bytes;1011 struct uds_record_name name;1012 1013 name_bytes = record_page + (j * BYTES_PER_RECORD);1014 memcpy(&name.name, name_bytes, UDS_RECORD_NAME_SIZE);1015 result = replay_record(index, &name, virtual, sparse);1016 if (result != UDS_SUCCESS)1017 return result;1018 }1019 }1020 1021 return UDS_SUCCESS;1022}1023 1024static int replay_volume(struct uds_index *index)1025{1026 int result;1027 u64 old_map_update;1028 u64 new_map_update;1029 u64 virtual;1030 u64 from_virtual = index->oldest_virtual_chapter;1031 u64 upto_virtual = index->newest_virtual_chapter;1032 bool will_be_sparse;1033 1034 vdo_log_info("Replaying volume from chapter %llu through chapter %llu",1035 (unsigned long long) from_virtual,1036 (unsigned long long) upto_virtual);1037 1038 /*1039 * The index failed to load, so the volume index is empty. Add records to the volume index1040 * in order, skipping non-hooks in chapters which will be sparse to save time.1041 *1042 * Go through each record page of each chapter and add the records back to the volume1043 * index. This should not cause anything to be written to either the open chapter or the1044 * on-disk volume. Also skip the on-disk chapter corresponding to upto_virtual, as this1045 * would have already been purged from the volume index when the chapter was opened.1046 *1047 * Also, go through each index page for each chapter and rebuild the index page map.1048 */1049 old_map_update = index->volume->index_page_map->last_update;1050 for (virtual = from_virtual; virtual < upto_virtual; virtual++) {1051 will_be_sparse = uds_is_chapter_sparse(index->volume->geometry,1052 from_virtual, upto_virtual,1053 virtual);1054 result = replay_chapter(index, virtual, will_be_sparse);1055 if (result != UDS_SUCCESS)1056 return result;1057 }1058 1059 /* Also reap the chapter being replaced by the open chapter. */1060 uds_set_volume_index_open_chapter(index->volume_index, upto_virtual);1061 1062 new_map_update = index->volume->index_page_map->last_update;1063 if (new_map_update != old_map_update) {1064 vdo_log_info("replay changed index page map update from %llu to %llu",1065 (unsigned long long) old_map_update,1066 (unsigned long long) new_map_update);1067 }1068 1069 return UDS_SUCCESS;1070}1071 1072static int rebuild_index(struct uds_index *index)1073{1074 int result;1075 u64 lowest;1076 u64 highest;1077 bool is_empty = false;1078 u32 chapters_per_volume = index->volume->geometry->chapters_per_volume;1079 1080 index->volume->lookup_mode = LOOKUP_FOR_REBUILD;1081 result = uds_find_volume_chapter_boundaries(index->volume, &lowest, &highest,1082 &is_empty);1083 if (result != UDS_SUCCESS) {1084 return vdo_log_fatal_strerror(result,1085 "cannot rebuild index: unknown volume chapter boundaries");1086 }1087 1088 if (is_empty) {1089 index->newest_virtual_chapter = 0;1090 index->oldest_virtual_chapter = 0;1091 index->volume->lookup_mode = LOOKUP_NORMAL;1092 return UDS_SUCCESS;1093 }1094 1095 index->newest_virtual_chapter = highest + 1;1096 index->oldest_virtual_chapter = lowest;1097 if (index->newest_virtual_chapter ==1098 (index->oldest_virtual_chapter + chapters_per_volume)) {1099 /* Skip the chapter shadowed by the open chapter. */1100 index->oldest_virtual_chapter++;1101 }1102 1103 result = replay_volume(index);1104 if (result != UDS_SUCCESS)1105 return result;1106 1107 index->volume->lookup_mode = LOOKUP_NORMAL;1108 return UDS_SUCCESS;1109}1110 1111static void free_index_zone(struct index_zone *zone)1112{1113 if (zone == NULL)1114 return;1115 1116 uds_free_open_chapter(zone->open_chapter);1117 uds_free_open_chapter(zone->writing_chapter);1118 vdo_free(zone);1119}1120 1121static int make_index_zone(struct uds_index *index, unsigned int zone_number)1122{1123 int result;1124 struct index_zone *zone;1125 1126 result = vdo_allocate(1, struct index_zone, "index zone", &zone);1127 if (result != VDO_SUCCESS)1128 return result;1129 1130 result = uds_make_open_chapter(index->volume->geometry, index->zone_count,1131 &zone->open_chapter);1132 if (result != UDS_SUCCESS) {1133 free_index_zone(zone);1134 return result;1135 }1136 1137 result = uds_make_open_chapter(index->volume->geometry, index->zone_count,1138 &zone->writing_chapter);1139 if (result != UDS_SUCCESS) {1140 free_index_zone(zone);1141 return result;1142 }1143 1144 zone->index = index;1145 zone->id = zone_number;1146 index->zones[zone_number] = zone;1147 1148 return UDS_SUCCESS;1149}1150 1151int uds_make_index(struct uds_configuration *config, enum uds_open_index_type open_type,1152 struct index_load_context *load_context, index_callback_fn callback,1153 struct uds_index **new_index)1154{1155 int result;1156 bool loaded = false;1157 bool new = (open_type == UDS_CREATE);1158 struct uds_index *index = NULL;1159 struct index_zone *zone;1160 u64 nonce;1161 unsigned int z;1162 1163 result = vdo_allocate_extended(struct uds_index, config->zone_count,1164 struct uds_request_queue *, "index", &index);1165 if (result != VDO_SUCCESS)1166 return result;1167 1168 index->zone_count = config->zone_count;1169 1170 result = uds_make_index_layout(config, new, &index->layout);1171 if (result != UDS_SUCCESS) {1172 uds_free_index(index);1173 return result;1174 }1175 1176 result = vdo_allocate(index->zone_count, struct index_zone *, "zones",1177 &index->zones);1178 if (result != VDO_SUCCESS) {1179 uds_free_index(index);1180 return result;1181 }1182 1183 result = uds_make_volume(config, index->layout, &index->volume);1184 if (result != UDS_SUCCESS) {1185 uds_free_index(index);1186 return result;1187 }1188 1189 index->volume->lookup_mode = LOOKUP_NORMAL;1190 for (z = 0; z < index->zone_count; z++) {1191 result = make_index_zone(index, z);1192 if (result != UDS_SUCCESS) {1193 uds_free_index(index);1194 return vdo_log_error_strerror(result,1195 "Could not create index zone");1196 }1197 }1198 1199 nonce = uds_get_volume_nonce(index->layout);1200 result = uds_make_volume_index(config, nonce, &index->volume_index);1201 if (result != UDS_SUCCESS) {1202 uds_free_index(index);1203 return vdo_log_error_strerror(result, "could not make volume index");1204 }1205 1206 index->load_context = load_context;1207 index->callback = callback;1208 1209 result = initialize_index_queues(index, config->geometry);1210 if (result != UDS_SUCCESS) {1211 uds_free_index(index);1212 return result;1213 }1214 1215 result = make_chapter_writer(index, &index->chapter_writer);1216 if (result != UDS_SUCCESS) {1217 uds_free_index(index);1218 return result;1219 }1220 1221 if (!new) {1222 result = load_index(index);1223 switch (result) {1224 case UDS_SUCCESS:1225 loaded = true;1226 break;1227 case -ENOMEM:1228 /* We should not try a rebuild for this error. */1229 vdo_log_error_strerror(result, "index could not be loaded");1230 break;1231 default:1232 vdo_log_error_strerror(result, "index could not be loaded");1233 if (open_type == UDS_LOAD) {1234 result = rebuild_index(index);1235 if (result != UDS_SUCCESS) {1236 vdo_log_error_strerror(result,1237 "index could not be rebuilt");1238 }1239 }1240 break;1241 }1242 }1243 1244 if (result != UDS_SUCCESS) {1245 uds_free_index(index);1246 return vdo_log_error_strerror(result, "fatal error in %s()", __func__);1247 }1248 1249 for (z = 0; z < index->zone_count; z++) {1250 zone = index->zones[z];1251 zone->oldest_virtual_chapter = index->oldest_virtual_chapter;1252 zone->newest_virtual_chapter = index->newest_virtual_chapter;1253 }1254 1255 if (index->load_context != NULL) {1256 mutex_lock(&index->load_context->mutex);1257 index->load_context->status = INDEX_READY;1258 /*1259 * If we get here, suspend is meaningless, but notify any thread trying to suspend1260 * us so it doesn't hang.1261 */1262 uds_broadcast_cond(&index->load_context->cond);1263 mutex_unlock(&index->load_context->mutex);1264 }1265 1266 index->has_saved_open_chapter = loaded;1267 index->need_to_save = !loaded;1268 *new_index = index;1269 return UDS_SUCCESS;1270}1271 1272void uds_free_index(struct uds_index *index)1273{1274 unsigned int i;1275 1276 if (index == NULL)1277 return;1278 1279 uds_request_queue_finish(index->triage_queue);1280 for (i = 0; i < index->zone_count; i++)1281 uds_request_queue_finish(index->zone_queues[i]);1282 1283 free_chapter_writer(index->chapter_writer);1284 1285 uds_free_volume_index(index->volume_index);1286 if (index->zones != NULL) {1287 for (i = 0; i < index->zone_count; i++)1288 free_index_zone(index->zones[i]);1289 vdo_free(index->zones);1290 }1291 1292 uds_free_volume(index->volume);1293 uds_free_index_layout(vdo_forget(index->layout));1294 vdo_free(index);1295}1296 1297/* Wait for the chapter writer to complete any outstanding writes. */1298void uds_wait_for_idle_index(struct uds_index *index)1299{1300 struct chapter_writer *writer = index->chapter_writer;1301 1302 mutex_lock(&writer->mutex);1303 while (writer->zones_to_write > 0)1304 uds_wait_cond(&writer->cond, &writer->mutex);1305 mutex_unlock(&writer->mutex);1306}1307 1308/* This function assumes that all requests have been drained. */1309int uds_save_index(struct uds_index *index)1310{1311 int result;1312 1313 if (!index->need_to_save)1314 return UDS_SUCCESS;1315 1316 uds_wait_for_idle_index(index);1317 index->prev_save = index->last_save;1318 index->last_save = ((index->newest_virtual_chapter == 0) ?1319 NO_LAST_SAVE : index->newest_virtual_chapter - 1);1320 vdo_log_info("beginning save (vcn %llu)", (unsigned long long) index->last_save);1321 1322 result = uds_save_index_state(index->layout, index);1323 if (result != UDS_SUCCESS) {1324 vdo_log_info("save index failed");1325 index->last_save = index->prev_save;1326 } else {1327 index->has_saved_open_chapter = true;1328 index->need_to_save = false;1329 vdo_log_info("finished save (vcn %llu)",1330 (unsigned long long) index->last_save);1331 }1332 1333 return result;1334}1335 1336int uds_replace_index_storage(struct uds_index *index, struct block_device *bdev)1337{1338 return uds_replace_volume_storage(index->volume, index->layout, bdev);1339}1340 1341/* Accessing statistics should be safe from any thread. */1342void uds_get_index_stats(struct uds_index *index, struct uds_index_stats *counters)1343{1344 struct volume_index_stats stats;1345 1346 uds_get_volume_index_stats(index->volume_index, &stats);1347 counters->entries_indexed = stats.record_count;1348 counters->collisions = stats.collision_count;1349 counters->entries_discarded = stats.discard_count;1350 1351 counters->memory_used = (index->volume_index->memory_size +1352 index->volume->cache_size +1353 index->chapter_writer->memory_size);1354}1355 1356void uds_enqueue_request(struct uds_request *request, enum request_stage stage)1357{1358 struct uds_index *index = request->index;1359 struct uds_request_queue *queue;1360 1361 switch (stage) {1362 case STAGE_TRIAGE:1363 if (index->triage_queue != NULL) {1364 queue = index->triage_queue;1365 break;1366 }1367 1368 fallthrough;1369 1370 case STAGE_INDEX:1371 request->zone_number =1372 uds_get_volume_index_zone(index->volume_index, &request->record_name);1373 fallthrough;1374 1375 case STAGE_MESSAGE:1376 queue = index->zone_queues[request->zone_number];1377 break;1378 1379 default:1380 VDO_ASSERT_LOG_ONLY(false, "invalid index stage: %d", stage);1381 return;1382 }1383 1384 uds_request_queue_enqueue(queue, request);1385}1386