501 lines · c
1// SPDX-License-Identifier: GPL-2.0-only2/*3 * Copyright 2023 Red Hat4 */5 6#include "vio.h"7 8#include <linux/bio.h>9#include <linux/blkdev.h>10#include <linux/kernel.h>11#include <linux/ratelimit.h>12 13#include "logger.h"14#include "memory-alloc.h"15#include "permassert.h"16 17#include "constants.h"18#include "io-submitter.h"19#include "vdo.h"20 21/* A vio_pool is a collection of preallocated vios. */22struct vio_pool {23 /* The number of objects managed by the pool */24 size_t size;25 /* The list of objects which are available */26 struct list_head available;27 /* The queue of requestors waiting for objects from the pool */28 struct vdo_wait_queue waiting;29 /* The number of objects currently in use */30 size_t busy_count;31 /* The list of objects which are in use */32 struct list_head busy;33 /* The ID of the thread on which this pool may be used */34 thread_id_t thread_id;35 /* The buffer backing the pool's vios */36 char *buffer;37 /* The pool entries */38 struct pooled_vio vios[];39};40 41physical_block_number_t pbn_from_vio_bio(struct bio *bio)42{43 struct vio *vio = bio->bi_private;44 struct vdo *vdo = vio->completion.vdo;45 physical_block_number_t pbn = bio->bi_iter.bi_sector / VDO_SECTORS_PER_BLOCK;46 47 return ((pbn == VDO_GEOMETRY_BLOCK_LOCATION) ? pbn : pbn + vdo->geometry.bio_offset);48}49 50static int create_multi_block_bio(block_count_t size, struct bio **bio_ptr)51{52 struct bio *bio = NULL;53 int result;54 55 result = vdo_allocate_extended(struct bio, size + 1, struct bio_vec,56 "bio", &bio);57 if (result != VDO_SUCCESS)58 return result;59 60 *bio_ptr = bio;61 return VDO_SUCCESS;62}63 64int vdo_create_bio(struct bio **bio_ptr)65{66 return create_multi_block_bio(1, bio_ptr);67}68 69void vdo_free_bio(struct bio *bio)70{71 if (bio == NULL)72 return;73 74 bio_uninit(bio);75 vdo_free(vdo_forget(bio));76}77 78int allocate_vio_components(struct vdo *vdo, enum vio_type vio_type,79 enum vio_priority priority, void *parent,80 unsigned int block_count, char *data, struct vio *vio)81{82 struct bio *bio;83 int result;84 85 result = VDO_ASSERT(block_count <= MAX_BLOCKS_PER_VIO,86 "block count %u does not exceed maximum %u", block_count,87 MAX_BLOCKS_PER_VIO);88 if (result != VDO_SUCCESS)89 return result;90 91 result = VDO_ASSERT(((vio_type != VIO_TYPE_UNINITIALIZED) && (vio_type != VIO_TYPE_DATA)),92 "%d is a metadata type", vio_type);93 if (result != VDO_SUCCESS)94 return result;95 96 result = create_multi_block_bio(block_count, &bio);97 if (result != VDO_SUCCESS)98 return result;99 100 initialize_vio(vio, bio, block_count, vio_type, priority, vdo);101 vio->completion.parent = parent;102 vio->data = data;103 return VDO_SUCCESS;104}105 106/**107 * create_multi_block_metadata_vio() - Create a vio.108 * @vdo: The vdo on which the vio will operate.109 * @vio_type: The type of vio to create.110 * @priority: The relative priority to assign to the vio.111 * @parent: The parent of the vio.112 * @block_count: The size of the vio in blocks.113 * @data: The buffer.114 * @vio_ptr: A pointer to hold the new vio.115 *116 * Return: VDO_SUCCESS or an error.117 */118int create_multi_block_metadata_vio(struct vdo *vdo, enum vio_type vio_type,119 enum vio_priority priority, void *parent,120 unsigned int block_count, char *data,121 struct vio **vio_ptr)122{123 struct vio *vio;124 int result;125 126 BUILD_BUG_ON(sizeof(struct vio) > 256);127 128 /*129 * Metadata vios should use direct allocation and not use the buffer pool, which is130 * reserved for submissions from the linux block layer.131 */132 result = vdo_allocate(1, struct vio, __func__, &vio);133 if (result != VDO_SUCCESS) {134 vdo_log_error("metadata vio allocation failure %d", result);135 return result;136 }137 138 result = allocate_vio_components(vdo, vio_type, priority, parent, block_count,139 data, vio);140 if (result != VDO_SUCCESS) {141 vdo_free(vio);142 return result;143 }144 145 *vio_ptr = vio;146 return VDO_SUCCESS;147}148 149/**150 * free_vio_components() - Free the components of a vio embedded in a larger structure.151 * @vio: The vio to destroy152 */153void free_vio_components(struct vio *vio)154{155 if (vio == NULL)156 return;157 158 BUG_ON(is_data_vio(vio));159 vdo_free_bio(vdo_forget(vio->bio));160}161 162/**163 * free_vio() - Destroy a vio.164 * @vio: The vio to destroy.165 */166void free_vio(struct vio *vio)167{168 free_vio_components(vio);169 vdo_free(vio);170}171 172/* Set bio properties for a VDO read or write. */173void vdo_set_bio_properties(struct bio *bio, struct vio *vio, bio_end_io_t callback,174 blk_opf_t bi_opf, physical_block_number_t pbn)175{176 struct vdo *vdo = vio->completion.vdo;177 struct device_config *config = vdo->device_config;178 179 pbn -= vdo->geometry.bio_offset;180 vio->bio_zone = ((pbn / config->thread_counts.bio_rotation_interval) %181 config->thread_counts.bio_threads);182 183 bio->bi_private = vio;184 bio->bi_end_io = callback;185 bio->bi_opf = bi_opf;186 bio->bi_iter.bi_sector = pbn * VDO_SECTORS_PER_BLOCK;187}188 189/*190 * Prepares the bio to perform IO with the specified buffer. May only be used on a VDO-allocated191 * bio, as it assumes the bio wraps a 4k buffer that is 4k aligned, but there does not have to be a192 * vio associated with the bio.193 */194int vio_reset_bio(struct vio *vio, char *data, bio_end_io_t callback,195 blk_opf_t bi_opf, physical_block_number_t pbn)196{197 int bvec_count, offset, len, i;198 struct bio *bio = vio->bio;199 200 bio_reset(bio, bio->bi_bdev, bi_opf);201 vdo_set_bio_properties(bio, vio, callback, bi_opf, pbn);202 if (data == NULL)203 return VDO_SUCCESS;204 205 bio->bi_io_vec = bio->bi_inline_vecs;206 bio->bi_max_vecs = vio->block_count + 1;207 len = VDO_BLOCK_SIZE * vio->block_count;208 offset = offset_in_page(data);209 bvec_count = DIV_ROUND_UP(offset + len, PAGE_SIZE);210 211 /*212 * If we knew that data was always on one page, or contiguous pages, we wouldn't need the213 * loop. But if we're using vmalloc, it's not impossible that the data is in different214 * pages that can't be merged in bio_add_page...215 */216 for (i = 0; (i < bvec_count) && (len > 0); i++) {217 struct page *page;218 int bytes_added;219 int bytes = PAGE_SIZE - offset;220 221 if (bytes > len)222 bytes = len;223 224 page = is_vmalloc_addr(data) ? vmalloc_to_page(data) : virt_to_page(data);225 bytes_added = bio_add_page(bio, page, bytes, offset);226 227 if (bytes_added != bytes) {228 return vdo_log_error_strerror(VDO_BIO_CREATION_FAILED,229 "Could only add %i bytes to bio",230 bytes_added);231 }232 233 data += bytes;234 len -= bytes;235 offset = 0;236 }237 238 return VDO_SUCCESS;239}240 241/**242 * update_vio_error_stats() - Update per-vio error stats and log the error.243 * @vio: The vio which got an error.244 * @format: The format of the message to log (a printf style format).245 */246void update_vio_error_stats(struct vio *vio, const char *format, ...)247{248 static DEFINE_RATELIMIT_STATE(error_limiter, DEFAULT_RATELIMIT_INTERVAL,249 DEFAULT_RATELIMIT_BURST);250 va_list args;251 int priority;252 struct vdo *vdo = vio->completion.vdo;253 254 switch (vio->completion.result) {255 case VDO_READ_ONLY:256 atomic64_inc(&vdo->stats.read_only_error_count);257 return;258 259 case VDO_NO_SPACE:260 atomic64_inc(&vdo->stats.no_space_error_count);261 priority = VDO_LOG_DEBUG;262 break;263 264 default:265 priority = VDO_LOG_ERR;266 }267 268 if (!__ratelimit(&error_limiter))269 return;270 271 va_start(args, format);272 vdo_vlog_strerror(priority, vio->completion.result, VDO_LOGGING_MODULE_NAME,273 format, args);274 va_end(args);275}276 277void vio_record_metadata_io_error(struct vio *vio)278{279 const char *description;280 physical_block_number_t pbn = pbn_from_vio_bio(vio->bio);281 282 if (bio_op(vio->bio) == REQ_OP_READ) {283 description = "read";284 } else if ((vio->bio->bi_opf & REQ_PREFLUSH) == REQ_PREFLUSH) {285 description = (((vio->bio->bi_opf & REQ_FUA) == REQ_FUA) ?286 "write+preflush+fua" :287 "write+preflush");288 } else if ((vio->bio->bi_opf & REQ_FUA) == REQ_FUA) {289 description = "write+fua";290 } else {291 description = "write";292 }293 294 update_vio_error_stats(vio,295 "Completing %s vio of type %u for physical block %llu with error",296 description, vio->type, (unsigned long long) pbn);297}298 299/**300 * make_vio_pool() - Create a new vio pool.301 * @vdo: The vdo.302 * @pool_size: The number of vios in the pool.303 * @thread_id: The ID of the thread using this pool.304 * @vio_type: The type of vios in the pool.305 * @priority: The priority with which vios from the pool should be enqueued.306 * @context: The context that each entry will have.307 * @pool_ptr: The resulting pool.308 *309 * Return: A success or error code.310 */311int make_vio_pool(struct vdo *vdo, size_t pool_size, thread_id_t thread_id,312 enum vio_type vio_type, enum vio_priority priority, void *context,313 struct vio_pool **pool_ptr)314{315 struct vio_pool *pool;316 char *ptr;317 int result;318 319 result = vdo_allocate_extended(struct vio_pool, pool_size, struct pooled_vio,320 __func__, &pool);321 if (result != VDO_SUCCESS)322 return result;323 324 pool->thread_id = thread_id;325 INIT_LIST_HEAD(&pool->available);326 INIT_LIST_HEAD(&pool->busy);327 328 result = vdo_allocate(pool_size * VDO_BLOCK_SIZE, char,329 "VIO pool buffer", &pool->buffer);330 if (result != VDO_SUCCESS) {331 free_vio_pool(pool);332 return result;333 }334 335 ptr = pool->buffer;336 for (pool->size = 0; pool->size < pool_size; pool->size++, ptr += VDO_BLOCK_SIZE) {337 struct pooled_vio *pooled = &pool->vios[pool->size];338 339 result = allocate_vio_components(vdo, vio_type, priority, NULL, 1, ptr,340 &pooled->vio);341 if (result != VDO_SUCCESS) {342 free_vio_pool(pool);343 return result;344 }345 346 pooled->context = context;347 list_add_tail(&pooled->pool_entry, &pool->available);348 }349 350 *pool_ptr = pool;351 return VDO_SUCCESS;352}353 354/**355 * free_vio_pool() - Destroy a vio pool.356 * @pool: The pool to free.357 */358void free_vio_pool(struct vio_pool *pool)359{360 struct pooled_vio *pooled, *tmp;361 362 if (pool == NULL)363 return;364 365 /* Remove all available vios from the object pool. */366 VDO_ASSERT_LOG_ONLY(!vdo_waitq_has_waiters(&pool->waiting),367 "VIO pool must not have any waiters when being freed");368 VDO_ASSERT_LOG_ONLY((pool->busy_count == 0),369 "VIO pool must not have %zu busy entries when being freed",370 pool->busy_count);371 VDO_ASSERT_LOG_ONLY(list_empty(&pool->busy),372 "VIO pool must not have busy entries when being freed");373 374 list_for_each_entry_safe(pooled, tmp, &pool->available, pool_entry) {375 list_del(&pooled->pool_entry);376 free_vio_components(&pooled->vio);377 pool->size--;378 }379 380 VDO_ASSERT_LOG_ONLY(pool->size == 0,381 "VIO pool must not have missing entries when being freed");382 383 vdo_free(vdo_forget(pool->buffer));384 vdo_free(pool);385}386 387/**388 * is_vio_pool_busy() - Check whether an vio pool has outstanding entries.389 *390 * Return: true if the pool is busy.391 */392bool is_vio_pool_busy(struct vio_pool *pool)393{394 return (pool->busy_count != 0);395}396 397/**398 * acquire_vio_from_pool() - Acquire a vio and buffer from the pool (asynchronous).399 * @pool: The vio pool.400 * @waiter: Object that is requesting a vio.401 */402void acquire_vio_from_pool(struct vio_pool *pool, struct vdo_waiter *waiter)403{404 struct pooled_vio *pooled;405 406 VDO_ASSERT_LOG_ONLY((pool->thread_id == vdo_get_callback_thread_id()),407 "acquire from active vio_pool called from correct thread");408 409 if (list_empty(&pool->available)) {410 vdo_waitq_enqueue_waiter(&pool->waiting, waiter);411 return;412 }413 414 pooled = list_first_entry(&pool->available, struct pooled_vio, pool_entry);415 pool->busy_count++;416 list_move_tail(&pooled->pool_entry, &pool->busy);417 (*waiter->callback)(waiter, pooled);418}419 420/**421 * return_vio_to_pool() - Return a vio to the pool422 * @pool: The vio pool.423 * @vio: The pooled vio to return.424 */425void return_vio_to_pool(struct vio_pool *pool, struct pooled_vio *vio)426{427 VDO_ASSERT_LOG_ONLY((pool->thread_id == vdo_get_callback_thread_id()),428 "vio pool entry returned on same thread as it was acquired");429 430 vio->vio.completion.error_handler = NULL;431 vio->vio.completion.parent = NULL;432 if (vdo_waitq_has_waiters(&pool->waiting)) {433 vdo_waitq_notify_next_waiter(&pool->waiting, NULL, vio);434 return;435 }436 437 list_move_tail(&vio->pool_entry, &pool->available);438 --pool->busy_count;439}440 441/*442 * Various counting functions for statistics.443 * These are used for bios coming into VDO, as well as bios generated by VDO.444 */445void vdo_count_bios(struct atomic_bio_stats *bio_stats, struct bio *bio)446{447 if (((bio->bi_opf & REQ_PREFLUSH) != 0) && (bio->bi_iter.bi_size == 0)) {448 atomic64_inc(&bio_stats->empty_flush);449 atomic64_inc(&bio_stats->flush);450 return;451 }452 453 switch (bio_op(bio)) {454 case REQ_OP_WRITE:455 atomic64_inc(&bio_stats->write);456 break;457 case REQ_OP_READ:458 atomic64_inc(&bio_stats->read);459 break;460 case REQ_OP_DISCARD:461 atomic64_inc(&bio_stats->discard);462 break;463 /*464 * All other operations are filtered out in dmvdo.c, or not created by VDO, so465 * shouldn't exist.466 */467 default:468 VDO_ASSERT_LOG_ONLY(0, "Bio operation %d not a write, read, discard, or empty flush",469 bio_op(bio));470 }471 472 if ((bio->bi_opf & REQ_PREFLUSH) != 0)473 atomic64_inc(&bio_stats->flush);474 if (bio->bi_opf & REQ_FUA)475 atomic64_inc(&bio_stats->fua);476}477 478static void count_all_bios_completed(struct vio *vio, struct bio *bio)479{480 struct atomic_statistics *stats = &vio->completion.vdo->stats;481 482 if (is_data_vio(vio)) {483 vdo_count_bios(&stats->bios_out_completed, bio);484 return;485 }486 487 vdo_count_bios(&stats->bios_meta_completed, bio);488 if (vio->type == VIO_TYPE_RECOVERY_JOURNAL)489 vdo_count_bios(&stats->bios_journal_completed, bio);490 else if (vio->type == VIO_TYPE_BLOCK_MAP)491 vdo_count_bios(&stats->bios_page_cache_completed, bio);492}493 494void vdo_count_completed_bios(struct bio *bio)495{496 struct vio *vio = (struct vio *) bio->bi_private;497 498 atomic64_inc(&vio->completion.vdo->stats.bios_completed);499 count_all_bios_completed(vio, bio);500}501