1 /*- 2 * BSD LICENSE 3 * 4 * Copyright (c) Intel Corporation. 5 * All rights reserved. 6 * 7 * Redistribution and use in source and binary forms, with or without 8 * modification, are permitted provided that the following conditions 9 * are met: 10 * 11 * * Redistributions of source code must retain the above copyright 12 * notice, this list of conditions and the following disclaimer. 13 * * Redistributions in binary form must reproduce the above copyright 14 * notice, this list of conditions and the following disclaimer in 15 * the documentation and/or other materials provided with the 16 * distribution. 17 * * Neither the name of Intel Corporation nor the names of its 18 * contributors may be used to endorse or promote products derived 19 * from this software without specific prior written permission. 20 * 21 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 22 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 23 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR 24 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT 25 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 26 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT 27 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, 28 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 29 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT 30 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 31 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 32 */ 33 34 #include "spdk/stdinc.h" 35 36 #include "spdk/blob.h" 37 #include "spdk/crc32.h" 38 #include "spdk/env.h" 39 #include "spdk/queue.h" 40 #include "spdk/thread.h" 41 #include "spdk/bit_array.h" 42 #include "spdk/likely.h" 43 #include "spdk/util.h" 44 #include "spdk/string.h" 45 46 #include "spdk_internal/assert.h" 47 #include "spdk_internal/log.h" 48 49 #include "blobstore.h" 50 51 #define BLOB_CRC32C_INITIAL 0xffffffffUL 52 53 static int spdk_bs_register_md_thread(struct spdk_blob_store *bs); 54 static int spdk_bs_unregister_md_thread(struct spdk_blob_store *bs); 55 static void _spdk_blob_close_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno); 56 static void _spdk_blob_insert_cluster_on_md_thread(struct spdk_blob *blob, uint32_t cluster_num, 57 uint64_t cluster, spdk_blob_op_complete cb_fn, void *cb_arg); 58 59 static int _spdk_blob_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 60 uint16_t value_len, bool internal); 61 static int _spdk_blob_get_xattr_value(struct spdk_blob *blob, const char *name, 62 const void **value, size_t *value_len, bool internal); 63 static int _spdk_blob_remove_xattr(struct spdk_blob *blob, const char *name, bool internal); 64 65 static void 66 _spdk_blob_verify_md_op(struct spdk_blob *blob) 67 { 68 assert(blob != NULL); 69 assert(spdk_get_thread() == blob->bs->md_thread); 70 assert(blob->state != SPDK_BLOB_STATE_LOADING); 71 } 72 73 static struct spdk_blob_list * 74 _spdk_bs_get_snapshot_entry(struct spdk_blob_store *bs, spdk_blob_id blobid) 75 { 76 struct spdk_blob_list *snapshot_entry = NULL; 77 78 TAILQ_FOREACH(snapshot_entry, &bs->snapshots, link) { 79 if (snapshot_entry->id == blobid) { 80 break; 81 } 82 } 83 84 return snapshot_entry; 85 } 86 87 static void 88 _spdk_bs_claim_cluster(struct spdk_blob_store *bs, uint32_t cluster_num) 89 { 90 assert(cluster_num < spdk_bit_array_capacity(bs->used_clusters)); 91 assert(spdk_bit_array_get(bs->used_clusters, cluster_num) == false); 92 assert(bs->num_free_clusters > 0); 93 94 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Claiming cluster %u\n", cluster_num); 95 96 spdk_bit_array_set(bs->used_clusters, cluster_num); 97 bs->num_free_clusters--; 98 } 99 100 static int 101 _spdk_blob_insert_cluster(struct spdk_blob *blob, uint32_t cluster_num, uint64_t cluster) 102 { 103 uint64_t *cluster_lba = &blob->active.clusters[cluster_num]; 104 105 _spdk_blob_verify_md_op(blob); 106 107 if (*cluster_lba != 0) { 108 return -EEXIST; 109 } 110 111 *cluster_lba = _spdk_bs_cluster_to_lba(blob->bs, cluster); 112 return 0; 113 } 114 115 static int 116 _spdk_bs_allocate_cluster(struct spdk_blob *blob, uint32_t cluster_num, 117 uint64_t *lowest_free_cluster, bool update_map) 118 { 119 pthread_mutex_lock(&blob->bs->used_clusters_mutex); 120 *lowest_free_cluster = spdk_bit_array_find_first_clear(blob->bs->used_clusters, 121 *lowest_free_cluster); 122 if (*lowest_free_cluster == UINT32_MAX) { 123 /* No more free clusters. Cannot satisfy the request */ 124 pthread_mutex_unlock(&blob->bs->used_clusters_mutex); 125 return -ENOSPC; 126 } 127 128 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Claiming cluster %lu for blob %lu\n", *lowest_free_cluster, blob->id); 129 _spdk_bs_claim_cluster(blob->bs, *lowest_free_cluster); 130 pthread_mutex_unlock(&blob->bs->used_clusters_mutex); 131 132 if (update_map) { 133 _spdk_blob_insert_cluster(blob, cluster_num, *lowest_free_cluster); 134 } 135 136 return 0; 137 } 138 139 static void 140 _spdk_bs_release_cluster(struct spdk_blob_store *bs, uint32_t cluster_num) 141 { 142 assert(cluster_num < spdk_bit_array_capacity(bs->used_clusters)); 143 assert(spdk_bit_array_get(bs->used_clusters, cluster_num) == true); 144 assert(bs->num_free_clusters < bs->total_clusters); 145 146 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Releasing cluster %u\n", cluster_num); 147 148 pthread_mutex_lock(&bs->used_clusters_mutex); 149 spdk_bit_array_clear(bs->used_clusters, cluster_num); 150 bs->num_free_clusters++; 151 pthread_mutex_unlock(&bs->used_clusters_mutex); 152 } 153 154 static void 155 _spdk_blob_xattrs_init(struct spdk_blob_xattr_opts *xattrs) 156 { 157 xattrs->count = 0; 158 xattrs->names = NULL; 159 xattrs->ctx = NULL; 160 xattrs->get_value = NULL; 161 } 162 163 void 164 spdk_blob_opts_init(struct spdk_blob_opts *opts) 165 { 166 opts->num_clusters = 0; 167 opts->thin_provision = false; 168 _spdk_blob_xattrs_init(&opts->xattrs); 169 } 170 171 void 172 spdk_blob_open_opts_init(struct spdk_blob_open_opts *opts) 173 { 174 opts->clear_method = BLOB_CLEAR_WITH_DEFAULT; 175 } 176 177 static struct spdk_blob * 178 _spdk_blob_alloc(struct spdk_blob_store *bs, spdk_blob_id id) 179 { 180 struct spdk_blob *blob; 181 182 blob = calloc(1, sizeof(*blob)); 183 if (!blob) { 184 return NULL; 185 } 186 187 blob->id = id; 188 blob->bs = bs; 189 190 blob->parent_id = SPDK_BLOBID_INVALID; 191 192 blob->state = SPDK_BLOB_STATE_DIRTY; 193 blob->active.num_pages = 1; 194 blob->active.pages = calloc(1, sizeof(*blob->active.pages)); 195 if (!blob->active.pages) { 196 free(blob); 197 return NULL; 198 } 199 200 blob->active.pages[0] = _spdk_bs_blobid_to_page(id); 201 202 TAILQ_INIT(&blob->xattrs); 203 TAILQ_INIT(&blob->xattrs_internal); 204 205 return blob; 206 } 207 208 static void 209 _spdk_xattrs_free(struct spdk_xattr_tailq *xattrs) 210 { 211 struct spdk_xattr *xattr, *xattr_tmp; 212 213 TAILQ_FOREACH_SAFE(xattr, xattrs, link, xattr_tmp) { 214 TAILQ_REMOVE(xattrs, xattr, link); 215 free(xattr->name); 216 free(xattr->value); 217 free(xattr); 218 } 219 } 220 221 static void 222 _spdk_blob_free(struct spdk_blob *blob) 223 { 224 assert(blob != NULL); 225 226 free(blob->active.clusters); 227 free(blob->clean.clusters); 228 free(blob->active.pages); 229 free(blob->clean.pages); 230 231 _spdk_xattrs_free(&blob->xattrs); 232 _spdk_xattrs_free(&blob->xattrs_internal); 233 234 if (blob->back_bs_dev) { 235 blob->back_bs_dev->destroy(blob->back_bs_dev); 236 } 237 238 free(blob); 239 } 240 241 struct freeze_io_ctx { 242 struct spdk_bs_cpl cpl; 243 struct spdk_blob *blob; 244 }; 245 246 static void 247 _spdk_blob_io_sync(struct spdk_io_channel_iter *i) 248 { 249 spdk_for_each_channel_continue(i, 0); 250 } 251 252 static void 253 _spdk_blob_execute_queued_io(struct spdk_io_channel_iter *i) 254 { 255 struct spdk_io_channel *_ch = spdk_io_channel_iter_get_channel(i); 256 struct spdk_bs_channel *ch = spdk_io_channel_get_ctx(_ch); 257 struct freeze_io_ctx *ctx = spdk_io_channel_iter_get_ctx(i); 258 struct spdk_bs_request_set *set; 259 struct spdk_bs_user_op_args *args; 260 spdk_bs_user_op_t *op, *tmp; 261 262 TAILQ_FOREACH_SAFE(op, &ch->queued_io, link, tmp) { 263 set = (struct spdk_bs_request_set *)op; 264 args = &set->u.user_op; 265 266 if (args->blob == ctx->blob) { 267 TAILQ_REMOVE(&ch->queued_io, op, link); 268 spdk_bs_user_op_execute(op); 269 } 270 } 271 272 spdk_for_each_channel_continue(i, 0); 273 } 274 275 static void 276 _spdk_blob_io_cpl(struct spdk_io_channel_iter *i, int status) 277 { 278 struct freeze_io_ctx *ctx = spdk_io_channel_iter_get_ctx(i); 279 280 ctx->cpl.u.blob_basic.cb_fn(ctx->cpl.u.blob_basic.cb_arg, 0); 281 282 free(ctx); 283 } 284 285 static void 286 _spdk_blob_freeze_io(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 287 { 288 struct freeze_io_ctx *ctx; 289 290 ctx = calloc(1, sizeof(*ctx)); 291 if (!ctx) { 292 cb_fn(cb_arg, -ENOMEM); 293 return; 294 } 295 296 ctx->cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 297 ctx->cpl.u.blob_basic.cb_fn = cb_fn; 298 ctx->cpl.u.blob_basic.cb_arg = cb_arg; 299 ctx->blob = blob; 300 301 /* Freeze I/O on blob */ 302 blob->frozen_refcnt++; 303 304 if (blob->frozen_refcnt == 1) { 305 spdk_for_each_channel(blob->bs, _spdk_blob_io_sync, ctx, _spdk_blob_io_cpl); 306 } else { 307 cb_fn(cb_arg, 0); 308 free(ctx); 309 } 310 } 311 312 static void 313 _spdk_blob_unfreeze_io(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 314 { 315 struct freeze_io_ctx *ctx; 316 317 ctx = calloc(1, sizeof(*ctx)); 318 if (!ctx) { 319 cb_fn(cb_arg, -ENOMEM); 320 return; 321 } 322 323 ctx->cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 324 ctx->cpl.u.blob_basic.cb_fn = cb_fn; 325 ctx->cpl.u.blob_basic.cb_arg = cb_arg; 326 ctx->blob = blob; 327 328 assert(blob->frozen_refcnt > 0); 329 330 blob->frozen_refcnt--; 331 332 if (blob->frozen_refcnt == 0) { 333 spdk_for_each_channel(blob->bs, _spdk_blob_execute_queued_io, ctx, _spdk_blob_io_cpl); 334 } else { 335 cb_fn(cb_arg, 0); 336 free(ctx); 337 } 338 } 339 340 static int 341 _spdk_blob_mark_clean(struct spdk_blob *blob) 342 { 343 uint64_t *clusters = NULL; 344 uint32_t *pages = NULL; 345 346 assert(blob != NULL); 347 348 if (blob->active.num_clusters) { 349 assert(blob->active.clusters); 350 clusters = calloc(blob->active.num_clusters, sizeof(*blob->active.clusters)); 351 if (!clusters) { 352 return -ENOMEM; 353 } 354 memcpy(clusters, blob->active.clusters, blob->active.num_clusters * sizeof(*blob->active.clusters)); 355 } 356 357 if (blob->active.num_pages) { 358 assert(blob->active.pages); 359 pages = calloc(blob->active.num_pages, sizeof(*blob->active.pages)); 360 if (!pages) { 361 free(clusters); 362 return -ENOMEM; 363 } 364 memcpy(pages, blob->active.pages, blob->active.num_pages * sizeof(*blob->active.pages)); 365 } 366 367 free(blob->clean.clusters); 368 free(blob->clean.pages); 369 370 blob->clean.num_clusters = blob->active.num_clusters; 371 blob->clean.clusters = blob->active.clusters; 372 blob->clean.num_pages = blob->active.num_pages; 373 blob->clean.pages = blob->active.pages; 374 375 blob->active.clusters = clusters; 376 blob->active.pages = pages; 377 378 /* If the metadata was dirtied again while the metadata was being written to disk, 379 * we do not want to revert the DIRTY state back to CLEAN here. 380 */ 381 if (blob->state == SPDK_BLOB_STATE_LOADING) { 382 blob->state = SPDK_BLOB_STATE_CLEAN; 383 } 384 385 return 0; 386 } 387 388 static int 389 _spdk_blob_deserialize_xattr(struct spdk_blob *blob, 390 struct spdk_blob_md_descriptor_xattr *desc_xattr, bool internal) 391 { 392 struct spdk_xattr *xattr; 393 394 if (desc_xattr->length != sizeof(desc_xattr->name_length) + 395 sizeof(desc_xattr->value_length) + 396 desc_xattr->name_length + desc_xattr->value_length) { 397 return -EINVAL; 398 } 399 400 xattr = calloc(1, sizeof(*xattr)); 401 if (xattr == NULL) { 402 return -ENOMEM; 403 } 404 405 xattr->name = malloc(desc_xattr->name_length + 1); 406 if (xattr->name == NULL) { 407 free(xattr); 408 return -ENOMEM; 409 } 410 memcpy(xattr->name, desc_xattr->name, desc_xattr->name_length); 411 xattr->name[desc_xattr->name_length] = '\0'; 412 413 xattr->value = malloc(desc_xattr->value_length); 414 if (xattr->value == NULL) { 415 free(xattr->name); 416 free(xattr); 417 return -ENOMEM; 418 } 419 xattr->value_len = desc_xattr->value_length; 420 memcpy(xattr->value, 421 (void *)((uintptr_t)desc_xattr->name + desc_xattr->name_length), 422 desc_xattr->value_length); 423 424 TAILQ_INSERT_TAIL(internal ? &blob->xattrs_internal : &blob->xattrs, xattr, link); 425 426 return 0; 427 } 428 429 430 static int 431 _spdk_blob_parse_page(const struct spdk_blob_md_page *page, struct spdk_blob *blob) 432 { 433 struct spdk_blob_md_descriptor *desc; 434 size_t cur_desc = 0; 435 void *tmp; 436 437 desc = (struct spdk_blob_md_descriptor *)page->descriptors; 438 while (cur_desc < sizeof(page->descriptors)) { 439 if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_PADDING) { 440 if (desc->length == 0) { 441 /* If padding and length are 0, this terminates the page */ 442 break; 443 } 444 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_FLAGS) { 445 struct spdk_blob_md_descriptor_flags *desc_flags; 446 447 desc_flags = (struct spdk_blob_md_descriptor_flags *)desc; 448 449 if (desc_flags->length != sizeof(*desc_flags) - sizeof(*desc)) { 450 return -EINVAL; 451 } 452 453 if ((desc_flags->invalid_flags | SPDK_BLOB_INVALID_FLAGS_MASK) != 454 SPDK_BLOB_INVALID_FLAGS_MASK) { 455 return -EINVAL; 456 } 457 458 if ((desc_flags->data_ro_flags | SPDK_BLOB_DATA_RO_FLAGS_MASK) != 459 SPDK_BLOB_DATA_RO_FLAGS_MASK) { 460 blob->data_ro = true; 461 blob->md_ro = true; 462 } 463 464 if ((desc_flags->md_ro_flags | SPDK_BLOB_MD_RO_FLAGS_MASK) != 465 SPDK_BLOB_MD_RO_FLAGS_MASK) { 466 blob->md_ro = true; 467 } 468 469 if ((desc_flags->data_ro_flags & SPDK_BLOB_READ_ONLY)) { 470 blob->data_ro = true; 471 blob->md_ro = true; 472 } 473 474 blob->invalid_flags = desc_flags->invalid_flags; 475 blob->data_ro_flags = desc_flags->data_ro_flags; 476 blob->md_ro_flags = desc_flags->md_ro_flags; 477 478 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_EXTENT_RLE) { 479 struct spdk_blob_md_descriptor_extent_rle *desc_extent_rle; 480 unsigned int i, j; 481 unsigned int cluster_count = blob->active.num_clusters; 482 483 desc_extent_rle = (struct spdk_blob_md_descriptor_extent_rle *)desc; 484 485 if (desc_extent_rle->length == 0 || 486 (desc_extent_rle->length % sizeof(desc_extent_rle->extents[0]) != 0)) { 487 return -EINVAL; 488 } 489 490 for (i = 0; i < desc_extent_rle->length / sizeof(desc_extent_rle->extents[0]); i++) { 491 for (j = 0; j < desc_extent_rle->extents[i].length; j++) { 492 if (desc_extent_rle->extents[i].cluster_idx != 0) { 493 if (!spdk_bit_array_get(blob->bs->used_clusters, 494 desc_extent_rle->extents[i].cluster_idx + j)) { 495 return -EINVAL; 496 } 497 } 498 cluster_count++; 499 } 500 } 501 502 if (cluster_count == 0) { 503 return -EINVAL; 504 } 505 tmp = realloc(blob->active.clusters, cluster_count * sizeof(*blob->active.clusters)); 506 if (tmp == NULL) { 507 return -ENOMEM; 508 } 509 blob->active.clusters = tmp; 510 blob->active.cluster_array_size = cluster_count; 511 512 for (i = 0; i < desc_extent_rle->length / sizeof(desc_extent_rle->extents[0]); i++) { 513 for (j = 0; j < desc_extent_rle->extents[i].length; j++) { 514 if (desc_extent_rle->extents[i].cluster_idx != 0) { 515 blob->active.clusters[blob->active.num_clusters++] = _spdk_bs_cluster_to_lba(blob->bs, 516 desc_extent_rle->extents[i].cluster_idx + j); 517 } else if (spdk_blob_is_thin_provisioned(blob)) { 518 blob->active.clusters[blob->active.num_clusters++] = 0; 519 } else { 520 return -EINVAL; 521 } 522 } 523 } 524 525 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR) { 526 int rc; 527 528 rc = _spdk_blob_deserialize_xattr(blob, 529 (struct spdk_blob_md_descriptor_xattr *) desc, false); 530 if (rc != 0) { 531 return rc; 532 } 533 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR_INTERNAL) { 534 int rc; 535 536 rc = _spdk_blob_deserialize_xattr(blob, 537 (struct spdk_blob_md_descriptor_xattr *) desc, true); 538 if (rc != 0) { 539 return rc; 540 } 541 } else { 542 /* Unrecognized descriptor type. Do not fail - just continue to the 543 * next descriptor. If this descriptor is associated with some feature 544 * defined in a newer version of blobstore, that version of blobstore 545 * should create and set an associated feature flag to specify if this 546 * blob can be loaded or not. 547 */ 548 } 549 550 /* Advance to the next descriptor */ 551 cur_desc += sizeof(*desc) + desc->length; 552 if (cur_desc + sizeof(*desc) > sizeof(page->descriptors)) { 553 break; 554 } 555 desc = (struct spdk_blob_md_descriptor *)((uintptr_t)page->descriptors + cur_desc); 556 } 557 558 return 0; 559 } 560 561 static int 562 _spdk_blob_parse(const struct spdk_blob_md_page *pages, uint32_t page_count, 563 struct spdk_blob *blob) 564 { 565 const struct spdk_blob_md_page *page; 566 uint32_t i; 567 int rc; 568 569 assert(page_count > 0); 570 assert(pages[0].sequence_num == 0); 571 assert(blob != NULL); 572 assert(blob->state == SPDK_BLOB_STATE_LOADING); 573 assert(blob->active.clusters == NULL); 574 575 /* The blobid provided doesn't match what's in the MD, this can 576 * happen for example if a bogus blobid is passed in through open. 577 */ 578 if (blob->id != pages[0].id) { 579 SPDK_ERRLOG("Blobid (%lu) doesn't match what's in metadata (%lu)\n", 580 blob->id, pages[0].id); 581 return -ENOENT; 582 } 583 584 for (i = 0; i < page_count; i++) { 585 page = &pages[i]; 586 587 assert(page->id == blob->id); 588 assert(page->sequence_num == i); 589 590 rc = _spdk_blob_parse_page(page, blob); 591 if (rc != 0) { 592 return rc; 593 } 594 } 595 596 return 0; 597 } 598 599 static int 600 _spdk_blob_serialize_add_page(const struct spdk_blob *blob, 601 struct spdk_blob_md_page **pages, 602 uint32_t *page_count, 603 struct spdk_blob_md_page **last_page) 604 { 605 struct spdk_blob_md_page *page; 606 607 assert(pages != NULL); 608 assert(page_count != NULL); 609 610 if (*page_count == 0) { 611 assert(*pages == NULL); 612 *page_count = 1; 613 *pages = spdk_malloc(SPDK_BS_PAGE_SIZE, SPDK_BS_PAGE_SIZE, 614 NULL, SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 615 } else { 616 assert(*pages != NULL); 617 (*page_count)++; 618 *pages = spdk_realloc(*pages, 619 SPDK_BS_PAGE_SIZE * (*page_count), 620 SPDK_BS_PAGE_SIZE); 621 } 622 623 if (*pages == NULL) { 624 *page_count = 0; 625 *last_page = NULL; 626 return -ENOMEM; 627 } 628 629 page = &(*pages)[*page_count - 1]; 630 memset(page, 0, sizeof(*page)); 631 page->id = blob->id; 632 page->sequence_num = *page_count - 1; 633 page->next = SPDK_INVALID_MD_PAGE; 634 *last_page = page; 635 636 return 0; 637 } 638 639 /* Transform the in-memory representation 'xattr' into an on-disk xattr descriptor. 640 * Update required_sz on both success and failure. 641 * 642 */ 643 static int 644 _spdk_blob_serialize_xattr(const struct spdk_xattr *xattr, 645 uint8_t *buf, size_t buf_sz, 646 size_t *required_sz, bool internal) 647 { 648 struct spdk_blob_md_descriptor_xattr *desc; 649 650 *required_sz = sizeof(struct spdk_blob_md_descriptor_xattr) + 651 strlen(xattr->name) + 652 xattr->value_len; 653 654 if (buf_sz < *required_sz) { 655 return -1; 656 } 657 658 desc = (struct spdk_blob_md_descriptor_xattr *)buf; 659 660 desc->type = internal ? SPDK_MD_DESCRIPTOR_TYPE_XATTR_INTERNAL : SPDK_MD_DESCRIPTOR_TYPE_XATTR; 661 desc->length = sizeof(desc->name_length) + 662 sizeof(desc->value_length) + 663 strlen(xattr->name) + 664 xattr->value_len; 665 desc->name_length = strlen(xattr->name); 666 desc->value_length = xattr->value_len; 667 668 memcpy(desc->name, xattr->name, desc->name_length); 669 memcpy((void *)((uintptr_t)desc->name + desc->name_length), 670 xattr->value, 671 desc->value_length); 672 673 return 0; 674 } 675 676 static void 677 _spdk_blob_serialize_extent_rle(const struct spdk_blob *blob, 678 uint64_t start_cluster, uint64_t *next_cluster, 679 uint8_t **buf, size_t *buf_sz) 680 { 681 struct spdk_blob_md_descriptor_extent_rle *desc_extent_rle; 682 size_t cur_sz; 683 uint64_t i, extent_idx; 684 uint64_t lba, lba_per_cluster, lba_count; 685 686 /* The buffer must have room for at least one extent */ 687 cur_sz = sizeof(struct spdk_blob_md_descriptor) + sizeof(desc_extent_rle->extents[0]); 688 if (*buf_sz < cur_sz) { 689 *next_cluster = start_cluster; 690 return; 691 } 692 693 desc_extent_rle = (struct spdk_blob_md_descriptor_extent_rle *)*buf; 694 desc_extent_rle->type = SPDK_MD_DESCRIPTOR_TYPE_EXTENT_RLE; 695 696 lba_per_cluster = _spdk_bs_cluster_to_lba(blob->bs, 1); 697 698 lba = blob->active.clusters[start_cluster]; 699 lba_count = lba_per_cluster; 700 extent_idx = 0; 701 for (i = start_cluster + 1; i < blob->active.num_clusters; i++) { 702 if ((lba + lba_count) == blob->active.clusters[i] && lba != 0) { 703 /* Run-length encode sequential non-zero LBA */ 704 lba_count += lba_per_cluster; 705 continue; 706 } else if (lba == 0 && blob->active.clusters[i] == 0) { 707 /* Run-length encode unallocated clusters */ 708 lba_count += lba_per_cluster; 709 continue; 710 } 711 desc_extent_rle->extents[extent_idx].cluster_idx = lba / lba_per_cluster; 712 desc_extent_rle->extents[extent_idx].length = lba_count / lba_per_cluster; 713 extent_idx++; 714 715 cur_sz += sizeof(desc_extent_rle->extents[extent_idx]); 716 717 if (*buf_sz < cur_sz) { 718 /* If we ran out of buffer space, return */ 719 *next_cluster = i; 720 goto finish; 721 } 722 723 lba = blob->active.clusters[i]; 724 lba_count = lba_per_cluster; 725 } 726 727 desc_extent_rle->extents[extent_idx].cluster_idx = lba / lba_per_cluster; 728 desc_extent_rle->extents[extent_idx].length = lba_count / lba_per_cluster; 729 extent_idx++; 730 731 *next_cluster = blob->active.num_clusters; 732 733 finish: 734 desc_extent_rle->length = sizeof(desc_extent_rle->extents[0]) * extent_idx; 735 *buf_sz -= sizeof(struct spdk_blob_md_descriptor) + desc_extent_rle->length; 736 *buf += sizeof(struct spdk_blob_md_descriptor) + desc_extent_rle->length; 737 738 return; 739 } 740 741 static int 742 _spdk_blob_serialize_extents_rle(const struct spdk_blob *blob, 743 struct spdk_blob_md_page **pages, 744 struct spdk_blob_md_page *cur_page, 745 uint32_t *page_count, uint8_t **buf, 746 size_t *remaining_sz) 747 { 748 uint64_t last_cluster; 749 int rc; 750 751 last_cluster = 0; 752 while (last_cluster < blob->active.num_clusters) { 753 _spdk_blob_serialize_extent_rle(blob, last_cluster, &last_cluster, buf, remaining_sz); 754 755 if (last_cluster == blob->active.num_clusters) { 756 break; 757 } 758 759 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, &cur_page); 760 if (rc < 0) { 761 return rc; 762 } 763 764 *buf = (uint8_t *)cur_page->descriptors; 765 *remaining_sz = sizeof(cur_page->descriptors); 766 } 767 768 return 0; 769 } 770 771 static void 772 _spdk_blob_serialize_flags(const struct spdk_blob *blob, 773 uint8_t *buf, size_t *buf_sz) 774 { 775 struct spdk_blob_md_descriptor_flags *desc; 776 777 /* 778 * Flags get serialized first, so we should always have room for the flags 779 * descriptor. 780 */ 781 assert(*buf_sz >= sizeof(*desc)); 782 783 desc = (struct spdk_blob_md_descriptor_flags *)buf; 784 desc->type = SPDK_MD_DESCRIPTOR_TYPE_FLAGS; 785 desc->length = sizeof(*desc) - sizeof(struct spdk_blob_md_descriptor); 786 desc->invalid_flags = blob->invalid_flags; 787 desc->data_ro_flags = blob->data_ro_flags; 788 desc->md_ro_flags = blob->md_ro_flags; 789 790 *buf_sz -= sizeof(*desc); 791 } 792 793 static int 794 _spdk_blob_serialize_xattrs(const struct spdk_blob *blob, 795 const struct spdk_xattr_tailq *xattrs, bool internal, 796 struct spdk_blob_md_page **pages, 797 struct spdk_blob_md_page *cur_page, 798 uint32_t *page_count, uint8_t **buf, 799 size_t *remaining_sz) 800 { 801 const struct spdk_xattr *xattr; 802 int rc; 803 804 TAILQ_FOREACH(xattr, xattrs, link) { 805 size_t required_sz = 0; 806 807 rc = _spdk_blob_serialize_xattr(xattr, 808 *buf, *remaining_sz, 809 &required_sz, internal); 810 if (rc < 0) { 811 /* Need to add a new page to the chain */ 812 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, 813 &cur_page); 814 if (rc < 0) { 815 spdk_free(*pages); 816 *pages = NULL; 817 *page_count = 0; 818 return rc; 819 } 820 821 *buf = (uint8_t *)cur_page->descriptors; 822 *remaining_sz = sizeof(cur_page->descriptors); 823 824 /* Try again */ 825 required_sz = 0; 826 rc = _spdk_blob_serialize_xattr(xattr, 827 *buf, *remaining_sz, 828 &required_sz, internal); 829 830 if (rc < 0) { 831 spdk_free(*pages); 832 *pages = NULL; 833 *page_count = 0; 834 return rc; 835 } 836 } 837 838 *remaining_sz -= required_sz; 839 *buf += required_sz; 840 } 841 842 return 0; 843 } 844 845 static int 846 _spdk_blob_serialize(const struct spdk_blob *blob, struct spdk_blob_md_page **pages, 847 uint32_t *page_count) 848 { 849 struct spdk_blob_md_page *cur_page; 850 int rc; 851 uint8_t *buf; 852 size_t remaining_sz; 853 854 assert(pages != NULL); 855 assert(page_count != NULL); 856 assert(blob != NULL); 857 assert(blob->state == SPDK_BLOB_STATE_DIRTY); 858 859 *pages = NULL; 860 *page_count = 0; 861 862 /* A blob always has at least 1 page, even if it has no descriptors */ 863 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, &cur_page); 864 if (rc < 0) { 865 return rc; 866 } 867 868 buf = (uint8_t *)cur_page->descriptors; 869 remaining_sz = sizeof(cur_page->descriptors); 870 871 /* Serialize flags */ 872 _spdk_blob_serialize_flags(blob, buf, &remaining_sz); 873 buf += sizeof(struct spdk_blob_md_descriptor_flags); 874 875 /* Serialize xattrs */ 876 rc = _spdk_blob_serialize_xattrs(blob, &blob->xattrs, false, 877 pages, cur_page, page_count, &buf, &remaining_sz); 878 if (rc < 0) { 879 return rc; 880 } 881 882 /* Serialize internal xattrs */ 883 rc = _spdk_blob_serialize_xattrs(blob, &blob->xattrs_internal, true, 884 pages, cur_page, page_count, &buf, &remaining_sz); 885 if (rc < 0) { 886 return rc; 887 } 888 889 /* Serialize extents */ 890 rc = _spdk_blob_serialize_extents_rle(blob, pages, cur_page, page_count, &buf, &remaining_sz); 891 892 return rc; 893 } 894 895 struct spdk_blob_load_ctx { 896 struct spdk_blob *blob; 897 898 struct spdk_blob_md_page *pages; 899 uint32_t num_pages; 900 spdk_bs_sequence_t *seq; 901 902 spdk_bs_sequence_cpl cb_fn; 903 void *cb_arg; 904 }; 905 906 static uint32_t 907 _spdk_blob_md_page_calc_crc(void *page) 908 { 909 uint32_t crc; 910 911 crc = BLOB_CRC32C_INITIAL; 912 crc = spdk_crc32c_update(page, SPDK_BS_PAGE_SIZE - 4, crc); 913 crc ^= BLOB_CRC32C_INITIAL; 914 915 return crc; 916 917 } 918 919 static void 920 _spdk_blob_load_final(void *cb_arg, int bserrno) 921 { 922 struct spdk_blob_load_ctx *ctx = cb_arg; 923 struct spdk_blob *blob = ctx->blob; 924 925 if (bserrno == 0) { 926 _spdk_blob_mark_clean(blob); 927 } 928 929 ctx->cb_fn(ctx->seq, ctx->cb_arg, bserrno); 930 931 /* Free the memory */ 932 spdk_free(ctx->pages); 933 free(ctx); 934 } 935 936 static void 937 _spdk_blob_load_snapshot_cpl(void *cb_arg, struct spdk_blob *snapshot, int bserrno) 938 { 939 struct spdk_blob_load_ctx *ctx = cb_arg; 940 struct spdk_blob *blob = ctx->blob; 941 942 if (bserrno == 0) { 943 blob->back_bs_dev = spdk_bs_create_blob_bs_dev(snapshot); 944 if (blob->back_bs_dev == NULL) { 945 bserrno = -ENOMEM; 946 } 947 } 948 if (bserrno != 0) { 949 SPDK_ERRLOG("Snapshot fail\n"); 950 } 951 952 _spdk_blob_load_final(ctx, bserrno); 953 } 954 955 static void 956 _spdk_blob_load_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 957 { 958 struct spdk_blob_load_ctx *ctx = cb_arg; 959 struct spdk_blob *blob = ctx->blob; 960 struct spdk_blob_md_page *page; 961 const void *value; 962 size_t len; 963 int rc; 964 uint32_t crc; 965 966 if (bserrno) { 967 SPDK_ERRLOG("Metadata page read failed: %d\n", bserrno); 968 _spdk_blob_load_final(ctx, bserrno); 969 return; 970 } 971 972 page = &ctx->pages[ctx->num_pages - 1]; 973 crc = _spdk_blob_md_page_calc_crc(page); 974 if (crc != page->crc) { 975 SPDK_ERRLOG("Metadata page %d crc mismatch\n", ctx->num_pages); 976 _spdk_blob_load_final(ctx, -EINVAL); 977 return; 978 } 979 980 if (page->next != SPDK_INVALID_MD_PAGE) { 981 uint32_t next_page = page->next; 982 uint64_t next_lba = _spdk_bs_md_page_to_lba(blob->bs, next_page); 983 984 /* Read the next page */ 985 ctx->num_pages++; 986 ctx->pages = spdk_realloc(ctx->pages, (sizeof(*page) * ctx->num_pages), 987 sizeof(*page)); 988 if (ctx->pages == NULL) { 989 _spdk_blob_load_final(ctx, -ENOMEM); 990 return; 991 } 992 993 spdk_bs_sequence_read_dev(seq, &ctx->pages[ctx->num_pages - 1], 994 next_lba, 995 _spdk_bs_byte_to_lba(blob->bs, sizeof(*page)), 996 _spdk_blob_load_cpl, ctx); 997 return; 998 } 999 1000 /* Parse the pages */ 1001 rc = _spdk_blob_parse(ctx->pages, ctx->num_pages, blob); 1002 if (rc) { 1003 _spdk_blob_load_final(ctx, rc); 1004 return; 1005 } 1006 1007 if (spdk_blob_is_thin_provisioned(blob)) { 1008 rc = _spdk_blob_get_xattr_value(blob, BLOB_SNAPSHOT, &value, &len, true); 1009 if (rc == 0) { 1010 if (len != sizeof(spdk_blob_id)) { 1011 _spdk_blob_load_final(ctx, -EINVAL); 1012 return; 1013 } 1014 /* open snapshot blob and continue in the callback function */ 1015 blob->parent_id = *(spdk_blob_id *)value; 1016 spdk_bs_open_blob(blob->bs, blob->parent_id, 1017 _spdk_blob_load_snapshot_cpl, ctx); 1018 return; 1019 } else { 1020 /* add zeroes_dev for thin provisioned blob */ 1021 blob->back_bs_dev = spdk_bs_create_zeroes_dev(); 1022 } 1023 } else { 1024 /* standard blob */ 1025 blob->back_bs_dev = NULL; 1026 } 1027 _spdk_blob_load_final(ctx, bserrno); 1028 } 1029 1030 /* Load a blob from disk given a blobid */ 1031 static void 1032 _spdk_blob_load(spdk_bs_sequence_t *seq, struct spdk_blob *blob, 1033 spdk_bs_sequence_cpl cb_fn, void *cb_arg) 1034 { 1035 struct spdk_blob_load_ctx *ctx; 1036 struct spdk_blob_store *bs; 1037 uint32_t page_num; 1038 uint64_t lba; 1039 1040 _spdk_blob_verify_md_op(blob); 1041 1042 bs = blob->bs; 1043 1044 ctx = calloc(1, sizeof(*ctx)); 1045 if (!ctx) { 1046 cb_fn(seq, cb_arg, -ENOMEM); 1047 return; 1048 } 1049 1050 ctx->blob = blob; 1051 ctx->pages = spdk_realloc(ctx->pages, SPDK_BS_PAGE_SIZE, SPDK_BS_PAGE_SIZE); 1052 if (!ctx->pages) { 1053 free(ctx); 1054 cb_fn(seq, cb_arg, -ENOMEM); 1055 return; 1056 } 1057 ctx->num_pages = 1; 1058 ctx->cb_fn = cb_fn; 1059 ctx->cb_arg = cb_arg; 1060 ctx->seq = seq; 1061 1062 page_num = _spdk_bs_blobid_to_page(blob->id); 1063 lba = _spdk_bs_md_page_to_lba(blob->bs, page_num); 1064 1065 blob->state = SPDK_BLOB_STATE_LOADING; 1066 1067 spdk_bs_sequence_read_dev(seq, &ctx->pages[0], lba, 1068 _spdk_bs_byte_to_lba(bs, SPDK_BS_PAGE_SIZE), 1069 _spdk_blob_load_cpl, ctx); 1070 } 1071 1072 struct spdk_blob_persist_ctx { 1073 struct spdk_blob *blob; 1074 1075 struct spdk_bs_super_block *super; 1076 1077 struct spdk_blob_md_page *pages; 1078 1079 spdk_bs_sequence_t *seq; 1080 spdk_bs_sequence_cpl cb_fn; 1081 void *cb_arg; 1082 }; 1083 1084 static void 1085 spdk_bs_batch_clear_dev(struct spdk_blob_persist_ctx *ctx, spdk_bs_batch_t *batch, uint64_t lba, 1086 uint32_t lba_count) 1087 { 1088 switch (ctx->blob->clear_method) { 1089 case BLOB_CLEAR_WITH_DEFAULT: 1090 case BLOB_CLEAR_WITH_UNMAP: 1091 spdk_bs_batch_unmap_dev(batch, lba, lba_count); 1092 break; 1093 case BLOB_CLEAR_WITH_WRITE_ZEROES: 1094 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1095 break; 1096 case BLOB_CLEAR_WITH_NONE: 1097 default: 1098 break; 1099 } 1100 } 1101 1102 static void 1103 _spdk_blob_persist_complete(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1104 { 1105 struct spdk_blob_persist_ctx *ctx = cb_arg; 1106 struct spdk_blob *blob = ctx->blob; 1107 1108 if (bserrno == 0) { 1109 _spdk_blob_mark_clean(blob); 1110 } 1111 1112 /* Call user callback */ 1113 ctx->cb_fn(seq, ctx->cb_arg, bserrno); 1114 1115 /* Free the memory */ 1116 spdk_free(ctx->pages); 1117 free(ctx); 1118 } 1119 1120 static void 1121 _spdk_blob_persist_clear_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1122 { 1123 struct spdk_blob_persist_ctx *ctx = cb_arg; 1124 struct spdk_blob *blob = ctx->blob; 1125 struct spdk_blob_store *bs = blob->bs; 1126 size_t i; 1127 1128 /* Release all clusters that were truncated */ 1129 for (i = blob->active.num_clusters; i < blob->active.cluster_array_size; i++) { 1130 uint32_t cluster_num = _spdk_bs_lba_to_cluster(bs, blob->active.clusters[i]); 1131 1132 /* Nothing to release if it was not allocated */ 1133 if (blob->active.clusters[i] != 0) { 1134 _spdk_bs_release_cluster(bs, cluster_num); 1135 } 1136 } 1137 1138 if (blob->active.num_clusters == 0) { 1139 free(blob->active.clusters); 1140 blob->active.clusters = NULL; 1141 blob->active.cluster_array_size = 0; 1142 } else if (blob->active.num_clusters != blob->active.cluster_array_size) { 1143 #ifndef __clang_analyzer__ 1144 void *tmp; 1145 1146 /* scan-build really can't figure reallocs, workaround it */ 1147 tmp = realloc(blob->active.clusters, sizeof(*blob->active.clusters) * blob->active.num_clusters); 1148 assert(tmp != NULL); 1149 blob->active.clusters = tmp; 1150 #endif 1151 blob->active.cluster_array_size = blob->active.num_clusters; 1152 } 1153 1154 _spdk_blob_persist_complete(seq, ctx, bserrno); 1155 } 1156 1157 static void 1158 _spdk_blob_persist_clear_clusters(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1159 { 1160 struct spdk_blob_persist_ctx *ctx = cb_arg; 1161 struct spdk_blob *blob = ctx->blob; 1162 struct spdk_blob_store *bs = blob->bs; 1163 spdk_bs_batch_t *batch; 1164 size_t i; 1165 uint64_t lba; 1166 uint32_t lba_count; 1167 1168 /* Clusters don't move around in blobs. The list shrinks or grows 1169 * at the end, but no changes ever occur in the middle of the list. 1170 */ 1171 1172 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_clear_clusters_cpl, ctx); 1173 1174 /* Clear all clusters that were truncated */ 1175 lba = 0; 1176 lba_count = 0; 1177 for (i = blob->active.num_clusters; i < blob->active.cluster_array_size; i++) { 1178 uint64_t next_lba = blob->active.clusters[i]; 1179 uint32_t next_lba_count = _spdk_bs_cluster_to_lba(bs, 1); 1180 1181 if (next_lba > 0 && (lba + lba_count) == next_lba) { 1182 /* This cluster is contiguous with the previous one. */ 1183 lba_count += next_lba_count; 1184 continue; 1185 } 1186 1187 /* This cluster is not contiguous with the previous one. */ 1188 1189 /* If a run of LBAs previously existing, clear them now */ 1190 if (lba_count > 0) { 1191 spdk_bs_batch_clear_dev(ctx, batch, lba, lba_count); 1192 } 1193 1194 /* Start building the next batch */ 1195 lba = next_lba; 1196 if (next_lba > 0) { 1197 lba_count = next_lba_count; 1198 } else { 1199 lba_count = 0; 1200 } 1201 } 1202 1203 /* If we ended with a contiguous set of LBAs, clear them now */ 1204 if (lba_count > 0) { 1205 spdk_bs_batch_clear_dev(ctx, batch, lba, lba_count); 1206 } 1207 1208 spdk_bs_batch_close(batch); 1209 } 1210 1211 static void 1212 _spdk_blob_persist_zero_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1213 { 1214 struct spdk_blob_persist_ctx *ctx = cb_arg; 1215 struct spdk_blob *blob = ctx->blob; 1216 struct spdk_blob_store *bs = blob->bs; 1217 size_t i; 1218 1219 /* This loop starts at 1 because the first page is special and handled 1220 * below. The pages (except the first) are never written in place, 1221 * so any pages in the clean list must be zeroed. 1222 */ 1223 for (i = 1; i < blob->clean.num_pages; i++) { 1224 spdk_bit_array_clear(bs->used_md_pages, blob->clean.pages[i]); 1225 } 1226 1227 if (blob->active.num_pages == 0) { 1228 uint32_t page_num; 1229 1230 page_num = _spdk_bs_blobid_to_page(blob->id); 1231 spdk_bit_array_clear(bs->used_md_pages, page_num); 1232 } 1233 1234 /* Move on to clearing clusters */ 1235 _spdk_blob_persist_clear_clusters(seq, ctx, 0); 1236 } 1237 1238 static void 1239 _spdk_blob_persist_zero_pages(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1240 { 1241 struct spdk_blob_persist_ctx *ctx = cb_arg; 1242 struct spdk_blob *blob = ctx->blob; 1243 struct spdk_blob_store *bs = blob->bs; 1244 uint64_t lba; 1245 uint32_t lba_count; 1246 spdk_bs_batch_t *batch; 1247 size_t i; 1248 1249 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_zero_pages_cpl, ctx); 1250 1251 lba_count = _spdk_bs_byte_to_lba(bs, SPDK_BS_PAGE_SIZE); 1252 1253 /* This loop starts at 1 because the first page is special and handled 1254 * below. The pages (except the first) are never written in place, 1255 * so any pages in the clean list must be zeroed. 1256 */ 1257 for (i = 1; i < blob->clean.num_pages; i++) { 1258 lba = _spdk_bs_md_page_to_lba(bs, blob->clean.pages[i]); 1259 1260 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1261 } 1262 1263 /* The first page will only be zeroed if this is a delete. */ 1264 if (blob->active.num_pages == 0) { 1265 uint32_t page_num; 1266 1267 /* The first page in the metadata goes where the blobid indicates */ 1268 page_num = _spdk_bs_blobid_to_page(blob->id); 1269 lba = _spdk_bs_md_page_to_lba(bs, page_num); 1270 1271 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1272 } 1273 1274 spdk_bs_batch_close(batch); 1275 } 1276 1277 static void 1278 _spdk_blob_persist_write_page_root(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1279 { 1280 struct spdk_blob_persist_ctx *ctx = cb_arg; 1281 struct spdk_blob *blob = ctx->blob; 1282 struct spdk_blob_store *bs = blob->bs; 1283 uint64_t lba; 1284 uint32_t lba_count; 1285 struct spdk_blob_md_page *page; 1286 1287 if (blob->active.num_pages == 0) { 1288 /* Move on to the next step */ 1289 _spdk_blob_persist_zero_pages(seq, ctx, 0); 1290 return; 1291 } 1292 1293 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(*page)); 1294 1295 page = &ctx->pages[0]; 1296 /* The first page in the metadata goes where the blobid indicates */ 1297 lba = _spdk_bs_md_page_to_lba(bs, _spdk_bs_blobid_to_page(blob->id)); 1298 1299 spdk_bs_sequence_write_dev(seq, page, lba, lba_count, 1300 _spdk_blob_persist_zero_pages, ctx); 1301 } 1302 1303 static void 1304 _spdk_blob_persist_write_page_chain(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1305 { 1306 struct spdk_blob_persist_ctx *ctx = cb_arg; 1307 struct spdk_blob *blob = ctx->blob; 1308 struct spdk_blob_store *bs = blob->bs; 1309 uint64_t lba; 1310 uint32_t lba_count; 1311 struct spdk_blob_md_page *page; 1312 spdk_bs_batch_t *batch; 1313 size_t i; 1314 1315 /* Clusters don't move around in blobs. The list shrinks or grows 1316 * at the end, but no changes ever occur in the middle of the list. 1317 */ 1318 1319 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(*page)); 1320 1321 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_write_page_root, ctx); 1322 1323 /* This starts at 1. The root page is not written until 1324 * all of the others are finished 1325 */ 1326 for (i = 1; i < blob->active.num_pages; i++) { 1327 page = &ctx->pages[i]; 1328 assert(page->sequence_num == i); 1329 1330 lba = _spdk_bs_md_page_to_lba(bs, blob->active.pages[i]); 1331 1332 spdk_bs_batch_write_dev(batch, page, lba, lba_count); 1333 } 1334 1335 spdk_bs_batch_close(batch); 1336 } 1337 1338 static int 1339 _spdk_blob_resize(struct spdk_blob *blob, uint64_t sz) 1340 { 1341 uint64_t i; 1342 uint64_t *tmp; 1343 uint64_t lfc; /* lowest free cluster */ 1344 uint64_t num_clusters; 1345 struct spdk_blob_store *bs; 1346 1347 bs = blob->bs; 1348 1349 _spdk_blob_verify_md_op(blob); 1350 1351 if (blob->active.num_clusters == sz) { 1352 return 0; 1353 } 1354 1355 if (blob->active.num_clusters < blob->active.cluster_array_size) { 1356 /* If this blob was resized to be larger, then smaller, then 1357 * larger without syncing, then the cluster array already 1358 * contains spare assigned clusters we can use. 1359 */ 1360 num_clusters = spdk_min(blob->active.cluster_array_size, 1361 sz); 1362 } else { 1363 num_clusters = blob->active.num_clusters; 1364 } 1365 1366 /* Do two passes - one to verify that we can obtain enough clusters 1367 * and another to actually claim them. 1368 */ 1369 1370 if (spdk_blob_is_thin_provisioned(blob) == false) { 1371 lfc = 0; 1372 for (i = num_clusters; i < sz; i++) { 1373 lfc = spdk_bit_array_find_first_clear(bs->used_clusters, lfc); 1374 if (lfc == UINT32_MAX) { 1375 /* No more free clusters. Cannot satisfy the request */ 1376 return -ENOSPC; 1377 } 1378 lfc++; 1379 } 1380 } 1381 1382 if (sz > num_clusters) { 1383 /* Expand the cluster array if necessary. 1384 * We only shrink the array when persisting. 1385 */ 1386 tmp = realloc(blob->active.clusters, sizeof(*blob->active.clusters) * sz); 1387 if (sz > 0 && tmp == NULL) { 1388 return -ENOMEM; 1389 } 1390 memset(tmp + blob->active.cluster_array_size, 0, 1391 sizeof(*blob->active.clusters) * (sz - blob->active.cluster_array_size)); 1392 blob->active.clusters = tmp; 1393 blob->active.cluster_array_size = sz; 1394 } 1395 1396 blob->state = SPDK_BLOB_STATE_DIRTY; 1397 1398 if (spdk_blob_is_thin_provisioned(blob) == false) { 1399 lfc = 0; 1400 for (i = num_clusters; i < sz; i++) { 1401 _spdk_bs_allocate_cluster(blob, i, &lfc, true); 1402 lfc++; 1403 } 1404 } 1405 1406 blob->active.num_clusters = sz; 1407 1408 return 0; 1409 } 1410 1411 static void 1412 _spdk_blob_persist_start(struct spdk_blob_persist_ctx *ctx) 1413 { 1414 spdk_bs_sequence_t *seq = ctx->seq; 1415 struct spdk_blob *blob = ctx->blob; 1416 struct spdk_blob_store *bs = blob->bs; 1417 uint64_t i; 1418 uint32_t page_num; 1419 void *tmp; 1420 int rc; 1421 1422 if (blob->active.num_pages == 0) { 1423 /* This is the signal that the blob should be deleted. 1424 * Immediately jump to the clean up routine. */ 1425 assert(blob->clean.num_pages > 0); 1426 blob->state = SPDK_BLOB_STATE_CLEAN; 1427 _spdk_blob_persist_zero_pages(seq, ctx, 0); 1428 return; 1429 1430 } 1431 1432 /* Generate the new metadata */ 1433 rc = _spdk_blob_serialize(blob, &ctx->pages, &blob->active.num_pages); 1434 if (rc < 0) { 1435 _spdk_blob_persist_complete(seq, ctx, rc); 1436 return; 1437 } 1438 1439 assert(blob->active.num_pages >= 1); 1440 1441 /* Resize the cache of page indices */ 1442 tmp = realloc(blob->active.pages, blob->active.num_pages * sizeof(*blob->active.pages)); 1443 if (!tmp) { 1444 _spdk_blob_persist_complete(seq, ctx, -ENOMEM); 1445 return; 1446 } 1447 blob->active.pages = tmp; 1448 1449 /* Assign this metadata to pages. This requires two passes - 1450 * one to verify that there are enough pages and a second 1451 * to actually claim them. */ 1452 page_num = 0; 1453 /* Note that this loop starts at one. The first page location is fixed by the blobid. */ 1454 for (i = 1; i < blob->active.num_pages; i++) { 1455 page_num = spdk_bit_array_find_first_clear(bs->used_md_pages, page_num); 1456 if (page_num == UINT32_MAX) { 1457 _spdk_blob_persist_complete(seq, ctx, -ENOMEM); 1458 return; 1459 } 1460 page_num++; 1461 } 1462 1463 page_num = 0; 1464 blob->active.pages[0] = _spdk_bs_blobid_to_page(blob->id); 1465 for (i = 1; i < blob->active.num_pages; i++) { 1466 page_num = spdk_bit_array_find_first_clear(bs->used_md_pages, page_num); 1467 ctx->pages[i - 1].next = page_num; 1468 /* Now that previous metadata page is complete, calculate the crc for it. */ 1469 ctx->pages[i - 1].crc = _spdk_blob_md_page_calc_crc(&ctx->pages[i - 1]); 1470 blob->active.pages[i] = page_num; 1471 spdk_bit_array_set(bs->used_md_pages, page_num); 1472 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Claiming page %u for blob %lu\n", page_num, blob->id); 1473 page_num++; 1474 } 1475 ctx->pages[i - 1].crc = _spdk_blob_md_page_calc_crc(&ctx->pages[i - 1]); 1476 /* Start writing the metadata from last page to first */ 1477 blob->state = SPDK_BLOB_STATE_CLEAN; 1478 _spdk_blob_persist_write_page_chain(seq, ctx, 0); 1479 } 1480 1481 static void 1482 _spdk_blob_persist_dirty_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1483 { 1484 struct spdk_blob_persist_ctx *ctx = cb_arg; 1485 1486 ctx->blob->bs->clean = 0; 1487 1488 spdk_free(ctx->super); 1489 1490 _spdk_blob_persist_start(ctx); 1491 } 1492 1493 static void 1494 _spdk_bs_write_super(spdk_bs_sequence_t *seq, struct spdk_blob_store *bs, 1495 struct spdk_bs_super_block *super, spdk_bs_sequence_cpl cb_fn, void *cb_arg); 1496 1497 1498 static void 1499 _spdk_blob_persist_dirty(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1500 { 1501 struct spdk_blob_persist_ctx *ctx = cb_arg; 1502 1503 ctx->super->clean = 0; 1504 if (ctx->super->size == 0) { 1505 ctx->super->size = ctx->blob->bs->dev->blockcnt * ctx->blob->bs->dev->blocklen; 1506 } 1507 1508 _spdk_bs_write_super(seq, ctx->blob->bs, ctx->super, _spdk_blob_persist_dirty_cpl, ctx); 1509 } 1510 1511 1512 /* Write a blob to disk */ 1513 static void 1514 _spdk_blob_persist(spdk_bs_sequence_t *seq, struct spdk_blob *blob, 1515 spdk_bs_sequence_cpl cb_fn, void *cb_arg) 1516 { 1517 struct spdk_blob_persist_ctx *ctx; 1518 1519 _spdk_blob_verify_md_op(blob); 1520 1521 if (blob->state == SPDK_BLOB_STATE_CLEAN) { 1522 cb_fn(seq, cb_arg, 0); 1523 return; 1524 } 1525 1526 ctx = calloc(1, sizeof(*ctx)); 1527 if (!ctx) { 1528 cb_fn(seq, cb_arg, -ENOMEM); 1529 return; 1530 } 1531 ctx->blob = blob; 1532 ctx->seq = seq; 1533 ctx->cb_fn = cb_fn; 1534 ctx->cb_arg = cb_arg; 1535 1536 if (blob->bs->clean) { 1537 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL, 1538 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 1539 if (!ctx->super) { 1540 cb_fn(seq, cb_arg, -ENOMEM); 1541 free(ctx); 1542 return; 1543 } 1544 1545 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(blob->bs, 0), 1546 _spdk_bs_byte_to_lba(blob->bs, sizeof(*ctx->super)), 1547 _spdk_blob_persist_dirty, ctx); 1548 } else { 1549 _spdk_blob_persist_start(ctx); 1550 } 1551 } 1552 1553 struct spdk_blob_copy_cluster_ctx { 1554 struct spdk_blob *blob; 1555 uint8_t *buf; 1556 uint64_t page; 1557 uint64_t new_cluster; 1558 spdk_bs_sequence_t *seq; 1559 }; 1560 1561 static void 1562 _spdk_blob_allocate_and_copy_cluster_cpl(void *cb_arg, int bserrno) 1563 { 1564 struct spdk_blob_copy_cluster_ctx *ctx = cb_arg; 1565 struct spdk_bs_request_set *set = (struct spdk_bs_request_set *)ctx->seq; 1566 TAILQ_HEAD(, spdk_bs_request_set) requests; 1567 spdk_bs_user_op_t *op; 1568 1569 TAILQ_INIT(&requests); 1570 TAILQ_SWAP(&set->channel->need_cluster_alloc, &requests, spdk_bs_request_set, link); 1571 1572 while (!TAILQ_EMPTY(&requests)) { 1573 op = TAILQ_FIRST(&requests); 1574 TAILQ_REMOVE(&requests, op, link); 1575 if (bserrno == 0) { 1576 spdk_bs_user_op_execute(op); 1577 } else { 1578 spdk_bs_user_op_abort(op); 1579 } 1580 } 1581 1582 spdk_free(ctx->buf); 1583 free(ctx); 1584 } 1585 1586 static void 1587 _spdk_blob_insert_cluster_cpl(void *cb_arg, int bserrno) 1588 { 1589 struct spdk_blob_copy_cluster_ctx *ctx = cb_arg; 1590 1591 if (bserrno) { 1592 if (bserrno == -EEXIST) { 1593 /* The metadata insert failed because another thread 1594 * allocated the cluster first. Free our cluster 1595 * but continue without error. */ 1596 bserrno = 0; 1597 } 1598 _spdk_bs_release_cluster(ctx->blob->bs, ctx->new_cluster); 1599 } 1600 1601 spdk_bs_sequence_finish(ctx->seq, bserrno); 1602 } 1603 1604 static void 1605 _spdk_blob_write_copy_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1606 { 1607 struct spdk_blob_copy_cluster_ctx *ctx = cb_arg; 1608 uint32_t cluster_number; 1609 1610 if (bserrno) { 1611 /* The write failed, so jump to the final completion handler */ 1612 spdk_bs_sequence_finish(seq, bserrno); 1613 return; 1614 } 1615 1616 cluster_number = _spdk_bs_page_to_cluster(ctx->blob->bs, ctx->page); 1617 1618 _spdk_blob_insert_cluster_on_md_thread(ctx->blob, cluster_number, ctx->new_cluster, 1619 _spdk_blob_insert_cluster_cpl, ctx); 1620 } 1621 1622 static void 1623 _spdk_blob_write_copy(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1624 { 1625 struct spdk_blob_copy_cluster_ctx *ctx = cb_arg; 1626 1627 if (bserrno != 0) { 1628 /* The read failed, so jump to the final completion handler */ 1629 spdk_bs_sequence_finish(seq, bserrno); 1630 return; 1631 } 1632 1633 /* Write whole cluster */ 1634 spdk_bs_sequence_write_dev(seq, ctx->buf, 1635 _spdk_bs_cluster_to_lba(ctx->blob->bs, ctx->new_cluster), 1636 _spdk_bs_cluster_to_lba(ctx->blob->bs, 1), 1637 _spdk_blob_write_copy_cpl, ctx); 1638 } 1639 1640 static void 1641 _spdk_bs_allocate_and_copy_cluster(struct spdk_blob *blob, 1642 struct spdk_io_channel *_ch, 1643 uint64_t io_unit, spdk_bs_user_op_t *op) 1644 { 1645 struct spdk_bs_cpl cpl; 1646 struct spdk_bs_channel *ch; 1647 struct spdk_blob_copy_cluster_ctx *ctx; 1648 uint32_t cluster_start_page; 1649 uint32_t cluster_number; 1650 int rc; 1651 1652 ch = spdk_io_channel_get_ctx(_ch); 1653 1654 if (!TAILQ_EMPTY(&ch->need_cluster_alloc)) { 1655 /* There are already operations pending. Queue this user op 1656 * and return because it will be re-executed when the outstanding 1657 * cluster allocation completes. */ 1658 TAILQ_INSERT_TAIL(&ch->need_cluster_alloc, op, link); 1659 return; 1660 } 1661 1662 /* Round the io_unit offset down to the first page in the cluster */ 1663 cluster_start_page = _spdk_bs_io_unit_to_cluster_start(blob, io_unit); 1664 1665 /* Calculate which index in the metadata cluster array the corresponding 1666 * cluster is supposed to be at. */ 1667 cluster_number = _spdk_bs_io_unit_to_cluster_number(blob, io_unit); 1668 1669 ctx = calloc(1, sizeof(*ctx)); 1670 if (!ctx) { 1671 spdk_bs_user_op_abort(op); 1672 return; 1673 } 1674 1675 assert(blob->bs->cluster_sz % blob->back_bs_dev->blocklen == 0); 1676 1677 ctx->blob = blob; 1678 ctx->page = cluster_start_page; 1679 1680 if (blob->parent_id != SPDK_BLOBID_INVALID) { 1681 ctx->buf = spdk_malloc(blob->bs->cluster_sz, blob->back_bs_dev->blocklen, 1682 NULL, SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 1683 if (!ctx->buf) { 1684 SPDK_ERRLOG("DMA allocation for cluster of size = %" PRIu32 " failed.\n", 1685 blob->bs->cluster_sz); 1686 free(ctx); 1687 spdk_bs_user_op_abort(op); 1688 return; 1689 } 1690 } 1691 1692 rc = _spdk_bs_allocate_cluster(blob, cluster_number, &ctx->new_cluster, false); 1693 if (rc != 0) { 1694 spdk_free(ctx->buf); 1695 free(ctx); 1696 spdk_bs_user_op_abort(op); 1697 return; 1698 } 1699 1700 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1701 cpl.u.blob_basic.cb_fn = _spdk_blob_allocate_and_copy_cluster_cpl; 1702 cpl.u.blob_basic.cb_arg = ctx; 1703 1704 ctx->seq = spdk_bs_sequence_start(_ch, &cpl); 1705 if (!ctx->seq) { 1706 _spdk_bs_release_cluster(blob->bs, ctx->new_cluster); 1707 spdk_free(ctx->buf); 1708 free(ctx); 1709 spdk_bs_user_op_abort(op); 1710 return; 1711 } 1712 1713 /* Queue the user op to block other incoming operations */ 1714 TAILQ_INSERT_TAIL(&ch->need_cluster_alloc, op, link); 1715 1716 if (blob->parent_id != SPDK_BLOBID_INVALID) { 1717 /* Read cluster from backing device */ 1718 spdk_bs_sequence_read_bs_dev(ctx->seq, blob->back_bs_dev, ctx->buf, 1719 _spdk_bs_dev_page_to_lba(blob->back_bs_dev, cluster_start_page), 1720 _spdk_bs_dev_byte_to_lba(blob->back_bs_dev, blob->bs->cluster_sz), 1721 _spdk_blob_write_copy, ctx); 1722 } else { 1723 _spdk_blob_insert_cluster_on_md_thread(ctx->blob, cluster_number, ctx->new_cluster, 1724 _spdk_blob_insert_cluster_cpl, ctx); 1725 } 1726 } 1727 1728 static void 1729 _spdk_blob_calculate_lba_and_lba_count(struct spdk_blob *blob, uint64_t io_unit, uint64_t length, 1730 uint64_t *lba, uint32_t *lba_count) 1731 { 1732 *lba_count = length; 1733 1734 if (!_spdk_bs_io_unit_is_allocated(blob, io_unit)) { 1735 assert(blob->back_bs_dev != NULL); 1736 *lba = _spdk_bs_io_unit_to_back_dev_lba(blob, io_unit); 1737 *lba_count = _spdk_bs_io_unit_to_back_dev_lba(blob, *lba_count); 1738 } else { 1739 *lba = _spdk_bs_blob_io_unit_to_lba(blob, io_unit); 1740 } 1741 } 1742 1743 struct op_split_ctx { 1744 struct spdk_blob *blob; 1745 struct spdk_io_channel *channel; 1746 uint64_t io_unit_offset; 1747 uint64_t io_units_remaining; 1748 void *curr_payload; 1749 enum spdk_blob_op_type op_type; 1750 spdk_bs_sequence_t *seq; 1751 }; 1752 1753 static void 1754 _spdk_blob_request_submit_op_split_next(void *cb_arg, int bserrno) 1755 { 1756 struct op_split_ctx *ctx = cb_arg; 1757 struct spdk_blob *blob = ctx->blob; 1758 struct spdk_io_channel *ch = ctx->channel; 1759 enum spdk_blob_op_type op_type = ctx->op_type; 1760 uint8_t *buf = ctx->curr_payload; 1761 uint64_t offset = ctx->io_unit_offset; 1762 uint64_t length = ctx->io_units_remaining; 1763 uint64_t op_length; 1764 1765 if (bserrno != 0 || ctx->io_units_remaining == 0) { 1766 spdk_bs_sequence_finish(ctx->seq, bserrno); 1767 free(ctx); 1768 return; 1769 } 1770 1771 op_length = spdk_min(length, _spdk_bs_num_io_units_to_cluster_boundary(blob, 1772 offset)); 1773 1774 /* Update length and payload for next operation */ 1775 ctx->io_units_remaining -= op_length; 1776 ctx->io_unit_offset += op_length; 1777 if (op_type == SPDK_BLOB_WRITE || op_type == SPDK_BLOB_READ) { 1778 ctx->curr_payload += op_length * blob->bs->io_unit_size; 1779 } 1780 1781 switch (op_type) { 1782 case SPDK_BLOB_READ: 1783 spdk_blob_io_read(blob, ch, buf, offset, op_length, 1784 _spdk_blob_request_submit_op_split_next, ctx); 1785 break; 1786 case SPDK_BLOB_WRITE: 1787 spdk_blob_io_write(blob, ch, buf, offset, op_length, 1788 _spdk_blob_request_submit_op_split_next, ctx); 1789 break; 1790 case SPDK_BLOB_UNMAP: 1791 spdk_blob_io_unmap(blob, ch, offset, op_length, 1792 _spdk_blob_request_submit_op_split_next, ctx); 1793 break; 1794 case SPDK_BLOB_WRITE_ZEROES: 1795 spdk_blob_io_write_zeroes(blob, ch, offset, op_length, 1796 _spdk_blob_request_submit_op_split_next, ctx); 1797 break; 1798 case SPDK_BLOB_READV: 1799 case SPDK_BLOB_WRITEV: 1800 SPDK_ERRLOG("readv/write not valid\n"); 1801 spdk_bs_sequence_finish(ctx->seq, -EINVAL); 1802 free(ctx); 1803 break; 1804 } 1805 } 1806 1807 static void 1808 _spdk_blob_request_submit_op_split(struct spdk_io_channel *ch, struct spdk_blob *blob, 1809 void *payload, uint64_t offset, uint64_t length, 1810 spdk_blob_op_complete cb_fn, void *cb_arg, enum spdk_blob_op_type op_type) 1811 { 1812 struct op_split_ctx *ctx; 1813 spdk_bs_sequence_t *seq; 1814 struct spdk_bs_cpl cpl; 1815 1816 assert(blob != NULL); 1817 1818 ctx = calloc(1, sizeof(struct op_split_ctx)); 1819 if (ctx == NULL) { 1820 cb_fn(cb_arg, -ENOMEM); 1821 return; 1822 } 1823 1824 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1825 cpl.u.blob_basic.cb_fn = cb_fn; 1826 cpl.u.blob_basic.cb_arg = cb_arg; 1827 1828 seq = spdk_bs_sequence_start(ch, &cpl); 1829 if (!seq) { 1830 free(ctx); 1831 cb_fn(cb_arg, -ENOMEM); 1832 return; 1833 } 1834 1835 ctx->blob = blob; 1836 ctx->channel = ch; 1837 ctx->curr_payload = payload; 1838 ctx->io_unit_offset = offset; 1839 ctx->io_units_remaining = length; 1840 ctx->op_type = op_type; 1841 ctx->seq = seq; 1842 1843 _spdk_blob_request_submit_op_split_next(ctx, 0); 1844 } 1845 1846 static void 1847 _spdk_blob_request_submit_op_single(struct spdk_io_channel *_ch, struct spdk_blob *blob, 1848 void *payload, uint64_t offset, uint64_t length, 1849 spdk_blob_op_complete cb_fn, void *cb_arg, enum spdk_blob_op_type op_type) 1850 { 1851 struct spdk_bs_cpl cpl; 1852 uint64_t lba; 1853 uint32_t lba_count; 1854 1855 assert(blob != NULL); 1856 1857 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1858 cpl.u.blob_basic.cb_fn = cb_fn; 1859 cpl.u.blob_basic.cb_arg = cb_arg; 1860 1861 _spdk_blob_calculate_lba_and_lba_count(blob, offset, length, &lba, &lba_count); 1862 1863 if (blob->frozen_refcnt) { 1864 /* This blob I/O is frozen */ 1865 spdk_bs_user_op_t *op; 1866 struct spdk_bs_channel *bs_channel = spdk_io_channel_get_ctx(_ch); 1867 1868 op = spdk_bs_user_op_alloc(_ch, &cpl, op_type, blob, payload, 0, offset, length); 1869 if (!op) { 1870 cb_fn(cb_arg, -ENOMEM); 1871 return; 1872 } 1873 1874 TAILQ_INSERT_TAIL(&bs_channel->queued_io, op, link); 1875 1876 return; 1877 } 1878 1879 switch (op_type) { 1880 case SPDK_BLOB_READ: { 1881 spdk_bs_batch_t *batch; 1882 1883 batch = spdk_bs_batch_open(_ch, &cpl); 1884 if (!batch) { 1885 cb_fn(cb_arg, -ENOMEM); 1886 return; 1887 } 1888 1889 if (_spdk_bs_io_unit_is_allocated(blob, offset)) { 1890 /* Read from the blob */ 1891 spdk_bs_batch_read_dev(batch, payload, lba, lba_count); 1892 } else { 1893 /* Read from the backing block device */ 1894 spdk_bs_batch_read_bs_dev(batch, blob->back_bs_dev, payload, lba, lba_count); 1895 } 1896 1897 spdk_bs_batch_close(batch); 1898 break; 1899 } 1900 case SPDK_BLOB_WRITE: 1901 case SPDK_BLOB_WRITE_ZEROES: { 1902 if (_spdk_bs_io_unit_is_allocated(blob, offset)) { 1903 /* Write to the blob */ 1904 spdk_bs_batch_t *batch; 1905 1906 if (lba_count == 0) { 1907 cb_fn(cb_arg, 0); 1908 return; 1909 } 1910 1911 batch = spdk_bs_batch_open(_ch, &cpl); 1912 if (!batch) { 1913 cb_fn(cb_arg, -ENOMEM); 1914 return; 1915 } 1916 1917 if (op_type == SPDK_BLOB_WRITE) { 1918 spdk_bs_batch_write_dev(batch, payload, lba, lba_count); 1919 } else { 1920 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1921 } 1922 1923 spdk_bs_batch_close(batch); 1924 } else { 1925 /* Queue this operation and allocate the cluster */ 1926 spdk_bs_user_op_t *op; 1927 1928 op = spdk_bs_user_op_alloc(_ch, &cpl, op_type, blob, payload, 0, offset, length); 1929 if (!op) { 1930 cb_fn(cb_arg, -ENOMEM); 1931 return; 1932 } 1933 1934 _spdk_bs_allocate_and_copy_cluster(blob, _ch, offset, op); 1935 } 1936 break; 1937 } 1938 case SPDK_BLOB_UNMAP: { 1939 spdk_bs_batch_t *batch; 1940 1941 batch = spdk_bs_batch_open(_ch, &cpl); 1942 if (!batch) { 1943 cb_fn(cb_arg, -ENOMEM); 1944 return; 1945 } 1946 1947 if (_spdk_bs_io_unit_is_allocated(blob, offset)) { 1948 spdk_bs_batch_unmap_dev(batch, lba, lba_count); 1949 } 1950 1951 spdk_bs_batch_close(batch); 1952 break; 1953 } 1954 case SPDK_BLOB_READV: 1955 case SPDK_BLOB_WRITEV: 1956 SPDK_ERRLOG("readv/write not valid\n"); 1957 cb_fn(cb_arg, -EINVAL); 1958 break; 1959 } 1960 } 1961 1962 static void 1963 _spdk_blob_request_submit_op(struct spdk_blob *blob, struct spdk_io_channel *_channel, 1964 void *payload, uint64_t offset, uint64_t length, 1965 spdk_blob_op_complete cb_fn, void *cb_arg, enum spdk_blob_op_type op_type) 1966 { 1967 assert(blob != NULL); 1968 1969 if (blob->data_ro && op_type != SPDK_BLOB_READ) { 1970 cb_fn(cb_arg, -EPERM); 1971 return; 1972 } 1973 1974 if (offset + length > _spdk_bs_cluster_to_lba(blob->bs, blob->active.num_clusters)) { 1975 cb_fn(cb_arg, -EINVAL); 1976 return; 1977 } 1978 if (length <= _spdk_bs_num_io_units_to_cluster_boundary(blob, offset)) { 1979 _spdk_blob_request_submit_op_single(_channel, blob, payload, offset, length, 1980 cb_fn, cb_arg, op_type); 1981 } else { 1982 _spdk_blob_request_submit_op_split(_channel, blob, payload, offset, length, 1983 cb_fn, cb_arg, op_type); 1984 } 1985 } 1986 1987 struct rw_iov_ctx { 1988 struct spdk_blob *blob; 1989 struct spdk_io_channel *channel; 1990 spdk_blob_op_complete cb_fn; 1991 void *cb_arg; 1992 bool read; 1993 int iovcnt; 1994 struct iovec *orig_iov; 1995 uint64_t io_unit_offset; 1996 uint64_t io_units_remaining; 1997 uint64_t io_units_done; 1998 struct iovec iov[0]; 1999 }; 2000 2001 static void 2002 _spdk_rw_iov_done(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2003 { 2004 assert(cb_arg == NULL); 2005 spdk_bs_sequence_finish(seq, bserrno); 2006 } 2007 2008 static void 2009 _spdk_rw_iov_split_next(void *cb_arg, int bserrno) 2010 { 2011 struct rw_iov_ctx *ctx = cb_arg; 2012 struct spdk_blob *blob = ctx->blob; 2013 struct iovec *iov, *orig_iov; 2014 int iovcnt; 2015 size_t orig_iovoff; 2016 uint64_t io_units_count, io_units_to_boundary, io_unit_offset; 2017 uint64_t byte_count; 2018 2019 if (bserrno != 0 || ctx->io_units_remaining == 0) { 2020 ctx->cb_fn(ctx->cb_arg, bserrno); 2021 free(ctx); 2022 return; 2023 } 2024 2025 io_unit_offset = ctx->io_unit_offset; 2026 io_units_to_boundary = _spdk_bs_num_io_units_to_cluster_boundary(blob, io_unit_offset); 2027 io_units_count = spdk_min(ctx->io_units_remaining, io_units_to_boundary); 2028 /* 2029 * Get index and offset into the original iov array for our current position in the I/O sequence. 2030 * byte_count will keep track of how many bytes remaining until orig_iov and orig_iovoff will 2031 * point to the current position in the I/O sequence. 2032 */ 2033 byte_count = ctx->io_units_done * blob->bs->io_unit_size; 2034 orig_iov = &ctx->orig_iov[0]; 2035 orig_iovoff = 0; 2036 while (byte_count > 0) { 2037 if (byte_count >= orig_iov->iov_len) { 2038 byte_count -= orig_iov->iov_len; 2039 orig_iov++; 2040 } else { 2041 orig_iovoff = byte_count; 2042 byte_count = 0; 2043 } 2044 } 2045 2046 /* 2047 * Build an iov array for the next I/O in the sequence. byte_count will keep track of how many 2048 * bytes of this next I/O remain to be accounted for in the new iov array. 2049 */ 2050 byte_count = io_units_count * blob->bs->io_unit_size; 2051 iov = &ctx->iov[0]; 2052 iovcnt = 0; 2053 while (byte_count > 0) { 2054 assert(iovcnt < ctx->iovcnt); 2055 iov->iov_len = spdk_min(byte_count, orig_iov->iov_len - orig_iovoff); 2056 iov->iov_base = orig_iov->iov_base + orig_iovoff; 2057 byte_count -= iov->iov_len; 2058 orig_iovoff = 0; 2059 orig_iov++; 2060 iov++; 2061 iovcnt++; 2062 } 2063 2064 ctx->io_unit_offset += io_units_count; 2065 ctx->io_units_remaining -= io_units_count; 2066 ctx->io_units_done += io_units_count; 2067 iov = &ctx->iov[0]; 2068 2069 if (ctx->read) { 2070 spdk_blob_io_readv(ctx->blob, ctx->channel, iov, iovcnt, io_unit_offset, 2071 io_units_count, _spdk_rw_iov_split_next, ctx); 2072 } else { 2073 spdk_blob_io_writev(ctx->blob, ctx->channel, iov, iovcnt, io_unit_offset, 2074 io_units_count, _spdk_rw_iov_split_next, ctx); 2075 } 2076 } 2077 2078 static void 2079 _spdk_blob_request_submit_rw_iov(struct spdk_blob *blob, struct spdk_io_channel *_channel, 2080 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 2081 spdk_blob_op_complete cb_fn, void *cb_arg, bool read) 2082 { 2083 struct spdk_bs_cpl cpl; 2084 2085 assert(blob != NULL); 2086 2087 if (!read && blob->data_ro) { 2088 cb_fn(cb_arg, -EPERM); 2089 return; 2090 } 2091 2092 if (length == 0) { 2093 cb_fn(cb_arg, 0); 2094 return; 2095 } 2096 2097 if (offset + length > _spdk_bs_cluster_to_lba(blob->bs, blob->active.num_clusters)) { 2098 cb_fn(cb_arg, -EINVAL); 2099 return; 2100 } 2101 2102 /* 2103 * For now, we implement readv/writev using a sequence (instead of a batch) to account for having 2104 * to split a request that spans a cluster boundary. For I/O that do not span a cluster boundary, 2105 * there will be no noticeable difference compared to using a batch. For I/O that do span a cluster 2106 * boundary, the target LBAs (after blob offset to LBA translation) may not be contiguous, so we need 2107 * to allocate a separate iov array and split the I/O such that none of the resulting 2108 * smaller I/O cross a cluster boundary. These smaller I/O will be issued in sequence (not in parallel) 2109 * but since this case happens very infrequently, any performance impact will be negligible. 2110 * 2111 * This could be optimized in the future to allocate a big enough iov array to account for all of the iovs 2112 * for all of the smaller I/Os, pre-build all of the iov arrays for the smaller I/Os, then issue them 2113 * in a batch. That would also require creating an intermediate spdk_bs_cpl that would get called 2114 * when the batch was completed, to allow for freeing the memory for the iov arrays. 2115 */ 2116 if (spdk_likely(length <= _spdk_bs_num_io_units_to_cluster_boundary(blob, offset))) { 2117 uint32_t lba_count; 2118 uint64_t lba; 2119 2120 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 2121 cpl.u.blob_basic.cb_fn = cb_fn; 2122 cpl.u.blob_basic.cb_arg = cb_arg; 2123 2124 if (blob->frozen_refcnt) { 2125 /* This blob I/O is frozen */ 2126 enum spdk_blob_op_type op_type; 2127 spdk_bs_user_op_t *op; 2128 struct spdk_bs_channel *bs_channel = spdk_io_channel_get_ctx(_channel); 2129 2130 op_type = read ? SPDK_BLOB_READV : SPDK_BLOB_WRITEV; 2131 op = spdk_bs_user_op_alloc(_channel, &cpl, op_type, blob, iov, iovcnt, offset, length); 2132 if (!op) { 2133 cb_fn(cb_arg, -ENOMEM); 2134 return; 2135 } 2136 2137 TAILQ_INSERT_TAIL(&bs_channel->queued_io, op, link); 2138 2139 return; 2140 } 2141 2142 _spdk_blob_calculate_lba_and_lba_count(blob, offset, length, &lba, &lba_count); 2143 2144 if (read) { 2145 spdk_bs_sequence_t *seq; 2146 2147 seq = spdk_bs_sequence_start(_channel, &cpl); 2148 if (!seq) { 2149 cb_fn(cb_arg, -ENOMEM); 2150 return; 2151 } 2152 2153 if (_spdk_bs_io_unit_is_allocated(blob, offset)) { 2154 spdk_bs_sequence_readv_dev(seq, iov, iovcnt, lba, lba_count, _spdk_rw_iov_done, NULL); 2155 } else { 2156 spdk_bs_sequence_readv_bs_dev(seq, blob->back_bs_dev, iov, iovcnt, lba, lba_count, 2157 _spdk_rw_iov_done, NULL); 2158 } 2159 } else { 2160 if (_spdk_bs_io_unit_is_allocated(blob, offset)) { 2161 spdk_bs_sequence_t *seq; 2162 2163 seq = spdk_bs_sequence_start(_channel, &cpl); 2164 if (!seq) { 2165 cb_fn(cb_arg, -ENOMEM); 2166 return; 2167 } 2168 2169 spdk_bs_sequence_writev_dev(seq, iov, iovcnt, lba, lba_count, _spdk_rw_iov_done, NULL); 2170 } else { 2171 /* Queue this operation and allocate the cluster */ 2172 spdk_bs_user_op_t *op; 2173 2174 op = spdk_bs_user_op_alloc(_channel, &cpl, SPDK_BLOB_WRITEV, blob, iov, iovcnt, offset, 2175 length); 2176 if (!op) { 2177 cb_fn(cb_arg, -ENOMEM); 2178 return; 2179 } 2180 2181 _spdk_bs_allocate_and_copy_cluster(blob, _channel, offset, op); 2182 } 2183 } 2184 } else { 2185 struct rw_iov_ctx *ctx; 2186 2187 ctx = calloc(1, sizeof(struct rw_iov_ctx) + iovcnt * sizeof(struct iovec)); 2188 if (ctx == NULL) { 2189 cb_fn(cb_arg, -ENOMEM); 2190 return; 2191 } 2192 2193 ctx->blob = blob; 2194 ctx->channel = _channel; 2195 ctx->cb_fn = cb_fn; 2196 ctx->cb_arg = cb_arg; 2197 ctx->read = read; 2198 ctx->orig_iov = iov; 2199 ctx->iovcnt = iovcnt; 2200 ctx->io_unit_offset = offset; 2201 ctx->io_units_remaining = length; 2202 ctx->io_units_done = 0; 2203 2204 _spdk_rw_iov_split_next(ctx, 0); 2205 } 2206 } 2207 2208 static struct spdk_blob * 2209 _spdk_blob_lookup(struct spdk_blob_store *bs, spdk_blob_id blobid) 2210 { 2211 struct spdk_blob *blob; 2212 2213 TAILQ_FOREACH(blob, &bs->blobs, link) { 2214 if (blob->id == blobid) { 2215 return blob; 2216 } 2217 } 2218 2219 return NULL; 2220 } 2221 2222 static void 2223 _spdk_blob_get_snapshot_and_clone_entries(struct spdk_blob *blob, 2224 struct spdk_blob_list **snapshot_entry, struct spdk_blob_list **clone_entry) 2225 { 2226 assert(blob != NULL); 2227 *snapshot_entry = NULL; 2228 *clone_entry = NULL; 2229 2230 if (blob->parent_id == SPDK_BLOBID_INVALID) { 2231 return; 2232 } 2233 2234 TAILQ_FOREACH(*snapshot_entry, &blob->bs->snapshots, link) { 2235 if ((*snapshot_entry)->id == blob->parent_id) { 2236 break; 2237 } 2238 } 2239 2240 if (*snapshot_entry != NULL) { 2241 TAILQ_FOREACH(*clone_entry, &(*snapshot_entry)->clones, link) { 2242 if ((*clone_entry)->id == blob->id) { 2243 break; 2244 } 2245 } 2246 2247 assert(clone_entry != NULL); 2248 } 2249 } 2250 2251 static int 2252 _spdk_bs_channel_create(void *io_device, void *ctx_buf) 2253 { 2254 struct spdk_blob_store *bs = io_device; 2255 struct spdk_bs_channel *channel = ctx_buf; 2256 struct spdk_bs_dev *dev; 2257 uint32_t max_ops = bs->max_channel_ops; 2258 uint32_t i; 2259 2260 dev = bs->dev; 2261 2262 channel->req_mem = calloc(max_ops, sizeof(struct spdk_bs_request_set)); 2263 if (!channel->req_mem) { 2264 return -1; 2265 } 2266 2267 TAILQ_INIT(&channel->reqs); 2268 2269 for (i = 0; i < max_ops; i++) { 2270 TAILQ_INSERT_TAIL(&channel->reqs, &channel->req_mem[i], link); 2271 } 2272 2273 channel->bs = bs; 2274 channel->dev = dev; 2275 channel->dev_channel = dev->create_channel(dev); 2276 2277 if (!channel->dev_channel) { 2278 SPDK_ERRLOG("Failed to create device channel.\n"); 2279 free(channel->req_mem); 2280 return -1; 2281 } 2282 2283 TAILQ_INIT(&channel->need_cluster_alloc); 2284 TAILQ_INIT(&channel->queued_io); 2285 2286 return 0; 2287 } 2288 2289 static void 2290 _spdk_bs_channel_destroy(void *io_device, void *ctx_buf) 2291 { 2292 struct spdk_bs_channel *channel = ctx_buf; 2293 spdk_bs_user_op_t *op; 2294 2295 while (!TAILQ_EMPTY(&channel->need_cluster_alloc)) { 2296 op = TAILQ_FIRST(&channel->need_cluster_alloc); 2297 TAILQ_REMOVE(&channel->need_cluster_alloc, op, link); 2298 spdk_bs_user_op_abort(op); 2299 } 2300 2301 while (!TAILQ_EMPTY(&channel->queued_io)) { 2302 op = TAILQ_FIRST(&channel->queued_io); 2303 TAILQ_REMOVE(&channel->queued_io, op, link); 2304 spdk_bs_user_op_abort(op); 2305 } 2306 2307 free(channel->req_mem); 2308 channel->dev->destroy_channel(channel->dev, channel->dev_channel); 2309 } 2310 2311 static void 2312 _spdk_bs_dev_destroy(void *io_device) 2313 { 2314 struct spdk_blob_store *bs = io_device; 2315 struct spdk_blob *blob, *blob_tmp; 2316 2317 bs->dev->destroy(bs->dev); 2318 2319 TAILQ_FOREACH_SAFE(blob, &bs->blobs, link, blob_tmp) { 2320 TAILQ_REMOVE(&bs->blobs, blob, link); 2321 _spdk_blob_free(blob); 2322 } 2323 2324 pthread_mutex_destroy(&bs->used_clusters_mutex); 2325 2326 spdk_bit_array_free(&bs->used_blobids); 2327 spdk_bit_array_free(&bs->used_md_pages); 2328 spdk_bit_array_free(&bs->used_clusters); 2329 /* 2330 * If this function is called for any reason except a successful unload, 2331 * the unload_cpl type will be NONE and this will be a nop. 2332 */ 2333 spdk_bs_call_cpl(&bs->unload_cpl, bs->unload_err); 2334 2335 free(bs); 2336 } 2337 2338 static int 2339 _spdk_bs_blob_list_add(struct spdk_blob *blob) 2340 { 2341 spdk_blob_id snapshot_id; 2342 struct spdk_blob_list *snapshot_entry = NULL; 2343 struct spdk_blob_list *clone_entry = NULL; 2344 2345 assert(blob != NULL); 2346 2347 snapshot_id = blob->parent_id; 2348 if (snapshot_id == SPDK_BLOBID_INVALID) { 2349 return 0; 2350 } 2351 2352 snapshot_entry = _spdk_bs_get_snapshot_entry(blob->bs, snapshot_id); 2353 if (snapshot_entry == NULL) { 2354 /* Snapshot not found */ 2355 snapshot_entry = calloc(1, sizeof(struct spdk_blob_list)); 2356 if (snapshot_entry == NULL) { 2357 return -ENOMEM; 2358 } 2359 snapshot_entry->id = snapshot_id; 2360 TAILQ_INIT(&snapshot_entry->clones); 2361 TAILQ_INSERT_TAIL(&blob->bs->snapshots, snapshot_entry, link); 2362 } else { 2363 TAILQ_FOREACH(clone_entry, &snapshot_entry->clones, link) { 2364 if (clone_entry->id == blob->id) { 2365 break; 2366 } 2367 } 2368 } 2369 2370 if (clone_entry == NULL) { 2371 /* Clone not found */ 2372 clone_entry = calloc(1, sizeof(struct spdk_blob_list)); 2373 if (clone_entry == NULL) { 2374 return -ENOMEM; 2375 } 2376 clone_entry->id = blob->id; 2377 TAILQ_INIT(&clone_entry->clones); 2378 TAILQ_INSERT_TAIL(&snapshot_entry->clones, clone_entry, link); 2379 snapshot_entry->clone_count++; 2380 } 2381 2382 return 0; 2383 } 2384 2385 static void 2386 _spdk_bs_blob_list_remove(struct spdk_blob *blob) 2387 { 2388 struct spdk_blob_list *snapshot_entry = NULL; 2389 struct spdk_blob_list *clone_entry = NULL; 2390 2391 _spdk_blob_get_snapshot_and_clone_entries(blob, &snapshot_entry, &clone_entry); 2392 2393 if (snapshot_entry == NULL) { 2394 return; 2395 } 2396 2397 blob->parent_id = SPDK_BLOBID_INVALID; 2398 TAILQ_REMOVE(&snapshot_entry->clones, clone_entry, link); 2399 free(clone_entry); 2400 2401 snapshot_entry->clone_count--; 2402 } 2403 2404 static int 2405 _spdk_bs_blob_list_free(struct spdk_blob_store *bs) 2406 { 2407 struct spdk_blob_list *snapshot_entry; 2408 struct spdk_blob_list *snapshot_entry_tmp; 2409 struct spdk_blob_list *clone_entry; 2410 struct spdk_blob_list *clone_entry_tmp; 2411 2412 TAILQ_FOREACH_SAFE(snapshot_entry, &bs->snapshots, link, snapshot_entry_tmp) { 2413 TAILQ_FOREACH_SAFE(clone_entry, &snapshot_entry->clones, link, clone_entry_tmp) { 2414 TAILQ_REMOVE(&snapshot_entry->clones, clone_entry, link); 2415 free(clone_entry); 2416 } 2417 TAILQ_REMOVE(&bs->snapshots, snapshot_entry, link); 2418 free(snapshot_entry); 2419 } 2420 2421 return 0; 2422 } 2423 2424 static void 2425 _spdk_bs_free(struct spdk_blob_store *bs) 2426 { 2427 _spdk_bs_blob_list_free(bs); 2428 2429 spdk_bs_unregister_md_thread(bs); 2430 spdk_io_device_unregister(bs, _spdk_bs_dev_destroy); 2431 } 2432 2433 void 2434 spdk_bs_opts_init(struct spdk_bs_opts *opts) 2435 { 2436 opts->cluster_sz = SPDK_BLOB_OPTS_CLUSTER_SZ; 2437 opts->num_md_pages = SPDK_BLOB_OPTS_NUM_MD_PAGES; 2438 opts->max_md_ops = SPDK_BLOB_OPTS_MAX_MD_OPS; 2439 opts->max_channel_ops = SPDK_BLOB_OPTS_DEFAULT_CHANNEL_OPS; 2440 opts->clear_method = BS_CLEAR_WITH_UNMAP; 2441 memset(&opts->bstype, 0, sizeof(opts->bstype)); 2442 opts->iter_cb_fn = NULL; 2443 opts->iter_cb_arg = NULL; 2444 } 2445 2446 static int 2447 _spdk_bs_opts_verify(struct spdk_bs_opts *opts) 2448 { 2449 if (opts->cluster_sz == 0 || opts->num_md_pages == 0 || opts->max_md_ops == 0 || 2450 opts->max_channel_ops == 0) { 2451 SPDK_ERRLOG("Blobstore options cannot be set to 0\n"); 2452 return -1; 2453 } 2454 2455 return 0; 2456 } 2457 2458 static int 2459 _spdk_bs_alloc(struct spdk_bs_dev *dev, struct spdk_bs_opts *opts, struct spdk_blob_store **_bs) 2460 { 2461 struct spdk_blob_store *bs; 2462 uint64_t dev_size; 2463 int rc; 2464 2465 dev_size = dev->blocklen * dev->blockcnt; 2466 if (dev_size < opts->cluster_sz) { 2467 /* Device size cannot be smaller than cluster size of blobstore */ 2468 SPDK_INFOLOG(SPDK_LOG_BLOB, "Device size %" PRIu64 " is smaller than cluster size %" PRIu32 "\n", 2469 dev_size, opts->cluster_sz); 2470 return -ENOSPC; 2471 } 2472 if (opts->cluster_sz < SPDK_BS_PAGE_SIZE) { 2473 /* Cluster size cannot be smaller than page size */ 2474 SPDK_ERRLOG("Cluster size %" PRIu32 " is smaller than page size %d\n", 2475 opts->cluster_sz, SPDK_BS_PAGE_SIZE); 2476 return -EINVAL; 2477 } 2478 bs = calloc(1, sizeof(struct spdk_blob_store)); 2479 if (!bs) { 2480 return -ENOMEM; 2481 } 2482 2483 TAILQ_INIT(&bs->blobs); 2484 TAILQ_INIT(&bs->snapshots); 2485 bs->dev = dev; 2486 bs->md_thread = spdk_get_thread(); 2487 assert(bs->md_thread != NULL); 2488 2489 /* 2490 * Do not use _spdk_bs_lba_to_cluster() here since blockcnt may not be an 2491 * even multiple of the cluster size. 2492 */ 2493 bs->cluster_sz = opts->cluster_sz; 2494 bs->total_clusters = dev->blockcnt / (bs->cluster_sz / dev->blocklen); 2495 bs->pages_per_cluster = bs->cluster_sz / SPDK_BS_PAGE_SIZE; 2496 bs->num_free_clusters = bs->total_clusters; 2497 bs->used_clusters = spdk_bit_array_create(bs->total_clusters); 2498 bs->io_unit_size = dev->blocklen; 2499 if (bs->used_clusters == NULL) { 2500 free(bs); 2501 return -ENOMEM; 2502 } 2503 2504 bs->max_channel_ops = opts->max_channel_ops; 2505 bs->super_blob = SPDK_BLOBID_INVALID; 2506 memcpy(&bs->bstype, &opts->bstype, sizeof(opts->bstype)); 2507 2508 /* The metadata is assumed to be at least 1 page */ 2509 bs->used_md_pages = spdk_bit_array_create(1); 2510 bs->used_blobids = spdk_bit_array_create(0); 2511 2512 pthread_mutex_init(&bs->used_clusters_mutex, NULL); 2513 2514 spdk_io_device_register(bs, _spdk_bs_channel_create, _spdk_bs_channel_destroy, 2515 sizeof(struct spdk_bs_channel), "blobstore"); 2516 rc = spdk_bs_register_md_thread(bs); 2517 if (rc == -1) { 2518 spdk_io_device_unregister(bs, NULL); 2519 pthread_mutex_destroy(&bs->used_clusters_mutex); 2520 spdk_bit_array_free(&bs->used_blobids); 2521 spdk_bit_array_free(&bs->used_md_pages); 2522 spdk_bit_array_free(&bs->used_clusters); 2523 free(bs); 2524 /* FIXME: this is a lie but don't know how to get a proper error code here */ 2525 return -ENOMEM; 2526 } 2527 2528 *_bs = bs; 2529 return 0; 2530 } 2531 2532 /* START spdk_bs_load, spdk_bs_load_ctx will used for both load and unload. */ 2533 2534 struct spdk_bs_load_ctx { 2535 struct spdk_blob_store *bs; 2536 struct spdk_bs_super_block *super; 2537 2538 struct spdk_bs_md_mask *mask; 2539 bool in_page_chain; 2540 uint32_t page_index; 2541 uint32_t cur_page; 2542 struct spdk_blob_md_page *page; 2543 2544 spdk_bs_sequence_t *seq; 2545 spdk_blob_op_with_handle_complete iter_cb_fn; 2546 void *iter_cb_arg; 2547 struct spdk_blob *blob; 2548 spdk_blob_id blobid; 2549 }; 2550 2551 static void 2552 _spdk_bs_load_ctx_fail(spdk_bs_sequence_t *seq, struct spdk_bs_load_ctx *ctx, int bserrno) 2553 { 2554 assert(bserrno != 0); 2555 2556 spdk_free(ctx->super); 2557 spdk_bs_sequence_finish(seq, bserrno); 2558 _spdk_bs_free(ctx->bs); 2559 free(ctx); 2560 } 2561 2562 static void 2563 _spdk_bs_set_mask(struct spdk_bit_array *array, struct spdk_bs_md_mask *mask) 2564 { 2565 uint32_t i = 0; 2566 2567 while (true) { 2568 i = spdk_bit_array_find_first_set(array, i); 2569 if (i >= mask->length) { 2570 break; 2571 } 2572 mask->mask[i / 8] |= 1U << (i % 8); 2573 i++; 2574 } 2575 } 2576 2577 static int 2578 _spdk_bs_load_mask(struct spdk_bit_array **array_ptr, struct spdk_bs_md_mask *mask) 2579 { 2580 struct spdk_bit_array *array; 2581 uint32_t i; 2582 2583 if (spdk_bit_array_resize(array_ptr, mask->length) < 0) { 2584 return -ENOMEM; 2585 } 2586 2587 array = *array_ptr; 2588 for (i = 0; i < mask->length; i++) { 2589 if (mask->mask[i / 8] & (1U << (i % 8))) { 2590 spdk_bit_array_set(array, i); 2591 } 2592 } 2593 2594 return 0; 2595 } 2596 2597 static void 2598 _spdk_bs_write_super(spdk_bs_sequence_t *seq, struct spdk_blob_store *bs, 2599 struct spdk_bs_super_block *super, spdk_bs_sequence_cpl cb_fn, void *cb_arg) 2600 { 2601 /* Update the values in the super block */ 2602 super->super_blob = bs->super_blob; 2603 memcpy(&super->bstype, &bs->bstype, sizeof(bs->bstype)); 2604 super->crc = _spdk_blob_md_page_calc_crc(super); 2605 spdk_bs_sequence_write_dev(seq, super, _spdk_bs_page_to_lba(bs, 0), 2606 _spdk_bs_byte_to_lba(bs, sizeof(*super)), 2607 cb_fn, cb_arg); 2608 } 2609 2610 static void 2611 _spdk_bs_write_used_clusters(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 2612 { 2613 struct spdk_bs_load_ctx *ctx = arg; 2614 uint64_t mask_size, lba, lba_count; 2615 2616 /* Write out the used clusters mask */ 2617 mask_size = ctx->super->used_cluster_mask_len * SPDK_BS_PAGE_SIZE; 2618 ctx->mask = spdk_zmalloc(mask_size, 0x1000, NULL, 2619 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 2620 if (!ctx->mask) { 2621 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2622 return; 2623 } 2624 2625 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_CLUSTERS; 2626 ctx->mask->length = ctx->bs->total_clusters; 2627 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_clusters)); 2628 2629 _spdk_bs_set_mask(ctx->bs->used_clusters, ctx->mask); 2630 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_start); 2631 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_len); 2632 spdk_bs_sequence_write_dev(seq, ctx->mask, lba, lba_count, cb_fn, arg); 2633 } 2634 2635 static void 2636 _spdk_bs_write_used_md(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 2637 { 2638 struct spdk_bs_load_ctx *ctx = arg; 2639 uint64_t mask_size, lba, lba_count; 2640 2641 if (seq->bserrno) { 2642 _spdk_bs_load_ctx_fail(seq, ctx, seq->bserrno); 2643 return; 2644 } 2645 2646 mask_size = ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE; 2647 ctx->mask = spdk_zmalloc(mask_size, 0x1000, NULL, 2648 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 2649 if (!ctx->mask) { 2650 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2651 return; 2652 } 2653 2654 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_PAGES; 2655 ctx->mask->length = ctx->super->md_len; 2656 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_md_pages)); 2657 2658 _spdk_bs_set_mask(ctx->bs->used_md_pages, ctx->mask); 2659 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_start); 2660 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_len); 2661 spdk_bs_sequence_write_dev(seq, ctx->mask, lba, lba_count, cb_fn, arg); 2662 } 2663 2664 static void 2665 _spdk_bs_write_used_blobids(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 2666 { 2667 struct spdk_bs_load_ctx *ctx = arg; 2668 uint64_t mask_size, lba, lba_count; 2669 2670 if (ctx->super->used_blobid_mask_len == 0) { 2671 /* 2672 * This is a pre-v3 on-disk format where the blobid mask does not get 2673 * written to disk. 2674 */ 2675 cb_fn(seq, arg, 0); 2676 return; 2677 } 2678 2679 mask_size = ctx->super->used_blobid_mask_len * SPDK_BS_PAGE_SIZE; 2680 ctx->mask = spdk_zmalloc(mask_size, 0x1000, NULL, SPDK_ENV_SOCKET_ID_ANY, 2681 SPDK_MALLOC_DMA); 2682 if (!ctx->mask) { 2683 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2684 return; 2685 } 2686 2687 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_BLOBIDS; 2688 ctx->mask->length = ctx->super->md_len; 2689 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_blobids)); 2690 2691 _spdk_bs_set_mask(ctx->bs->used_blobids, ctx->mask); 2692 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_start); 2693 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_len); 2694 spdk_bs_sequence_write_dev(seq, ctx->mask, lba, lba_count, cb_fn, arg); 2695 } 2696 2697 static void 2698 _spdk_blob_set_thin_provision(struct spdk_blob *blob) 2699 { 2700 _spdk_blob_verify_md_op(blob); 2701 blob->invalid_flags |= SPDK_BLOB_THIN_PROV; 2702 blob->state = SPDK_BLOB_STATE_DIRTY; 2703 } 2704 2705 static void _spdk_bs_load_iter(void *arg, struct spdk_blob *blob, int bserrno); 2706 2707 static void 2708 _spdk_bs_delete_corrupted_blob_cpl(void *cb_arg, int bserrno) 2709 { 2710 struct spdk_bs_load_ctx *ctx = cb_arg; 2711 spdk_blob_id id; 2712 int64_t page_num; 2713 2714 /* Iterate to next blob (we can't use spdk_bs_iter_next function as our 2715 * last blob has been removed */ 2716 page_num = _spdk_bs_blobid_to_page(ctx->blobid); 2717 page_num++; 2718 page_num = spdk_bit_array_find_first_set(ctx->bs->used_blobids, page_num); 2719 if (page_num >= spdk_bit_array_capacity(ctx->bs->used_blobids)) { 2720 _spdk_bs_load_iter(ctx, NULL, -ENOENT); 2721 return; 2722 } 2723 2724 id = _spdk_bs_page_to_blobid(page_num); 2725 2726 spdk_bs_open_blob(ctx->bs, id, _spdk_bs_load_iter, ctx); 2727 } 2728 2729 static void 2730 _spdk_bs_delete_corrupted_close_cb(void *cb_arg, int bserrno) 2731 { 2732 struct spdk_bs_load_ctx *ctx = cb_arg; 2733 2734 if (bserrno != 0) { 2735 SPDK_ERRLOG("Failed to close corrupted blob\n"); 2736 spdk_bs_iter_next(ctx->bs, ctx->blob, _spdk_bs_load_iter, ctx); 2737 return; 2738 } 2739 2740 spdk_bs_delete_blob(ctx->bs, ctx->blobid, _spdk_bs_delete_corrupted_blob_cpl, ctx); 2741 } 2742 2743 static void 2744 _spdk_bs_delete_corrupted_blob(void *cb_arg, int bserrno) 2745 { 2746 struct spdk_bs_load_ctx *ctx = cb_arg; 2747 uint64_t i; 2748 2749 if (bserrno != 0) { 2750 SPDK_ERRLOG("Failed to close clone of a corrupted blob\n"); 2751 spdk_bs_iter_next(ctx->bs, ctx->blob, _spdk_bs_load_iter, ctx); 2752 return; 2753 } 2754 2755 /* Snapshot and clone have the same copy of cluster map at this point. 2756 * Let's clear cluster map for snpashot now so that it won't be cleared 2757 * for clone later when we remove snapshot. Also set thin provision to 2758 * pass data corruption check */ 2759 for (i = 0; i < ctx->blob->active.num_clusters; i++) { 2760 ctx->blob->active.clusters[i] = 0; 2761 } 2762 2763 ctx->blob->md_ro = false; 2764 2765 _spdk_blob_set_thin_provision(ctx->blob); 2766 2767 ctx->blobid = ctx->blob->id; 2768 2769 spdk_blob_close(ctx->blob, _spdk_bs_delete_corrupted_close_cb, ctx); 2770 } 2771 2772 static void 2773 _spdk_bs_update_corrupted_blob(void *cb_arg, int bserrno) 2774 { 2775 struct spdk_bs_load_ctx *ctx = cb_arg; 2776 2777 if (bserrno != 0) { 2778 SPDK_ERRLOG("Failed to close clone of a corrupted blob\n"); 2779 spdk_bs_iter_next(ctx->bs, ctx->blob, _spdk_bs_load_iter, ctx); 2780 return; 2781 } 2782 2783 ctx->blob->md_ro = false; 2784 _spdk_blob_remove_xattr(ctx->blob, SNAPSHOT_PENDING_REMOVAL, true); 2785 _spdk_blob_remove_xattr(ctx->blob, SNAPSHOT_IN_PROGRESS, true); 2786 spdk_blob_set_read_only(ctx->blob); 2787 2788 if (ctx->iter_cb_fn) { 2789 ctx->iter_cb_fn(ctx->iter_cb_arg, ctx->blob, 0); 2790 } 2791 _spdk_bs_blob_list_add(ctx->blob); 2792 2793 spdk_bs_iter_next(ctx->bs, ctx->blob, _spdk_bs_load_iter, ctx); 2794 } 2795 2796 static void 2797 _spdk_bs_examine_clone(void *cb_arg, struct spdk_blob *blob, int bserrno) 2798 { 2799 struct spdk_bs_load_ctx *ctx = cb_arg; 2800 2801 if (bserrno != 0) { 2802 SPDK_ERRLOG("Failed to open clone of a corrupted blob\n"); 2803 spdk_bs_iter_next(ctx->bs, ctx->blob, _spdk_bs_load_iter, ctx); 2804 return; 2805 } 2806 2807 if (blob->parent_id == ctx->blob->id) { 2808 /* Power failure occured before updating clone (snapshot delete case) 2809 * or after updating clone (creating snapshot case) - keep snapshot */ 2810 spdk_blob_close(blob, _spdk_bs_update_corrupted_blob, ctx); 2811 } else { 2812 /* Power failure occured after updating clone (snapshot delete case) 2813 * or before updating clone (creating snapshot case) - remove snapshot */ 2814 spdk_blob_close(blob, _spdk_bs_delete_corrupted_blob, ctx); 2815 } 2816 } 2817 2818 static void 2819 _spdk_bs_load_iter(void *arg, struct spdk_blob *blob, int bserrno) 2820 { 2821 struct spdk_bs_load_ctx *ctx = arg; 2822 const void *value; 2823 size_t len; 2824 int rc = 0; 2825 2826 if (bserrno == 0) { 2827 /* Examine blob if it is corrupted after power failure. Fix 2828 * the ones that can be fixed and remove any other corrupted 2829 * ones. If it is not corrupted just process it */ 2830 rc = _spdk_blob_get_xattr_value(blob, SNAPSHOT_PENDING_REMOVAL, &value, &len, true); 2831 if (rc != 0) { 2832 rc = _spdk_blob_get_xattr_value(blob, SNAPSHOT_IN_PROGRESS, &value, &len, true); 2833 if (rc != 0) { 2834 /* Not corrupted - process it and continue with iterating through blobs */ 2835 if (ctx->iter_cb_fn) { 2836 ctx->iter_cb_fn(ctx->iter_cb_arg, blob, 0); 2837 } 2838 _spdk_bs_blob_list_add(blob); 2839 spdk_bs_iter_next(ctx->bs, blob, _spdk_bs_load_iter, ctx); 2840 return; 2841 } 2842 2843 } 2844 2845 assert(len == sizeof(spdk_blob_id)); 2846 2847 ctx->blob = blob; 2848 2849 /* Open clone to check if we are able to fix this blob or should we remove it */ 2850 spdk_bs_open_blob(ctx->bs, *(spdk_blob_id *)value, _spdk_bs_examine_clone, ctx); 2851 return; 2852 } else if (bserrno == -ENOENT) { 2853 bserrno = 0; 2854 } else { 2855 /* 2856 * This case needs to be looked at further. Same problem 2857 * exists with applications that rely on explicit blob 2858 * iteration. We should just skip the blob that failed 2859 * to load and continue on to the next one. 2860 */ 2861 SPDK_ERRLOG("Error in iterating blobs\n"); 2862 } 2863 2864 ctx->iter_cb_fn = NULL; 2865 2866 spdk_free(ctx->super); 2867 spdk_free(ctx->mask); 2868 spdk_bs_sequence_finish(ctx->seq, bserrno); 2869 free(ctx); 2870 } 2871 2872 static void 2873 _spdk_bs_load_complete(spdk_bs_sequence_t *seq, struct spdk_bs_load_ctx *ctx, int bserrno) 2874 { 2875 ctx->seq = seq; 2876 spdk_bs_iter_first(ctx->bs, _spdk_bs_load_iter, ctx); 2877 } 2878 2879 static void 2880 _spdk_bs_load_used_blobids_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2881 { 2882 struct spdk_bs_load_ctx *ctx = cb_arg; 2883 int rc; 2884 2885 /* The type must be correct */ 2886 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_BLOBIDS); 2887 2888 /* The length of the mask (in bits) must not be greater than 2889 * the length of the buffer (converted to bits) */ 2890 assert(ctx->mask->length <= (ctx->super->used_blobid_mask_len * SPDK_BS_PAGE_SIZE * 8)); 2891 2892 /* The length of the mask must be exactly equal to the size 2893 * (in pages) of the metadata region */ 2894 assert(ctx->mask->length == ctx->super->md_len); 2895 2896 rc = _spdk_bs_load_mask(&ctx->bs->used_blobids, ctx->mask); 2897 if (rc < 0) { 2898 spdk_free(ctx->mask); 2899 _spdk_bs_load_ctx_fail(seq, ctx, rc); 2900 return; 2901 } 2902 2903 _spdk_bs_load_complete(seq, ctx, bserrno); 2904 } 2905 2906 static void 2907 _spdk_bs_load_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2908 { 2909 struct spdk_bs_load_ctx *ctx = cb_arg; 2910 uint64_t lba, lba_count, mask_size; 2911 int rc; 2912 2913 /* The type must be correct */ 2914 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_CLUSTERS); 2915 /* The length of the mask (in bits) must not be greater than the length of the buffer (converted to bits) */ 2916 assert(ctx->mask->length <= (ctx->super->used_cluster_mask_len * sizeof( 2917 struct spdk_blob_md_page) * 8)); 2918 /* The length of the mask must be exactly equal to the total number of clusters */ 2919 assert(ctx->mask->length == ctx->bs->total_clusters); 2920 2921 rc = _spdk_bs_load_mask(&ctx->bs->used_clusters, ctx->mask); 2922 if (rc < 0) { 2923 spdk_free(ctx->mask); 2924 _spdk_bs_load_ctx_fail(seq, ctx, rc); 2925 return; 2926 } 2927 2928 ctx->bs->num_free_clusters = spdk_bit_array_count_clear(ctx->bs->used_clusters); 2929 assert(ctx->bs->num_free_clusters <= ctx->bs->total_clusters); 2930 2931 spdk_free(ctx->mask); 2932 2933 /* Read the used blobids mask */ 2934 mask_size = ctx->super->used_blobid_mask_len * SPDK_BS_PAGE_SIZE; 2935 ctx->mask = spdk_zmalloc(mask_size, 0x1000, NULL, SPDK_ENV_SOCKET_ID_ANY, 2936 SPDK_MALLOC_DMA); 2937 if (!ctx->mask) { 2938 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2939 return; 2940 } 2941 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_start); 2942 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_len); 2943 spdk_bs_sequence_read_dev(seq, ctx->mask, lba, lba_count, 2944 _spdk_bs_load_used_blobids_cpl, ctx); 2945 } 2946 2947 static void 2948 _spdk_bs_load_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2949 { 2950 struct spdk_bs_load_ctx *ctx = cb_arg; 2951 uint64_t lba, lba_count, mask_size; 2952 int rc; 2953 2954 /* The type must be correct */ 2955 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_PAGES); 2956 /* The length of the mask (in bits) must not be greater than the length of the buffer (converted to bits) */ 2957 assert(ctx->mask->length <= (ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE * 2958 8)); 2959 /* The length of the mask must be exactly equal to the size (in pages) of the metadata region */ 2960 assert(ctx->mask->length == ctx->super->md_len); 2961 2962 rc = _spdk_bs_load_mask(&ctx->bs->used_md_pages, ctx->mask); 2963 if (rc < 0) { 2964 spdk_free(ctx->mask); 2965 _spdk_bs_load_ctx_fail(seq, ctx, rc); 2966 return; 2967 } 2968 2969 spdk_free(ctx->mask); 2970 2971 /* Read the used clusters mask */ 2972 mask_size = ctx->super->used_cluster_mask_len * SPDK_BS_PAGE_SIZE; 2973 ctx->mask = spdk_zmalloc(mask_size, 0x1000, NULL, SPDK_ENV_SOCKET_ID_ANY, 2974 SPDK_MALLOC_DMA); 2975 if (!ctx->mask) { 2976 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2977 return; 2978 } 2979 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_start); 2980 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_len); 2981 spdk_bs_sequence_read_dev(seq, ctx->mask, lba, lba_count, 2982 _spdk_bs_load_used_clusters_cpl, ctx); 2983 } 2984 2985 static void 2986 _spdk_bs_load_read_used_pages(spdk_bs_sequence_t *seq, void *cb_arg) 2987 { 2988 struct spdk_bs_load_ctx *ctx = cb_arg; 2989 uint64_t lba, lba_count, mask_size; 2990 2991 /* Read the used pages mask */ 2992 mask_size = ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE; 2993 ctx->mask = spdk_zmalloc(mask_size, 0x1000, NULL, 2994 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 2995 if (!ctx->mask) { 2996 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2997 return; 2998 } 2999 3000 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_start); 3001 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_len); 3002 spdk_bs_sequence_read_dev(seq, ctx->mask, lba, lba_count, 3003 _spdk_bs_load_used_pages_cpl, ctx); 3004 } 3005 3006 static int 3007 _spdk_bs_load_replay_md_parse_page(const struct spdk_blob_md_page *page, struct spdk_blob_store *bs) 3008 { 3009 struct spdk_blob_md_descriptor *desc; 3010 size_t cur_desc = 0; 3011 3012 desc = (struct spdk_blob_md_descriptor *)page->descriptors; 3013 while (cur_desc < sizeof(page->descriptors)) { 3014 if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_PADDING) { 3015 if (desc->length == 0) { 3016 /* If padding and length are 0, this terminates the page */ 3017 break; 3018 } 3019 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_EXTENT_RLE) { 3020 struct spdk_blob_md_descriptor_extent_rle *desc_extent_rle; 3021 unsigned int i, j; 3022 unsigned int cluster_count = 0; 3023 uint32_t cluster_idx; 3024 3025 desc_extent_rle = (struct spdk_blob_md_descriptor_extent_rle *)desc; 3026 3027 for (i = 0; i < desc_extent_rle->length / sizeof(desc_extent_rle->extents[0]); i++) { 3028 for (j = 0; j < desc_extent_rle->extents[i].length; j++) { 3029 cluster_idx = desc_extent_rle->extents[i].cluster_idx; 3030 /* 3031 * cluster_idx = 0 means an unallocated cluster - don't mark that 3032 * in the used cluster map. 3033 */ 3034 if (cluster_idx != 0) { 3035 spdk_bit_array_set(bs->used_clusters, cluster_idx + j); 3036 if (bs->num_free_clusters == 0) { 3037 return -ENOSPC; 3038 } 3039 bs->num_free_clusters--; 3040 } 3041 cluster_count++; 3042 } 3043 } 3044 if (cluster_count == 0) { 3045 return -EINVAL; 3046 } 3047 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR) { 3048 /* Skip this item */ 3049 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR_INTERNAL) { 3050 /* Skip this item */ 3051 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_FLAGS) { 3052 /* Skip this item */ 3053 } else { 3054 /* Error */ 3055 return -EINVAL; 3056 } 3057 /* Advance to the next descriptor */ 3058 cur_desc += sizeof(*desc) + desc->length; 3059 if (cur_desc + sizeof(*desc) > sizeof(page->descriptors)) { 3060 break; 3061 } 3062 desc = (struct spdk_blob_md_descriptor *)((uintptr_t)page->descriptors + cur_desc); 3063 } 3064 return 0; 3065 } 3066 3067 static bool _spdk_bs_load_cur_md_page_valid(struct spdk_bs_load_ctx *ctx) 3068 { 3069 uint32_t crc; 3070 3071 crc = _spdk_blob_md_page_calc_crc(ctx->page); 3072 if (crc != ctx->page->crc) { 3073 return false; 3074 } 3075 3076 if (ctx->page->sequence_num == 0 && 3077 _spdk_bs_page_to_blobid(ctx->cur_page) != ctx->page->id) { 3078 return false; 3079 } 3080 return true; 3081 } 3082 3083 static void 3084 _spdk_bs_load_replay_cur_md_page(spdk_bs_sequence_t *seq, void *cb_arg); 3085 3086 static void 3087 _spdk_bs_load_write_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3088 { 3089 struct spdk_bs_load_ctx *ctx = cb_arg; 3090 3091 _spdk_bs_load_complete(seq, ctx, bserrno); 3092 } 3093 3094 static void 3095 _spdk_bs_load_write_used_blobids_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3096 { 3097 struct spdk_bs_load_ctx *ctx = cb_arg; 3098 3099 spdk_free(ctx->mask); 3100 ctx->mask = NULL; 3101 3102 _spdk_bs_write_used_clusters(seq, cb_arg, _spdk_bs_load_write_used_clusters_cpl); 3103 } 3104 3105 static void 3106 _spdk_bs_load_write_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3107 { 3108 struct spdk_bs_load_ctx *ctx = cb_arg; 3109 3110 spdk_free(ctx->mask); 3111 ctx->mask = NULL; 3112 3113 _spdk_bs_write_used_blobids(seq, cb_arg, _spdk_bs_load_write_used_blobids_cpl); 3114 } 3115 3116 static void 3117 _spdk_bs_load_write_used_md(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3118 { 3119 _spdk_bs_write_used_md(seq, cb_arg, _spdk_bs_load_write_used_pages_cpl); 3120 } 3121 3122 static void 3123 _spdk_bs_load_replay_md_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3124 { 3125 struct spdk_bs_load_ctx *ctx = cb_arg; 3126 uint64_t num_md_clusters; 3127 uint64_t i; 3128 uint32_t page_num; 3129 3130 if (bserrno != 0) { 3131 _spdk_bs_load_ctx_fail(seq, ctx, bserrno); 3132 return; 3133 } 3134 3135 page_num = ctx->cur_page; 3136 if (_spdk_bs_load_cur_md_page_valid(ctx) == true) { 3137 if (ctx->page->sequence_num == 0 || ctx->in_page_chain == true) { 3138 spdk_bit_array_set(ctx->bs->used_md_pages, page_num); 3139 if (ctx->page->sequence_num == 0) { 3140 spdk_bit_array_set(ctx->bs->used_blobids, page_num); 3141 } 3142 if (_spdk_bs_load_replay_md_parse_page(ctx->page, ctx->bs)) { 3143 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 3144 return; 3145 } 3146 if (ctx->page->next != SPDK_INVALID_MD_PAGE) { 3147 ctx->in_page_chain = true; 3148 ctx->cur_page = ctx->page->next; 3149 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 3150 return; 3151 } 3152 } 3153 } 3154 3155 ctx->in_page_chain = false; 3156 3157 do { 3158 ctx->page_index++; 3159 } while (spdk_bit_array_get(ctx->bs->used_md_pages, ctx->page_index) == true); 3160 3161 if (ctx->page_index < ctx->super->md_len) { 3162 ctx->cur_page = ctx->page_index; 3163 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 3164 } else { 3165 /* Claim all of the clusters used by the metadata */ 3166 num_md_clusters = spdk_divide_round_up(ctx->super->md_len, ctx->bs->pages_per_cluster); 3167 for (i = 0; i < num_md_clusters; i++) { 3168 _spdk_bs_claim_cluster(ctx->bs, i); 3169 } 3170 spdk_free(ctx->page); 3171 _spdk_bs_load_write_used_md(seq, ctx, bserrno); 3172 } 3173 } 3174 3175 static void 3176 _spdk_bs_load_replay_cur_md_page(spdk_bs_sequence_t *seq, void *cb_arg) 3177 { 3178 struct spdk_bs_load_ctx *ctx = cb_arg; 3179 uint64_t lba; 3180 3181 assert(ctx->cur_page < ctx->super->md_len); 3182 lba = _spdk_bs_md_page_to_lba(ctx->bs, ctx->cur_page); 3183 spdk_bs_sequence_read_dev(seq, ctx->page, lba, 3184 _spdk_bs_byte_to_lba(ctx->bs, SPDK_BS_PAGE_SIZE), 3185 _spdk_bs_load_replay_md_cpl, ctx); 3186 } 3187 3188 static void 3189 _spdk_bs_load_replay_md(spdk_bs_sequence_t *seq, void *cb_arg) 3190 { 3191 struct spdk_bs_load_ctx *ctx = cb_arg; 3192 3193 ctx->page_index = 0; 3194 ctx->cur_page = 0; 3195 ctx->page = spdk_zmalloc(SPDK_BS_PAGE_SIZE, SPDK_BS_PAGE_SIZE, 3196 NULL, SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 3197 if (!ctx->page) { 3198 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 3199 return; 3200 } 3201 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 3202 } 3203 3204 static void 3205 _spdk_bs_recover(spdk_bs_sequence_t *seq, void *cb_arg) 3206 { 3207 struct spdk_bs_load_ctx *ctx = cb_arg; 3208 int rc; 3209 3210 rc = spdk_bit_array_resize(&ctx->bs->used_md_pages, ctx->super->md_len); 3211 if (rc < 0) { 3212 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 3213 return; 3214 } 3215 3216 rc = spdk_bit_array_resize(&ctx->bs->used_blobids, ctx->super->md_len); 3217 if (rc < 0) { 3218 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 3219 return; 3220 } 3221 3222 rc = spdk_bit_array_resize(&ctx->bs->used_clusters, ctx->bs->total_clusters); 3223 if (rc < 0) { 3224 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 3225 return; 3226 } 3227 3228 ctx->bs->num_free_clusters = ctx->bs->total_clusters; 3229 _spdk_bs_load_replay_md(seq, cb_arg); 3230 } 3231 3232 static void 3233 _spdk_bs_load_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3234 { 3235 struct spdk_bs_load_ctx *ctx = cb_arg; 3236 uint32_t crc; 3237 int rc; 3238 static const char zeros[SPDK_BLOBSTORE_TYPE_LENGTH]; 3239 3240 if (ctx->super->version > SPDK_BS_VERSION || 3241 ctx->super->version < SPDK_BS_INITIAL_VERSION) { 3242 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 3243 return; 3244 } 3245 3246 if (memcmp(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 3247 sizeof(ctx->super->signature)) != 0) { 3248 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 3249 return; 3250 } 3251 3252 crc = _spdk_blob_md_page_calc_crc(ctx->super); 3253 if (crc != ctx->super->crc) { 3254 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 3255 return; 3256 } 3257 3258 if (memcmp(&ctx->bs->bstype, &ctx->super->bstype, SPDK_BLOBSTORE_TYPE_LENGTH) == 0) { 3259 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Bstype matched - loading blobstore\n"); 3260 } else if (memcmp(&ctx->bs->bstype, zeros, SPDK_BLOBSTORE_TYPE_LENGTH) == 0) { 3261 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Bstype wildcard used - loading blobstore regardless bstype\n"); 3262 } else { 3263 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Unexpected bstype\n"); 3264 SPDK_LOGDUMP(SPDK_LOG_BLOB, "Expected:", ctx->bs->bstype.bstype, SPDK_BLOBSTORE_TYPE_LENGTH); 3265 SPDK_LOGDUMP(SPDK_LOG_BLOB, "Found:", ctx->super->bstype.bstype, SPDK_BLOBSTORE_TYPE_LENGTH); 3266 _spdk_bs_load_ctx_fail(seq, ctx, -ENXIO); 3267 return; 3268 } 3269 3270 if (ctx->super->size > ctx->bs->dev->blockcnt * ctx->bs->dev->blocklen) { 3271 SPDK_NOTICELOG("Size mismatch, dev size: %lu, blobstore size: %lu\n", 3272 ctx->bs->dev->blockcnt * ctx->bs->dev->blocklen, ctx->super->size); 3273 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 3274 return; 3275 } 3276 3277 if (ctx->super->size == 0) { 3278 ctx->super->size = ctx->bs->dev->blockcnt * ctx->bs->dev->blocklen; 3279 } 3280 3281 if (ctx->super->io_unit_size == 0) { 3282 ctx->super->io_unit_size = SPDK_BS_PAGE_SIZE; 3283 } 3284 3285 /* Parse the super block */ 3286 ctx->bs->clean = 1; 3287 ctx->bs->cluster_sz = ctx->super->cluster_size; 3288 ctx->bs->total_clusters = ctx->super->size / ctx->super->cluster_size; 3289 ctx->bs->pages_per_cluster = ctx->bs->cluster_sz / SPDK_BS_PAGE_SIZE; 3290 ctx->bs->io_unit_size = ctx->super->io_unit_size; 3291 rc = spdk_bit_array_resize(&ctx->bs->used_clusters, ctx->bs->total_clusters); 3292 if (rc < 0) { 3293 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 3294 return; 3295 } 3296 ctx->bs->md_start = ctx->super->md_start; 3297 ctx->bs->md_len = ctx->super->md_len; 3298 ctx->bs->total_data_clusters = ctx->bs->total_clusters - spdk_divide_round_up( 3299 ctx->bs->md_start + ctx->bs->md_len, ctx->bs->pages_per_cluster); 3300 ctx->bs->super_blob = ctx->super->super_blob; 3301 memcpy(&ctx->bs->bstype, &ctx->super->bstype, sizeof(ctx->super->bstype)); 3302 3303 if (ctx->super->used_blobid_mask_len == 0 || ctx->super->clean == 0) { 3304 _spdk_bs_recover(seq, ctx); 3305 } else { 3306 _spdk_bs_load_read_used_pages(seq, ctx); 3307 } 3308 } 3309 3310 void 3311 spdk_bs_load(struct spdk_bs_dev *dev, struct spdk_bs_opts *o, 3312 spdk_bs_op_with_handle_complete cb_fn, void *cb_arg) 3313 { 3314 struct spdk_blob_store *bs; 3315 struct spdk_bs_cpl cpl; 3316 spdk_bs_sequence_t *seq; 3317 struct spdk_bs_load_ctx *ctx; 3318 struct spdk_bs_opts opts = {}; 3319 int err; 3320 3321 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Loading blobstore from dev %p\n", dev); 3322 3323 if ((SPDK_BS_PAGE_SIZE % dev->blocklen) != 0) { 3324 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "unsupported dev block length of %d\n", dev->blocklen); 3325 dev->destroy(dev); 3326 cb_fn(cb_arg, NULL, -EINVAL); 3327 return; 3328 } 3329 3330 if (o) { 3331 opts = *o; 3332 } else { 3333 spdk_bs_opts_init(&opts); 3334 } 3335 3336 if (opts.max_md_ops == 0 || opts.max_channel_ops == 0) { 3337 dev->destroy(dev); 3338 cb_fn(cb_arg, NULL, -EINVAL); 3339 return; 3340 } 3341 3342 err = _spdk_bs_alloc(dev, &opts, &bs); 3343 if (err) { 3344 dev->destroy(dev); 3345 cb_fn(cb_arg, NULL, err); 3346 return; 3347 } 3348 3349 ctx = calloc(1, sizeof(*ctx)); 3350 if (!ctx) { 3351 _spdk_bs_free(bs); 3352 cb_fn(cb_arg, NULL, -ENOMEM); 3353 return; 3354 } 3355 3356 ctx->bs = bs; 3357 ctx->iter_cb_fn = opts.iter_cb_fn; 3358 ctx->iter_cb_arg = opts.iter_cb_arg; 3359 3360 /* Allocate memory for the super block */ 3361 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL, 3362 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 3363 if (!ctx->super) { 3364 free(ctx); 3365 _spdk_bs_free(bs); 3366 cb_fn(cb_arg, NULL, -ENOMEM); 3367 return; 3368 } 3369 3370 cpl.type = SPDK_BS_CPL_TYPE_BS_HANDLE; 3371 cpl.u.bs_handle.cb_fn = cb_fn; 3372 cpl.u.bs_handle.cb_arg = cb_arg; 3373 cpl.u.bs_handle.bs = bs; 3374 3375 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3376 if (!seq) { 3377 spdk_free(ctx->super); 3378 free(ctx); 3379 _spdk_bs_free(bs); 3380 cb_fn(cb_arg, NULL, -ENOMEM); 3381 return; 3382 } 3383 3384 /* Read the super block */ 3385 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 3386 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 3387 _spdk_bs_load_super_cpl, ctx); 3388 } 3389 3390 /* END spdk_bs_load */ 3391 3392 /* START spdk_bs_dump */ 3393 3394 struct spdk_bs_dump_ctx { 3395 struct spdk_blob_store *bs; 3396 struct spdk_bs_super_block *super; 3397 uint32_t cur_page; 3398 struct spdk_blob_md_page *page; 3399 spdk_bs_sequence_t *seq; 3400 FILE *fp; 3401 spdk_bs_dump_print_xattr print_xattr_fn; 3402 char xattr_name[4096]; 3403 }; 3404 3405 static void 3406 _spdk_bs_dump_finish(spdk_bs_sequence_t *seq, struct spdk_bs_dump_ctx *ctx, int bserrno) 3407 { 3408 spdk_free(ctx->super); 3409 3410 /* 3411 * We need to defer calling spdk_bs_call_cpl() until after 3412 * dev destruction, so tuck these away for later use. 3413 */ 3414 ctx->bs->unload_err = bserrno; 3415 memcpy(&ctx->bs->unload_cpl, &seq->cpl, sizeof(struct spdk_bs_cpl)); 3416 seq->cpl.type = SPDK_BS_CPL_TYPE_NONE; 3417 3418 spdk_bs_sequence_finish(seq, 0); 3419 _spdk_bs_free(ctx->bs); 3420 free(ctx); 3421 } 3422 3423 static void _spdk_bs_dump_read_md_page(spdk_bs_sequence_t *seq, void *cb_arg); 3424 3425 static void 3426 _spdk_bs_dump_print_md_page(struct spdk_bs_dump_ctx *ctx) 3427 { 3428 uint32_t page_idx = ctx->cur_page; 3429 struct spdk_blob_md_page *page = ctx->page; 3430 struct spdk_blob_md_descriptor *desc; 3431 size_t cur_desc = 0; 3432 uint32_t crc; 3433 3434 fprintf(ctx->fp, "=========\n"); 3435 fprintf(ctx->fp, "Metadata Page Index: %" PRIu32 " (0x%" PRIx32 ")\n", page_idx, page_idx); 3436 fprintf(ctx->fp, "Blob ID: 0x%" PRIx64 "\n", page->id); 3437 3438 crc = _spdk_blob_md_page_calc_crc(page); 3439 fprintf(ctx->fp, "CRC: 0x%" PRIx32 " (%s)\n", page->crc, crc == page->crc ? "OK" : "Mismatch"); 3440 3441 desc = (struct spdk_blob_md_descriptor *)page->descriptors; 3442 while (cur_desc < sizeof(page->descriptors)) { 3443 if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_PADDING) { 3444 if (desc->length == 0) { 3445 /* If padding and length are 0, this terminates the page */ 3446 break; 3447 } 3448 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_EXTENT_RLE) { 3449 struct spdk_blob_md_descriptor_extent_rle *desc_extent_rle; 3450 unsigned int i; 3451 3452 desc_extent_rle = (struct spdk_blob_md_descriptor_extent_rle *)desc; 3453 3454 for (i = 0; i < desc_extent_rle->length / sizeof(desc_extent_rle->extents[0]); i++) { 3455 if (desc_extent_rle->extents[i].cluster_idx != 0) { 3456 fprintf(ctx->fp, "Allocated Extent - Start: %" PRIu32, 3457 desc_extent_rle->extents[i].cluster_idx); 3458 } else { 3459 fprintf(ctx->fp, "Unallocated Extent - "); 3460 } 3461 fprintf(ctx->fp, " Length: %" PRIu32, desc_extent_rle->extents[i].length); 3462 fprintf(ctx->fp, "\n"); 3463 } 3464 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR) { 3465 struct spdk_blob_md_descriptor_xattr *desc_xattr; 3466 uint32_t i; 3467 3468 desc_xattr = (struct spdk_blob_md_descriptor_xattr *)desc; 3469 3470 if (desc_xattr->length != 3471 sizeof(desc_xattr->name_length) + sizeof(desc_xattr->value_length) + 3472 desc_xattr->name_length + desc_xattr->value_length) { 3473 } 3474 3475 memcpy(ctx->xattr_name, desc_xattr->name, desc_xattr->name_length); 3476 ctx->xattr_name[desc_xattr->name_length] = '\0'; 3477 fprintf(ctx->fp, "XATTR: name = \"%s\"\n", ctx->xattr_name); 3478 fprintf(ctx->fp, " value = \""); 3479 ctx->print_xattr_fn(ctx->fp, ctx->super->bstype.bstype, ctx->xattr_name, 3480 (void *)((uintptr_t)desc_xattr->name + desc_xattr->name_length), 3481 desc_xattr->value_length); 3482 fprintf(ctx->fp, "\"\n"); 3483 for (i = 0; i < desc_xattr->value_length; i++) { 3484 if (i % 16 == 0) { 3485 fprintf(ctx->fp, " "); 3486 } 3487 fprintf(ctx->fp, "%02" PRIx8 " ", *((uint8_t *)desc_xattr->name + desc_xattr->name_length + i)); 3488 if ((i + 1) % 16 == 0) { 3489 fprintf(ctx->fp, "\n"); 3490 } 3491 } 3492 if (i % 16 != 0) { 3493 fprintf(ctx->fp, "\n"); 3494 } 3495 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR_INTERNAL) { 3496 /* TODO */ 3497 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_FLAGS) { 3498 /* TODO */ 3499 } else { 3500 /* Error */ 3501 } 3502 /* Advance to the next descriptor */ 3503 cur_desc += sizeof(*desc) + desc->length; 3504 if (cur_desc + sizeof(*desc) > sizeof(page->descriptors)) { 3505 break; 3506 } 3507 desc = (struct spdk_blob_md_descriptor *)((uintptr_t)page->descriptors + cur_desc); 3508 } 3509 } 3510 3511 static void 3512 _spdk_bs_dump_read_md_page_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3513 { 3514 struct spdk_bs_dump_ctx *ctx = cb_arg; 3515 3516 if (bserrno != 0) { 3517 _spdk_bs_dump_finish(seq, ctx, bserrno); 3518 return; 3519 } 3520 3521 if (ctx->page->id != 0) { 3522 _spdk_bs_dump_print_md_page(ctx); 3523 } 3524 3525 ctx->cur_page++; 3526 3527 if (ctx->cur_page < ctx->super->md_len) { 3528 _spdk_bs_dump_read_md_page(seq, cb_arg); 3529 } else { 3530 spdk_free(ctx->page); 3531 _spdk_bs_dump_finish(seq, ctx, 0); 3532 } 3533 } 3534 3535 static void 3536 _spdk_bs_dump_read_md_page(spdk_bs_sequence_t *seq, void *cb_arg) 3537 { 3538 struct spdk_bs_dump_ctx *ctx = cb_arg; 3539 uint64_t lba; 3540 3541 assert(ctx->cur_page < ctx->super->md_len); 3542 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->md_start + ctx->cur_page); 3543 spdk_bs_sequence_read_dev(seq, ctx->page, lba, 3544 _spdk_bs_byte_to_lba(ctx->bs, SPDK_BS_PAGE_SIZE), 3545 _spdk_bs_dump_read_md_page_cpl, ctx); 3546 } 3547 3548 static void 3549 _spdk_bs_dump_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3550 { 3551 struct spdk_bs_dump_ctx *ctx = cb_arg; 3552 3553 fprintf(ctx->fp, "Signature: \"%.8s\" ", ctx->super->signature); 3554 if (memcmp(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 3555 sizeof(ctx->super->signature)) != 0) { 3556 fprintf(ctx->fp, "(Mismatch)\n"); 3557 _spdk_bs_dump_finish(seq, ctx, bserrno); 3558 return; 3559 } else { 3560 fprintf(ctx->fp, "(OK)\n"); 3561 } 3562 fprintf(ctx->fp, "Version: %" PRIu32 "\n", ctx->super->version); 3563 fprintf(ctx->fp, "CRC: 0x%x (%s)\n", ctx->super->crc, 3564 (ctx->super->crc == _spdk_blob_md_page_calc_crc(ctx->super)) ? "OK" : "Mismatch"); 3565 fprintf(ctx->fp, "Blobstore Type: %.*s\n", SPDK_BLOBSTORE_TYPE_LENGTH, ctx->super->bstype.bstype); 3566 fprintf(ctx->fp, "Cluster Size: %" PRIu32 "\n", ctx->super->cluster_size); 3567 fprintf(ctx->fp, "Super Blob ID: "); 3568 if (ctx->super->super_blob == SPDK_BLOBID_INVALID) { 3569 fprintf(ctx->fp, "(None)\n"); 3570 } else { 3571 fprintf(ctx->fp, "%" PRIu64 "\n", ctx->super->super_blob); 3572 } 3573 fprintf(ctx->fp, "Clean: %" PRIu32 "\n", ctx->super->clean); 3574 fprintf(ctx->fp, "Used Metadata Page Mask Start: %" PRIu32 "\n", ctx->super->used_page_mask_start); 3575 fprintf(ctx->fp, "Used Metadata Page Mask Length: %" PRIu32 "\n", ctx->super->used_page_mask_len); 3576 fprintf(ctx->fp, "Used Cluster Mask Start: %" PRIu32 "\n", ctx->super->used_cluster_mask_start); 3577 fprintf(ctx->fp, "Used Cluster Mask Length: %" PRIu32 "\n", ctx->super->used_cluster_mask_len); 3578 fprintf(ctx->fp, "Used Blob ID Mask Start: %" PRIu32 "\n", ctx->super->used_blobid_mask_start); 3579 fprintf(ctx->fp, "Used Blob ID Mask Length: %" PRIu32 "\n", ctx->super->used_blobid_mask_len); 3580 fprintf(ctx->fp, "Metadata Start: %" PRIu32 "\n", ctx->super->md_start); 3581 fprintf(ctx->fp, "Metadata Length: %" PRIu32 "\n", ctx->super->md_len); 3582 3583 ctx->cur_page = 0; 3584 ctx->page = spdk_zmalloc(SPDK_BS_PAGE_SIZE, SPDK_BS_PAGE_SIZE, 3585 NULL, SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 3586 if (!ctx->page) { 3587 _spdk_bs_dump_finish(seq, ctx, -ENOMEM); 3588 return; 3589 } 3590 _spdk_bs_dump_read_md_page(seq, cb_arg); 3591 } 3592 3593 void 3594 spdk_bs_dump(struct spdk_bs_dev *dev, FILE *fp, spdk_bs_dump_print_xattr print_xattr_fn, 3595 spdk_bs_op_complete cb_fn, void *cb_arg) 3596 { 3597 struct spdk_blob_store *bs; 3598 struct spdk_bs_cpl cpl; 3599 spdk_bs_sequence_t *seq; 3600 struct spdk_bs_dump_ctx *ctx; 3601 struct spdk_bs_opts opts = {}; 3602 int err; 3603 3604 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Dumping blobstore from dev %p\n", dev); 3605 3606 spdk_bs_opts_init(&opts); 3607 3608 err = _spdk_bs_alloc(dev, &opts, &bs); 3609 if (err) { 3610 dev->destroy(dev); 3611 cb_fn(cb_arg, err); 3612 return; 3613 } 3614 3615 ctx = calloc(1, sizeof(*ctx)); 3616 if (!ctx) { 3617 _spdk_bs_free(bs); 3618 cb_fn(cb_arg, -ENOMEM); 3619 return; 3620 } 3621 3622 ctx->bs = bs; 3623 ctx->fp = fp; 3624 ctx->print_xattr_fn = print_xattr_fn; 3625 3626 /* Allocate memory for the super block */ 3627 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL, 3628 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 3629 if (!ctx->super) { 3630 free(ctx); 3631 _spdk_bs_free(bs); 3632 cb_fn(cb_arg, -ENOMEM); 3633 return; 3634 } 3635 3636 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 3637 cpl.u.bs_basic.cb_fn = cb_fn; 3638 cpl.u.bs_basic.cb_arg = cb_arg; 3639 3640 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3641 if (!seq) { 3642 spdk_free(ctx->super); 3643 free(ctx); 3644 _spdk_bs_free(bs); 3645 cb_fn(cb_arg, -ENOMEM); 3646 return; 3647 } 3648 3649 /* Read the super block */ 3650 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 3651 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 3652 _spdk_bs_dump_super_cpl, ctx); 3653 } 3654 3655 /* END spdk_bs_dump */ 3656 3657 /* START spdk_bs_init */ 3658 3659 struct spdk_bs_init_ctx { 3660 struct spdk_blob_store *bs; 3661 struct spdk_bs_super_block *super; 3662 }; 3663 3664 static void 3665 _spdk_bs_init_persist_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3666 { 3667 struct spdk_bs_init_ctx *ctx = cb_arg; 3668 3669 spdk_free(ctx->super); 3670 free(ctx); 3671 3672 spdk_bs_sequence_finish(seq, bserrno); 3673 } 3674 3675 static void 3676 _spdk_bs_init_trim_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3677 { 3678 struct spdk_bs_init_ctx *ctx = cb_arg; 3679 3680 /* Write super block */ 3681 spdk_bs_sequence_write_dev(seq, ctx->super, _spdk_bs_page_to_lba(ctx->bs, 0), 3682 _spdk_bs_byte_to_lba(ctx->bs, sizeof(*ctx->super)), 3683 _spdk_bs_init_persist_super_cpl, ctx); 3684 } 3685 3686 void 3687 spdk_bs_init(struct spdk_bs_dev *dev, struct spdk_bs_opts *o, 3688 spdk_bs_op_with_handle_complete cb_fn, void *cb_arg) 3689 { 3690 struct spdk_bs_init_ctx *ctx; 3691 struct spdk_blob_store *bs; 3692 struct spdk_bs_cpl cpl; 3693 spdk_bs_sequence_t *seq; 3694 spdk_bs_batch_t *batch; 3695 uint64_t num_md_lba; 3696 uint64_t num_md_pages; 3697 uint64_t num_md_clusters; 3698 uint32_t i; 3699 struct spdk_bs_opts opts = {}; 3700 int rc; 3701 3702 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Initializing blobstore on dev %p\n", dev); 3703 3704 if ((SPDK_BS_PAGE_SIZE % dev->blocklen) != 0) { 3705 SPDK_ERRLOG("unsupported dev block length of %d\n", 3706 dev->blocklen); 3707 dev->destroy(dev); 3708 cb_fn(cb_arg, NULL, -EINVAL); 3709 return; 3710 } 3711 3712 if (o) { 3713 opts = *o; 3714 } else { 3715 spdk_bs_opts_init(&opts); 3716 } 3717 3718 if (_spdk_bs_opts_verify(&opts) != 0) { 3719 dev->destroy(dev); 3720 cb_fn(cb_arg, NULL, -EINVAL); 3721 return; 3722 } 3723 3724 rc = _spdk_bs_alloc(dev, &opts, &bs); 3725 if (rc) { 3726 dev->destroy(dev); 3727 cb_fn(cb_arg, NULL, rc); 3728 return; 3729 } 3730 3731 if (opts.num_md_pages == SPDK_BLOB_OPTS_NUM_MD_PAGES) { 3732 /* By default, allocate 1 page per cluster. 3733 * Technically, this over-allocates metadata 3734 * because more metadata will reduce the number 3735 * of usable clusters. This can be addressed with 3736 * more complex math in the future. 3737 */ 3738 bs->md_len = bs->total_clusters; 3739 } else { 3740 bs->md_len = opts.num_md_pages; 3741 } 3742 rc = spdk_bit_array_resize(&bs->used_md_pages, bs->md_len); 3743 if (rc < 0) { 3744 _spdk_bs_free(bs); 3745 cb_fn(cb_arg, NULL, -ENOMEM); 3746 return; 3747 } 3748 3749 rc = spdk_bit_array_resize(&bs->used_blobids, bs->md_len); 3750 if (rc < 0) { 3751 _spdk_bs_free(bs); 3752 cb_fn(cb_arg, NULL, -ENOMEM); 3753 return; 3754 } 3755 3756 ctx = calloc(1, sizeof(*ctx)); 3757 if (!ctx) { 3758 _spdk_bs_free(bs); 3759 cb_fn(cb_arg, NULL, -ENOMEM); 3760 return; 3761 } 3762 3763 ctx->bs = bs; 3764 3765 /* Allocate memory for the super block */ 3766 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL, 3767 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 3768 if (!ctx->super) { 3769 free(ctx); 3770 _spdk_bs_free(bs); 3771 cb_fn(cb_arg, NULL, -ENOMEM); 3772 return; 3773 } 3774 memcpy(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 3775 sizeof(ctx->super->signature)); 3776 ctx->super->version = SPDK_BS_VERSION; 3777 ctx->super->length = sizeof(*ctx->super); 3778 ctx->super->super_blob = bs->super_blob; 3779 ctx->super->clean = 0; 3780 ctx->super->cluster_size = bs->cluster_sz; 3781 ctx->super->io_unit_size = bs->io_unit_size; 3782 memcpy(&ctx->super->bstype, &bs->bstype, sizeof(bs->bstype)); 3783 3784 /* Calculate how many pages the metadata consumes at the front 3785 * of the disk. 3786 */ 3787 3788 /* The super block uses 1 page */ 3789 num_md_pages = 1; 3790 3791 /* The used_md_pages mask requires 1 bit per metadata page, rounded 3792 * up to the nearest page, plus a header. 3793 */ 3794 ctx->super->used_page_mask_start = num_md_pages; 3795 ctx->super->used_page_mask_len = spdk_divide_round_up(sizeof(struct spdk_bs_md_mask) + 3796 spdk_divide_round_up(bs->md_len, 8), 3797 SPDK_BS_PAGE_SIZE); 3798 num_md_pages += ctx->super->used_page_mask_len; 3799 3800 /* The used_clusters mask requires 1 bit per cluster, rounded 3801 * up to the nearest page, plus a header. 3802 */ 3803 ctx->super->used_cluster_mask_start = num_md_pages; 3804 ctx->super->used_cluster_mask_len = spdk_divide_round_up(sizeof(struct spdk_bs_md_mask) + 3805 spdk_divide_round_up(bs->total_clusters, 8), 3806 SPDK_BS_PAGE_SIZE); 3807 num_md_pages += ctx->super->used_cluster_mask_len; 3808 3809 /* The used_blobids mask requires 1 bit per metadata page, rounded 3810 * up to the nearest page, plus a header. 3811 */ 3812 ctx->super->used_blobid_mask_start = num_md_pages; 3813 ctx->super->used_blobid_mask_len = spdk_divide_round_up(sizeof(struct spdk_bs_md_mask) + 3814 spdk_divide_round_up(bs->md_len, 8), 3815 SPDK_BS_PAGE_SIZE); 3816 num_md_pages += ctx->super->used_blobid_mask_len; 3817 3818 /* The metadata region size was chosen above */ 3819 ctx->super->md_start = bs->md_start = num_md_pages; 3820 ctx->super->md_len = bs->md_len; 3821 num_md_pages += bs->md_len; 3822 3823 num_md_lba = _spdk_bs_page_to_lba(bs, num_md_pages); 3824 3825 ctx->super->size = dev->blockcnt * dev->blocklen; 3826 3827 ctx->super->crc = _spdk_blob_md_page_calc_crc(ctx->super); 3828 3829 num_md_clusters = spdk_divide_round_up(num_md_pages, bs->pages_per_cluster); 3830 if (num_md_clusters > bs->total_clusters) { 3831 SPDK_ERRLOG("Blobstore metadata cannot use more clusters than is available, " 3832 "please decrease number of pages reserved for metadata " 3833 "or increase cluster size.\n"); 3834 spdk_free(ctx->super); 3835 free(ctx); 3836 _spdk_bs_free(bs); 3837 cb_fn(cb_arg, NULL, -ENOMEM); 3838 return; 3839 } 3840 /* Claim all of the clusters used by the metadata */ 3841 for (i = 0; i < num_md_clusters; i++) { 3842 _spdk_bs_claim_cluster(bs, i); 3843 } 3844 3845 bs->total_data_clusters = bs->num_free_clusters; 3846 3847 cpl.type = SPDK_BS_CPL_TYPE_BS_HANDLE; 3848 cpl.u.bs_handle.cb_fn = cb_fn; 3849 cpl.u.bs_handle.cb_arg = cb_arg; 3850 cpl.u.bs_handle.bs = bs; 3851 3852 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3853 if (!seq) { 3854 spdk_free(ctx->super); 3855 free(ctx); 3856 _spdk_bs_free(bs); 3857 cb_fn(cb_arg, NULL, -ENOMEM); 3858 return; 3859 } 3860 3861 batch = spdk_bs_sequence_to_batch(seq, _spdk_bs_init_trim_cpl, ctx); 3862 3863 /* Clear metadata space */ 3864 spdk_bs_batch_write_zeroes_dev(batch, 0, num_md_lba); 3865 3866 switch (opts.clear_method) { 3867 case BS_CLEAR_WITH_UNMAP: 3868 /* Trim data clusters */ 3869 spdk_bs_batch_unmap_dev(batch, num_md_lba, ctx->bs->dev->blockcnt - num_md_lba); 3870 break; 3871 case BS_CLEAR_WITH_WRITE_ZEROES: 3872 /* Write_zeroes to data clusters */ 3873 spdk_bs_batch_write_zeroes_dev(batch, num_md_lba, ctx->bs->dev->blockcnt - num_md_lba); 3874 break; 3875 case BS_CLEAR_WITH_NONE: 3876 default: 3877 break; 3878 } 3879 3880 spdk_bs_batch_close(batch); 3881 } 3882 3883 /* END spdk_bs_init */ 3884 3885 /* START spdk_bs_destroy */ 3886 3887 static void 3888 _spdk_bs_destroy_trim_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3889 { 3890 struct spdk_bs_init_ctx *ctx = cb_arg; 3891 struct spdk_blob_store *bs = ctx->bs; 3892 3893 /* 3894 * We need to defer calling spdk_bs_call_cpl() until after 3895 * dev destruction, so tuck these away for later use. 3896 */ 3897 bs->unload_err = bserrno; 3898 memcpy(&bs->unload_cpl, &seq->cpl, sizeof(struct spdk_bs_cpl)); 3899 seq->cpl.type = SPDK_BS_CPL_TYPE_NONE; 3900 3901 spdk_bs_sequence_finish(seq, bserrno); 3902 3903 _spdk_bs_free(bs); 3904 free(ctx); 3905 } 3906 3907 void 3908 spdk_bs_destroy(struct spdk_blob_store *bs, spdk_bs_op_complete cb_fn, 3909 void *cb_arg) 3910 { 3911 struct spdk_bs_cpl cpl; 3912 spdk_bs_sequence_t *seq; 3913 struct spdk_bs_init_ctx *ctx; 3914 3915 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Destroying blobstore\n"); 3916 3917 if (!TAILQ_EMPTY(&bs->blobs)) { 3918 SPDK_ERRLOG("Blobstore still has open blobs\n"); 3919 cb_fn(cb_arg, -EBUSY); 3920 return; 3921 } 3922 3923 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 3924 cpl.u.bs_basic.cb_fn = cb_fn; 3925 cpl.u.bs_basic.cb_arg = cb_arg; 3926 3927 ctx = calloc(1, sizeof(*ctx)); 3928 if (!ctx) { 3929 cb_fn(cb_arg, -ENOMEM); 3930 return; 3931 } 3932 3933 ctx->bs = bs; 3934 3935 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3936 if (!seq) { 3937 free(ctx); 3938 cb_fn(cb_arg, -ENOMEM); 3939 return; 3940 } 3941 3942 /* Write zeroes to the super block */ 3943 spdk_bs_sequence_write_zeroes_dev(seq, 3944 _spdk_bs_page_to_lba(bs, 0), 3945 _spdk_bs_byte_to_lba(bs, sizeof(struct spdk_bs_super_block)), 3946 _spdk_bs_destroy_trim_cpl, ctx); 3947 } 3948 3949 /* END spdk_bs_destroy */ 3950 3951 /* START spdk_bs_unload */ 3952 3953 static void 3954 _spdk_bs_unload_write_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3955 { 3956 struct spdk_bs_load_ctx *ctx = cb_arg; 3957 3958 spdk_free(ctx->super); 3959 3960 /* 3961 * We need to defer calling spdk_bs_call_cpl() until after 3962 * dev destruction, so tuck these away for later use. 3963 */ 3964 ctx->bs->unload_err = bserrno; 3965 memcpy(&ctx->bs->unload_cpl, &seq->cpl, sizeof(struct spdk_bs_cpl)); 3966 seq->cpl.type = SPDK_BS_CPL_TYPE_NONE; 3967 3968 spdk_bs_sequence_finish(seq, bserrno); 3969 3970 _spdk_bs_free(ctx->bs); 3971 free(ctx); 3972 } 3973 3974 static void 3975 _spdk_bs_unload_write_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3976 { 3977 struct spdk_bs_load_ctx *ctx = cb_arg; 3978 3979 spdk_free(ctx->mask); 3980 ctx->super->clean = 1; 3981 3982 _spdk_bs_write_super(seq, ctx->bs, ctx->super, _spdk_bs_unload_write_super_cpl, ctx); 3983 } 3984 3985 static void 3986 _spdk_bs_unload_write_used_blobids_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3987 { 3988 struct spdk_bs_load_ctx *ctx = cb_arg; 3989 3990 spdk_free(ctx->mask); 3991 ctx->mask = NULL; 3992 3993 _spdk_bs_write_used_clusters(seq, cb_arg, _spdk_bs_unload_write_used_clusters_cpl); 3994 } 3995 3996 static void 3997 _spdk_bs_unload_write_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3998 { 3999 struct spdk_bs_load_ctx *ctx = cb_arg; 4000 4001 spdk_free(ctx->mask); 4002 ctx->mask = NULL; 4003 4004 _spdk_bs_write_used_blobids(seq, cb_arg, _spdk_bs_unload_write_used_blobids_cpl); 4005 } 4006 4007 static void 4008 _spdk_bs_unload_read_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 4009 { 4010 _spdk_bs_write_used_md(seq, cb_arg, _spdk_bs_unload_write_used_pages_cpl); 4011 } 4012 4013 void 4014 spdk_bs_unload(struct spdk_blob_store *bs, spdk_bs_op_complete cb_fn, void *cb_arg) 4015 { 4016 struct spdk_bs_cpl cpl; 4017 spdk_bs_sequence_t *seq; 4018 struct spdk_bs_load_ctx *ctx; 4019 4020 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Syncing blobstore\n"); 4021 4022 if (!TAILQ_EMPTY(&bs->blobs)) { 4023 SPDK_ERRLOG("Blobstore still has open blobs\n"); 4024 cb_fn(cb_arg, -EBUSY); 4025 return; 4026 } 4027 4028 ctx = calloc(1, sizeof(*ctx)); 4029 if (!ctx) { 4030 cb_fn(cb_arg, -ENOMEM); 4031 return; 4032 } 4033 4034 ctx->bs = bs; 4035 4036 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL, 4037 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 4038 if (!ctx->super) { 4039 free(ctx); 4040 cb_fn(cb_arg, -ENOMEM); 4041 return; 4042 } 4043 4044 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 4045 cpl.u.bs_basic.cb_fn = cb_fn; 4046 cpl.u.bs_basic.cb_arg = cb_arg; 4047 4048 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 4049 if (!seq) { 4050 spdk_free(ctx->super); 4051 free(ctx); 4052 cb_fn(cb_arg, -ENOMEM); 4053 return; 4054 } 4055 4056 /* Read super block */ 4057 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 4058 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 4059 _spdk_bs_unload_read_super_cpl, ctx); 4060 } 4061 4062 /* END spdk_bs_unload */ 4063 4064 /* START spdk_bs_set_super */ 4065 4066 struct spdk_bs_set_super_ctx { 4067 struct spdk_blob_store *bs; 4068 struct spdk_bs_super_block *super; 4069 }; 4070 4071 static void 4072 _spdk_bs_set_super_write_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 4073 { 4074 struct spdk_bs_set_super_ctx *ctx = cb_arg; 4075 4076 if (bserrno != 0) { 4077 SPDK_ERRLOG("Unable to write to super block of blobstore\n"); 4078 } 4079 4080 spdk_free(ctx->super); 4081 4082 spdk_bs_sequence_finish(seq, bserrno); 4083 4084 free(ctx); 4085 } 4086 4087 static void 4088 _spdk_bs_set_super_read_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 4089 { 4090 struct spdk_bs_set_super_ctx *ctx = cb_arg; 4091 4092 if (bserrno != 0) { 4093 SPDK_ERRLOG("Unable to read super block of blobstore\n"); 4094 spdk_free(ctx->super); 4095 spdk_bs_sequence_finish(seq, bserrno); 4096 free(ctx); 4097 return; 4098 } 4099 4100 _spdk_bs_write_super(seq, ctx->bs, ctx->super, _spdk_bs_set_super_write_cpl, ctx); 4101 } 4102 4103 void 4104 spdk_bs_set_super(struct spdk_blob_store *bs, spdk_blob_id blobid, 4105 spdk_bs_op_complete cb_fn, void *cb_arg) 4106 { 4107 struct spdk_bs_cpl cpl; 4108 spdk_bs_sequence_t *seq; 4109 struct spdk_bs_set_super_ctx *ctx; 4110 4111 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Setting super blob id on blobstore\n"); 4112 4113 ctx = calloc(1, sizeof(*ctx)); 4114 if (!ctx) { 4115 cb_fn(cb_arg, -ENOMEM); 4116 return; 4117 } 4118 4119 ctx->bs = bs; 4120 4121 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL, 4122 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 4123 if (!ctx->super) { 4124 free(ctx); 4125 cb_fn(cb_arg, -ENOMEM); 4126 return; 4127 } 4128 4129 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 4130 cpl.u.bs_basic.cb_fn = cb_fn; 4131 cpl.u.bs_basic.cb_arg = cb_arg; 4132 4133 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 4134 if (!seq) { 4135 spdk_free(ctx->super); 4136 free(ctx); 4137 cb_fn(cb_arg, -ENOMEM); 4138 return; 4139 } 4140 4141 bs->super_blob = blobid; 4142 4143 /* Read super block */ 4144 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 4145 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 4146 _spdk_bs_set_super_read_cpl, ctx); 4147 } 4148 4149 /* END spdk_bs_set_super */ 4150 4151 void 4152 spdk_bs_get_super(struct spdk_blob_store *bs, 4153 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 4154 { 4155 if (bs->super_blob == SPDK_BLOBID_INVALID) { 4156 cb_fn(cb_arg, SPDK_BLOBID_INVALID, -ENOENT); 4157 } else { 4158 cb_fn(cb_arg, bs->super_blob, 0); 4159 } 4160 } 4161 4162 uint64_t 4163 spdk_bs_get_cluster_size(struct spdk_blob_store *bs) 4164 { 4165 return bs->cluster_sz; 4166 } 4167 4168 uint64_t 4169 spdk_bs_get_page_size(struct spdk_blob_store *bs) 4170 { 4171 return SPDK_BS_PAGE_SIZE; 4172 } 4173 4174 uint64_t 4175 spdk_bs_get_io_unit_size(struct spdk_blob_store *bs) 4176 { 4177 return bs->io_unit_size; 4178 } 4179 4180 uint64_t 4181 spdk_bs_free_cluster_count(struct spdk_blob_store *bs) 4182 { 4183 return bs->num_free_clusters; 4184 } 4185 4186 uint64_t 4187 spdk_bs_total_data_cluster_count(struct spdk_blob_store *bs) 4188 { 4189 return bs->total_data_clusters; 4190 } 4191 4192 static int 4193 spdk_bs_register_md_thread(struct spdk_blob_store *bs) 4194 { 4195 bs->md_channel = spdk_get_io_channel(bs); 4196 if (!bs->md_channel) { 4197 SPDK_ERRLOG("Failed to get IO channel.\n"); 4198 return -1; 4199 } 4200 4201 return 0; 4202 } 4203 4204 static int 4205 spdk_bs_unregister_md_thread(struct spdk_blob_store *bs) 4206 { 4207 spdk_put_io_channel(bs->md_channel); 4208 4209 return 0; 4210 } 4211 4212 spdk_blob_id spdk_blob_get_id(struct spdk_blob *blob) 4213 { 4214 assert(blob != NULL); 4215 4216 return blob->id; 4217 } 4218 4219 uint64_t spdk_blob_get_num_pages(struct spdk_blob *blob) 4220 { 4221 assert(blob != NULL); 4222 4223 return _spdk_bs_cluster_to_page(blob->bs, blob->active.num_clusters); 4224 } 4225 4226 uint64_t spdk_blob_get_num_io_units(struct spdk_blob *blob) 4227 { 4228 assert(blob != NULL); 4229 4230 return spdk_blob_get_num_pages(blob) * _spdk_bs_io_unit_per_page(blob->bs); 4231 } 4232 4233 uint64_t spdk_blob_get_num_clusters(struct spdk_blob *blob) 4234 { 4235 assert(blob != NULL); 4236 4237 return blob->active.num_clusters; 4238 } 4239 4240 /* START spdk_bs_create_blob */ 4241 4242 static void 4243 _spdk_bs_create_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 4244 { 4245 struct spdk_blob *blob = cb_arg; 4246 4247 _spdk_blob_free(blob); 4248 4249 spdk_bs_sequence_finish(seq, bserrno); 4250 } 4251 4252 static int 4253 _spdk_blob_set_xattrs(struct spdk_blob *blob, const struct spdk_blob_xattr_opts *xattrs, 4254 bool internal) 4255 { 4256 uint64_t i; 4257 size_t value_len = 0; 4258 int rc; 4259 const void *value = NULL; 4260 if (xattrs->count > 0 && xattrs->get_value == NULL) { 4261 return -EINVAL; 4262 } 4263 for (i = 0; i < xattrs->count; i++) { 4264 xattrs->get_value(xattrs->ctx, xattrs->names[i], &value, &value_len); 4265 if (value == NULL || value_len == 0) { 4266 return -EINVAL; 4267 } 4268 rc = _spdk_blob_set_xattr(blob, xattrs->names[i], value, value_len, internal); 4269 if (rc < 0) { 4270 return rc; 4271 } 4272 } 4273 return 0; 4274 } 4275 4276 static void 4277 _spdk_bs_create_blob(struct spdk_blob_store *bs, 4278 const struct spdk_blob_opts *opts, 4279 const struct spdk_blob_xattr_opts *internal_xattrs, 4280 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 4281 { 4282 struct spdk_blob *blob; 4283 uint32_t page_idx; 4284 struct spdk_bs_cpl cpl; 4285 struct spdk_blob_opts opts_default; 4286 struct spdk_blob_xattr_opts internal_xattrs_default; 4287 spdk_bs_sequence_t *seq; 4288 spdk_blob_id id; 4289 int rc; 4290 4291 assert(spdk_get_thread() == bs->md_thread); 4292 4293 page_idx = spdk_bit_array_find_first_clear(bs->used_md_pages, 0); 4294 if (page_idx == UINT32_MAX) { 4295 cb_fn(cb_arg, 0, -ENOMEM); 4296 return; 4297 } 4298 spdk_bit_array_set(bs->used_blobids, page_idx); 4299 spdk_bit_array_set(bs->used_md_pages, page_idx); 4300 4301 id = _spdk_bs_page_to_blobid(page_idx); 4302 4303 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Creating blob with id %lu at page %u\n", id, page_idx); 4304 4305 blob = _spdk_blob_alloc(bs, id); 4306 if (!blob) { 4307 cb_fn(cb_arg, 0, -ENOMEM); 4308 return; 4309 } 4310 4311 if (!opts) { 4312 spdk_blob_opts_init(&opts_default); 4313 opts = &opts_default; 4314 } 4315 if (!internal_xattrs) { 4316 _spdk_blob_xattrs_init(&internal_xattrs_default); 4317 internal_xattrs = &internal_xattrs_default; 4318 } 4319 4320 rc = _spdk_blob_set_xattrs(blob, &opts->xattrs, false); 4321 if (rc < 0) { 4322 _spdk_blob_free(blob); 4323 cb_fn(cb_arg, 0, rc); 4324 return; 4325 } 4326 4327 rc = _spdk_blob_set_xattrs(blob, internal_xattrs, true); 4328 if (rc < 0) { 4329 _spdk_blob_free(blob); 4330 cb_fn(cb_arg, 0, rc); 4331 return; 4332 } 4333 4334 if (opts->thin_provision) { 4335 _spdk_blob_set_thin_provision(blob); 4336 } 4337 4338 rc = _spdk_blob_resize(blob, opts->num_clusters); 4339 if (rc < 0) { 4340 _spdk_blob_free(blob); 4341 cb_fn(cb_arg, 0, rc); 4342 return; 4343 } 4344 cpl.type = SPDK_BS_CPL_TYPE_BLOBID; 4345 cpl.u.blobid.cb_fn = cb_fn; 4346 cpl.u.blobid.cb_arg = cb_arg; 4347 cpl.u.blobid.blobid = blob->id; 4348 4349 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 4350 if (!seq) { 4351 _spdk_blob_free(blob); 4352 cb_fn(cb_arg, 0, -ENOMEM); 4353 return; 4354 } 4355 4356 _spdk_blob_persist(seq, blob, _spdk_bs_create_blob_cpl, blob); 4357 } 4358 4359 void spdk_bs_create_blob(struct spdk_blob_store *bs, 4360 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 4361 { 4362 _spdk_bs_create_blob(bs, NULL, NULL, cb_fn, cb_arg); 4363 } 4364 4365 void spdk_bs_create_blob_ext(struct spdk_blob_store *bs, const struct spdk_blob_opts *opts, 4366 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 4367 { 4368 _spdk_bs_create_blob(bs, opts, NULL, cb_fn, cb_arg); 4369 } 4370 4371 /* END spdk_bs_create_blob */ 4372 4373 /* START blob_cleanup */ 4374 4375 struct spdk_clone_snapshot_ctx { 4376 struct spdk_bs_cpl cpl; 4377 int bserrno; 4378 bool frozen; 4379 4380 struct spdk_io_channel *channel; 4381 4382 /* Current cluster for inflate operation */ 4383 uint64_t cluster; 4384 4385 /* For inflation force allocation of all unallocated clusters and remove 4386 * thin-provisioning. Otherwise only decouple parent and keep clone thin. */ 4387 bool allocate_all; 4388 4389 struct { 4390 spdk_blob_id id; 4391 struct spdk_blob *blob; 4392 } original; 4393 struct { 4394 spdk_blob_id id; 4395 struct spdk_blob *blob; 4396 } new; 4397 4398 /* xattrs specified for snapshot/clones only. They have no impact on 4399 * the original blobs xattrs. */ 4400 const struct spdk_blob_xattr_opts *xattrs; 4401 }; 4402 4403 static void 4404 _spdk_bs_clone_snapshot_cleanup_finish(void *cb_arg, int bserrno) 4405 { 4406 struct spdk_clone_snapshot_ctx *ctx = cb_arg; 4407 struct spdk_bs_cpl *cpl = &ctx->cpl; 4408 4409 if (bserrno != 0) { 4410 if (ctx->bserrno != 0) { 4411 SPDK_ERRLOG("Cleanup error %d\n", bserrno); 4412 } else { 4413 ctx->bserrno = bserrno; 4414 } 4415 } 4416 4417 switch (cpl->type) { 4418 case SPDK_BS_CPL_TYPE_BLOBID: 4419 cpl->u.blobid.cb_fn(cpl->u.blobid.cb_arg, cpl->u.blobid.blobid, ctx->bserrno); 4420 break; 4421 case SPDK_BS_CPL_TYPE_BLOB_BASIC: 4422 cpl->u.blob_basic.cb_fn(cpl->u.blob_basic.cb_arg, ctx->bserrno); 4423 break; 4424 default: 4425 SPDK_UNREACHABLE(); 4426 break; 4427 } 4428 4429 free(ctx); 4430 } 4431 4432 static void 4433 _spdk_bs_snapshot_unfreeze_cpl(void *cb_arg, int bserrno) 4434 { 4435 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4436 struct spdk_blob *origblob = ctx->original.blob; 4437 4438 if (bserrno != 0) { 4439 if (ctx->bserrno != 0) { 4440 SPDK_ERRLOG("Unfreeze error %d\n", bserrno); 4441 } else { 4442 ctx->bserrno = bserrno; 4443 } 4444 } 4445 4446 ctx->original.id = origblob->id; 4447 origblob->locked_operation_in_progress = false; 4448 4449 spdk_blob_close(origblob, _spdk_bs_clone_snapshot_cleanup_finish, ctx); 4450 } 4451 4452 static void 4453 _spdk_bs_clone_snapshot_origblob_cleanup(void *cb_arg, int bserrno) 4454 { 4455 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4456 struct spdk_blob *origblob = ctx->original.blob; 4457 4458 if (bserrno != 0) { 4459 if (ctx->bserrno != 0) { 4460 SPDK_ERRLOG("Cleanup error %d\n", bserrno); 4461 } else { 4462 ctx->bserrno = bserrno; 4463 } 4464 } 4465 4466 if (ctx->frozen) { 4467 /* Unfreeze any outstanding I/O */ 4468 _spdk_blob_unfreeze_io(origblob, _spdk_bs_snapshot_unfreeze_cpl, ctx); 4469 } else { 4470 _spdk_bs_snapshot_unfreeze_cpl(ctx, 0); 4471 } 4472 4473 } 4474 4475 static void 4476 _spdk_bs_clone_snapshot_newblob_cleanup(void *cb_arg, int bserrno) 4477 { 4478 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4479 struct spdk_blob *newblob = ctx->new.blob; 4480 4481 if (bserrno != 0) { 4482 if (ctx->bserrno != 0) { 4483 SPDK_ERRLOG("Cleanup error %d\n", bserrno); 4484 } else { 4485 ctx->bserrno = bserrno; 4486 } 4487 } 4488 4489 ctx->new.id = newblob->id; 4490 spdk_blob_close(newblob, _spdk_bs_clone_snapshot_origblob_cleanup, ctx); 4491 } 4492 4493 /* END blob_cleanup */ 4494 4495 /* START spdk_bs_create_snapshot */ 4496 4497 static void 4498 _spdk_bs_snapshot_swap_cluster_maps(struct spdk_blob *blob1, struct spdk_blob *blob2) 4499 { 4500 uint64_t *cluster_temp; 4501 4502 cluster_temp = blob1->active.clusters; 4503 blob1->active.clusters = blob2->active.clusters; 4504 blob2->active.clusters = cluster_temp; 4505 } 4506 4507 static void 4508 _spdk_bs_snapshot_origblob_sync_cpl(void *cb_arg, int bserrno) 4509 { 4510 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4511 struct spdk_blob *origblob = ctx->original.blob; 4512 struct spdk_blob *newblob = ctx->new.blob; 4513 4514 if (bserrno != 0) { 4515 _spdk_bs_snapshot_swap_cluster_maps(newblob, origblob); 4516 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4517 return; 4518 } 4519 4520 /* Remove metadata descriptor SNAPSHOT_IN_PROGRESS */ 4521 bserrno = _spdk_blob_remove_xattr(newblob, SNAPSHOT_IN_PROGRESS, true); 4522 if (bserrno != 0) { 4523 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4524 return; 4525 } 4526 4527 _spdk_bs_blob_list_add(ctx->original.blob); 4528 4529 spdk_blob_set_read_only(newblob); 4530 4531 /* sync snapshot metadata */ 4532 spdk_blob_sync_md(newblob, _spdk_bs_clone_snapshot_origblob_cleanup, cb_arg); 4533 } 4534 4535 static void 4536 _spdk_bs_snapshot_newblob_sync_cpl(void *cb_arg, int bserrno) 4537 { 4538 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4539 struct spdk_blob *origblob = ctx->original.blob; 4540 struct spdk_blob *newblob = ctx->new.blob; 4541 4542 if (bserrno != 0) { 4543 /* return cluster map back to original */ 4544 _spdk_bs_snapshot_swap_cluster_maps(newblob, origblob); 4545 _spdk_bs_clone_snapshot_newblob_cleanup(ctx, bserrno); 4546 return; 4547 } 4548 4549 /* Set internal xattr for snapshot id */ 4550 bserrno = _spdk_blob_set_xattr(origblob, BLOB_SNAPSHOT, &newblob->id, sizeof(spdk_blob_id), true); 4551 if (bserrno != 0) { 4552 /* return cluster map back to original */ 4553 _spdk_bs_snapshot_swap_cluster_maps(newblob, origblob); 4554 _spdk_bs_clone_snapshot_newblob_cleanup(ctx, bserrno); 4555 return; 4556 } 4557 4558 _spdk_bs_blob_list_remove(origblob); 4559 origblob->parent_id = newblob->id; 4560 4561 /* Create new back_bs_dev for snapshot */ 4562 origblob->back_bs_dev = spdk_bs_create_blob_bs_dev(newblob); 4563 if (origblob->back_bs_dev == NULL) { 4564 /* return cluster map back to original */ 4565 _spdk_bs_snapshot_swap_cluster_maps(newblob, origblob); 4566 _spdk_bs_clone_snapshot_newblob_cleanup(ctx, -EINVAL); 4567 return; 4568 } 4569 4570 /* set clone blob as thin provisioned */ 4571 _spdk_blob_set_thin_provision(origblob); 4572 4573 _spdk_bs_blob_list_add(newblob); 4574 4575 /* sync clone metadata */ 4576 spdk_blob_sync_md(origblob, _spdk_bs_snapshot_origblob_sync_cpl, ctx); 4577 } 4578 4579 static void 4580 _spdk_bs_snapshot_freeze_cpl(void *cb_arg, int rc) 4581 { 4582 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4583 struct spdk_blob *origblob = ctx->original.blob; 4584 struct spdk_blob *newblob = ctx->new.blob; 4585 int bserrno; 4586 4587 if (rc != 0) { 4588 _spdk_bs_clone_snapshot_newblob_cleanup(ctx, rc); 4589 return; 4590 } 4591 4592 ctx->frozen = true; 4593 4594 /* set new back_bs_dev for snapshot */ 4595 newblob->back_bs_dev = origblob->back_bs_dev; 4596 /* Set invalid flags from origblob */ 4597 newblob->invalid_flags = origblob->invalid_flags; 4598 4599 /* inherit parent from original blob if set */ 4600 newblob->parent_id = origblob->parent_id; 4601 if (origblob->parent_id != SPDK_BLOBID_INVALID) { 4602 /* Set internal xattr for snapshot id */ 4603 bserrno = _spdk_blob_set_xattr(newblob, BLOB_SNAPSHOT, 4604 &origblob->parent_id, sizeof(spdk_blob_id), true); 4605 if (bserrno != 0) { 4606 _spdk_bs_clone_snapshot_newblob_cleanup(ctx, bserrno); 4607 return; 4608 } 4609 } 4610 4611 /* swap cluster maps */ 4612 _spdk_bs_snapshot_swap_cluster_maps(newblob, origblob); 4613 4614 /* sync snapshot metadata */ 4615 spdk_blob_sync_md(newblob, _spdk_bs_snapshot_newblob_sync_cpl, ctx); 4616 } 4617 4618 static void 4619 _spdk_bs_snapshot_newblob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 4620 { 4621 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4622 struct spdk_blob *origblob = ctx->original.blob; 4623 struct spdk_blob *newblob = _blob; 4624 4625 if (bserrno != 0) { 4626 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4627 return; 4628 } 4629 4630 ctx->new.blob = newblob; 4631 assert(spdk_blob_is_thin_provisioned(newblob)); 4632 assert(spdk_mem_all_zero(newblob->active.clusters, 4633 newblob->active.num_clusters * sizeof(*newblob->active.clusters))); 4634 4635 _spdk_blob_freeze_io(origblob, _spdk_bs_snapshot_freeze_cpl, ctx); 4636 } 4637 4638 static void 4639 _spdk_bs_snapshot_newblob_create_cpl(void *cb_arg, spdk_blob_id blobid, int bserrno) 4640 { 4641 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4642 struct spdk_blob *origblob = ctx->original.blob; 4643 4644 if (bserrno != 0) { 4645 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4646 return; 4647 } 4648 4649 ctx->new.id = blobid; 4650 ctx->cpl.u.blobid.blobid = blobid; 4651 4652 spdk_bs_open_blob(origblob->bs, ctx->new.id, _spdk_bs_snapshot_newblob_open_cpl, ctx); 4653 } 4654 4655 4656 static void 4657 _spdk_bs_xattr_snapshot(void *arg, const char *name, 4658 const void **value, size_t *value_len) 4659 { 4660 assert(strncmp(name, SNAPSHOT_IN_PROGRESS, sizeof(SNAPSHOT_IN_PROGRESS)) == 0); 4661 4662 struct spdk_blob *blob = (struct spdk_blob *)arg; 4663 *value = &blob->id; 4664 *value_len = sizeof(blob->id); 4665 } 4666 4667 static void 4668 _spdk_bs_snapshot_origblob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 4669 { 4670 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4671 struct spdk_blob_opts opts; 4672 struct spdk_blob_xattr_opts internal_xattrs; 4673 char *xattrs_names[] = { SNAPSHOT_IN_PROGRESS }; 4674 4675 if (bserrno != 0) { 4676 _spdk_bs_clone_snapshot_cleanup_finish(ctx, bserrno); 4677 return; 4678 } 4679 4680 ctx->original.blob = _blob; 4681 4682 if (_blob->data_ro || _blob->md_ro) { 4683 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Cannot create snapshot from read only blob with id %lu\n", 4684 _blob->id); 4685 ctx->bserrno = -EINVAL; 4686 spdk_blob_close(_blob, _spdk_bs_clone_snapshot_cleanup_finish, ctx); 4687 return; 4688 } 4689 4690 if (_blob->locked_operation_in_progress) { 4691 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Cannot create snapshot - another operation in progress\n"); 4692 ctx->bserrno = -EBUSY; 4693 spdk_blob_close(_blob, _spdk_bs_clone_snapshot_cleanup_finish, ctx); 4694 return; 4695 } 4696 4697 _blob->locked_operation_in_progress = true; 4698 4699 spdk_blob_opts_init(&opts); 4700 _spdk_blob_xattrs_init(&internal_xattrs); 4701 4702 /* Change the size of new blob to the same as in original blob, 4703 * but do not allocate clusters */ 4704 opts.thin_provision = true; 4705 opts.num_clusters = spdk_blob_get_num_clusters(_blob); 4706 4707 /* If there are any xattrs specified for snapshot, set them now */ 4708 if (ctx->xattrs) { 4709 memcpy(&opts.xattrs, ctx->xattrs, sizeof(*ctx->xattrs)); 4710 } 4711 /* Set internal xattr SNAPSHOT_IN_PROGRESS */ 4712 internal_xattrs.count = 1; 4713 internal_xattrs.ctx = _blob; 4714 internal_xattrs.names = xattrs_names; 4715 internal_xattrs.get_value = _spdk_bs_xattr_snapshot; 4716 4717 _spdk_bs_create_blob(_blob->bs, &opts, &internal_xattrs, 4718 _spdk_bs_snapshot_newblob_create_cpl, ctx); 4719 } 4720 4721 void spdk_bs_create_snapshot(struct spdk_blob_store *bs, spdk_blob_id blobid, 4722 const struct spdk_blob_xattr_opts *snapshot_xattrs, 4723 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 4724 { 4725 struct spdk_clone_snapshot_ctx *ctx = calloc(1, sizeof(*ctx)); 4726 4727 if (!ctx) { 4728 cb_fn(cb_arg, SPDK_BLOBID_INVALID, -ENOMEM); 4729 return; 4730 } 4731 ctx->cpl.type = SPDK_BS_CPL_TYPE_BLOBID; 4732 ctx->cpl.u.blobid.cb_fn = cb_fn; 4733 ctx->cpl.u.blobid.cb_arg = cb_arg; 4734 ctx->cpl.u.blobid.blobid = SPDK_BLOBID_INVALID; 4735 ctx->bserrno = 0; 4736 ctx->frozen = false; 4737 ctx->original.id = blobid; 4738 ctx->xattrs = snapshot_xattrs; 4739 4740 spdk_bs_open_blob(bs, ctx->original.id, _spdk_bs_snapshot_origblob_open_cpl, ctx); 4741 } 4742 /* END spdk_bs_create_snapshot */ 4743 4744 /* START spdk_bs_create_clone */ 4745 4746 static void 4747 _spdk_bs_xattr_clone(void *arg, const char *name, 4748 const void **value, size_t *value_len) 4749 { 4750 assert(strncmp(name, BLOB_SNAPSHOT, sizeof(BLOB_SNAPSHOT)) == 0); 4751 4752 struct spdk_blob *blob = (struct spdk_blob *)arg; 4753 *value = &blob->id; 4754 *value_len = sizeof(blob->id); 4755 } 4756 4757 static void 4758 _spdk_bs_clone_newblob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 4759 { 4760 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4761 struct spdk_blob *clone = _blob; 4762 4763 ctx->new.blob = clone; 4764 _spdk_bs_blob_list_add(clone); 4765 4766 spdk_blob_close(clone, _spdk_bs_clone_snapshot_origblob_cleanup, ctx); 4767 } 4768 4769 static void 4770 _spdk_bs_clone_newblob_create_cpl(void *cb_arg, spdk_blob_id blobid, int bserrno) 4771 { 4772 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4773 4774 ctx->cpl.u.blobid.blobid = blobid; 4775 spdk_bs_open_blob(ctx->original.blob->bs, blobid, _spdk_bs_clone_newblob_open_cpl, ctx); 4776 } 4777 4778 static void 4779 _spdk_bs_clone_origblob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 4780 { 4781 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4782 struct spdk_blob_opts opts; 4783 struct spdk_blob_xattr_opts internal_xattrs; 4784 char *xattr_names[] = { BLOB_SNAPSHOT }; 4785 4786 if (bserrno != 0) { 4787 _spdk_bs_clone_snapshot_cleanup_finish(ctx, bserrno); 4788 return; 4789 } 4790 4791 ctx->original.blob = _blob; 4792 4793 if (!_blob->data_ro || !_blob->md_ro) { 4794 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Clone not from read-only blob\n"); 4795 ctx->bserrno = -EINVAL; 4796 spdk_blob_close(_blob, _spdk_bs_clone_snapshot_cleanup_finish, ctx); 4797 return; 4798 } 4799 4800 if (_blob->locked_operation_in_progress) { 4801 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Cannot create clone - another operation in progress\n"); 4802 ctx->bserrno = -EBUSY; 4803 spdk_blob_close(_blob, _spdk_bs_clone_snapshot_cleanup_finish, ctx); 4804 return; 4805 } 4806 4807 _blob->locked_operation_in_progress = true; 4808 4809 spdk_blob_opts_init(&opts); 4810 _spdk_blob_xattrs_init(&internal_xattrs); 4811 4812 opts.thin_provision = true; 4813 opts.num_clusters = spdk_blob_get_num_clusters(_blob); 4814 if (ctx->xattrs) { 4815 memcpy(&opts.xattrs, ctx->xattrs, sizeof(*ctx->xattrs)); 4816 } 4817 4818 /* Set internal xattr BLOB_SNAPSHOT */ 4819 internal_xattrs.count = 1; 4820 internal_xattrs.ctx = _blob; 4821 internal_xattrs.names = xattr_names; 4822 internal_xattrs.get_value = _spdk_bs_xattr_clone; 4823 4824 _spdk_bs_create_blob(_blob->bs, &opts, &internal_xattrs, 4825 _spdk_bs_clone_newblob_create_cpl, ctx); 4826 } 4827 4828 void spdk_bs_create_clone(struct spdk_blob_store *bs, spdk_blob_id blobid, 4829 const struct spdk_blob_xattr_opts *clone_xattrs, 4830 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 4831 { 4832 struct spdk_clone_snapshot_ctx *ctx = calloc(1, sizeof(*ctx)); 4833 4834 if (!ctx) { 4835 cb_fn(cb_arg, SPDK_BLOBID_INVALID, -ENOMEM); 4836 return; 4837 } 4838 4839 ctx->cpl.type = SPDK_BS_CPL_TYPE_BLOBID; 4840 ctx->cpl.u.blobid.cb_fn = cb_fn; 4841 ctx->cpl.u.blobid.cb_arg = cb_arg; 4842 ctx->cpl.u.blobid.blobid = SPDK_BLOBID_INVALID; 4843 ctx->bserrno = 0; 4844 ctx->xattrs = clone_xattrs; 4845 ctx->original.id = blobid; 4846 4847 spdk_bs_open_blob(bs, ctx->original.id, _spdk_bs_clone_origblob_open_cpl, ctx); 4848 } 4849 4850 /* END spdk_bs_create_clone */ 4851 4852 /* START spdk_bs_inflate_blob */ 4853 4854 static void 4855 _spdk_bs_inflate_blob_set_parent_cpl(void *cb_arg, struct spdk_blob *_parent, int bserrno) 4856 { 4857 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4858 struct spdk_blob *_blob = ctx->original.blob; 4859 4860 if (bserrno != 0) { 4861 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4862 return; 4863 } 4864 4865 assert(_parent != NULL); 4866 4867 _spdk_bs_blob_list_remove(_blob); 4868 _blob->parent_id = _parent->id; 4869 _spdk_blob_set_xattr(_blob, BLOB_SNAPSHOT, &_blob->parent_id, 4870 sizeof(spdk_blob_id), true); 4871 4872 _blob->back_bs_dev->destroy(_blob->back_bs_dev); 4873 _blob->back_bs_dev = spdk_bs_create_blob_bs_dev(_parent); 4874 _spdk_bs_blob_list_add(_blob); 4875 4876 spdk_blob_sync_md(_blob, _spdk_bs_clone_snapshot_origblob_cleanup, ctx); 4877 } 4878 4879 static void 4880 _spdk_bs_inflate_blob_done(void *cb_arg, int bserrno) 4881 { 4882 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4883 struct spdk_blob *_blob = ctx->original.blob; 4884 struct spdk_blob *_parent; 4885 4886 if (bserrno != 0) { 4887 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4888 return; 4889 } 4890 4891 if (ctx->allocate_all) { 4892 /* remove thin provisioning */ 4893 _spdk_bs_blob_list_remove(_blob); 4894 _spdk_blob_remove_xattr(_blob, BLOB_SNAPSHOT, true); 4895 _blob->invalid_flags = _blob->invalid_flags & ~SPDK_BLOB_THIN_PROV; 4896 _blob->back_bs_dev->destroy(_blob->back_bs_dev); 4897 _blob->back_bs_dev = NULL; 4898 _blob->parent_id = SPDK_BLOBID_INVALID; 4899 } else { 4900 _parent = ((struct spdk_blob_bs_dev *)(_blob->back_bs_dev))->blob; 4901 if (_parent->parent_id != SPDK_BLOBID_INVALID) { 4902 /* We must change the parent of the inflated blob */ 4903 spdk_bs_open_blob(_blob->bs, _parent->parent_id, 4904 _spdk_bs_inflate_blob_set_parent_cpl, ctx); 4905 return; 4906 } 4907 4908 _spdk_bs_blob_list_remove(_blob); 4909 _spdk_blob_remove_xattr(_blob, BLOB_SNAPSHOT, true); 4910 _blob->parent_id = SPDK_BLOBID_INVALID; 4911 _blob->back_bs_dev->destroy(_blob->back_bs_dev); 4912 _blob->back_bs_dev = spdk_bs_create_zeroes_dev(); 4913 } 4914 4915 _blob->state = SPDK_BLOB_STATE_DIRTY; 4916 spdk_blob_sync_md(_blob, _spdk_bs_clone_snapshot_origblob_cleanup, ctx); 4917 } 4918 4919 /* Check if cluster needs allocation */ 4920 static inline bool 4921 _spdk_bs_cluster_needs_allocation(struct spdk_blob *blob, uint64_t cluster, bool allocate_all) 4922 { 4923 struct spdk_blob_bs_dev *b; 4924 4925 assert(blob != NULL); 4926 4927 if (blob->active.clusters[cluster] != 0) { 4928 /* Cluster is already allocated */ 4929 return false; 4930 } 4931 4932 if (blob->parent_id == SPDK_BLOBID_INVALID) { 4933 /* Blob have no parent blob */ 4934 return allocate_all; 4935 } 4936 4937 b = (struct spdk_blob_bs_dev *)blob->back_bs_dev; 4938 return (allocate_all || b->blob->active.clusters[cluster] != 0); 4939 } 4940 4941 static void 4942 _spdk_bs_inflate_blob_touch_next(void *cb_arg, int bserrno) 4943 { 4944 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4945 struct spdk_blob *_blob = ctx->original.blob; 4946 uint64_t offset; 4947 4948 if (bserrno != 0) { 4949 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4950 return; 4951 } 4952 4953 for (; ctx->cluster < _blob->active.num_clusters; ctx->cluster++) { 4954 if (_spdk_bs_cluster_needs_allocation(_blob, ctx->cluster, ctx->allocate_all)) { 4955 break; 4956 } 4957 } 4958 4959 if (ctx->cluster < _blob->active.num_clusters) { 4960 offset = _spdk_bs_cluster_to_lba(_blob->bs, ctx->cluster); 4961 4962 /* We may safely increment a cluster before write */ 4963 ctx->cluster++; 4964 4965 /* Use zero length write to touch a cluster */ 4966 spdk_blob_io_write(_blob, ctx->channel, NULL, offset, 0, 4967 _spdk_bs_inflate_blob_touch_next, ctx); 4968 } else { 4969 _spdk_bs_inflate_blob_done(cb_arg, bserrno); 4970 } 4971 } 4972 4973 static void 4974 _spdk_bs_inflate_blob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 4975 { 4976 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4977 uint64_t lfc; /* lowest free cluster */ 4978 uint64_t i; 4979 4980 if (bserrno != 0) { 4981 _spdk_bs_clone_snapshot_cleanup_finish(ctx, bserrno); 4982 return; 4983 } 4984 4985 ctx->original.blob = _blob; 4986 4987 if (_blob->locked_operation_in_progress) { 4988 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Cannot inflate blob - another operation in progress\n"); 4989 ctx->bserrno = -EBUSY; 4990 spdk_blob_close(_blob, _spdk_bs_clone_snapshot_cleanup_finish, ctx); 4991 return; 4992 } 4993 4994 _blob->locked_operation_in_progress = true; 4995 4996 if (!ctx->allocate_all && _blob->parent_id == SPDK_BLOBID_INVALID) { 4997 /* This blob have no parent, so we cannot decouple it. */ 4998 SPDK_ERRLOG("Cannot decouple parent of blob with no parent.\n"); 4999 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, -EINVAL); 5000 return; 5001 } 5002 5003 if (spdk_blob_is_thin_provisioned(_blob) == false) { 5004 /* This is not thin provisioned blob. No need to inflate. */ 5005 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, 0); 5006 return; 5007 } 5008 5009 /* Do two passes - one to verify that we can obtain enough clusters 5010 * and another to actually claim them. 5011 */ 5012 lfc = 0; 5013 for (i = 0; i < _blob->active.num_clusters; i++) { 5014 if (_spdk_bs_cluster_needs_allocation(_blob, i, ctx->allocate_all)) { 5015 lfc = spdk_bit_array_find_first_clear(_blob->bs->used_clusters, lfc); 5016 if (lfc == UINT32_MAX) { 5017 /* No more free clusters. Cannot satisfy the request */ 5018 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, -ENOSPC); 5019 return; 5020 } 5021 lfc++; 5022 } 5023 } 5024 5025 ctx->cluster = 0; 5026 _spdk_bs_inflate_blob_touch_next(ctx, 0); 5027 } 5028 5029 static void 5030 _spdk_bs_inflate_blob(struct spdk_blob_store *bs, struct spdk_io_channel *channel, 5031 spdk_blob_id blobid, bool allocate_all, spdk_blob_op_complete cb_fn, void *cb_arg) 5032 { 5033 struct spdk_clone_snapshot_ctx *ctx = calloc(1, sizeof(*ctx)); 5034 5035 if (!ctx) { 5036 cb_fn(cb_arg, -ENOMEM); 5037 return; 5038 } 5039 ctx->cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 5040 ctx->cpl.u.bs_basic.cb_fn = cb_fn; 5041 ctx->cpl.u.bs_basic.cb_arg = cb_arg; 5042 ctx->bserrno = 0; 5043 ctx->original.id = blobid; 5044 ctx->channel = channel; 5045 ctx->allocate_all = allocate_all; 5046 5047 spdk_bs_open_blob(bs, ctx->original.id, _spdk_bs_inflate_blob_open_cpl, ctx); 5048 } 5049 5050 void 5051 spdk_bs_inflate_blob(struct spdk_blob_store *bs, struct spdk_io_channel *channel, 5052 spdk_blob_id blobid, spdk_blob_op_complete cb_fn, void *cb_arg) 5053 { 5054 _spdk_bs_inflate_blob(bs, channel, blobid, true, cb_fn, cb_arg); 5055 } 5056 5057 void 5058 spdk_bs_blob_decouple_parent(struct spdk_blob_store *bs, struct spdk_io_channel *channel, 5059 spdk_blob_id blobid, spdk_blob_op_complete cb_fn, void *cb_arg) 5060 { 5061 _spdk_bs_inflate_blob(bs, channel, blobid, false, cb_fn, cb_arg); 5062 } 5063 /* END spdk_bs_inflate_blob */ 5064 5065 /* START spdk_blob_resize */ 5066 struct spdk_bs_resize_ctx { 5067 spdk_blob_op_complete cb_fn; 5068 void *cb_arg; 5069 struct spdk_blob *blob; 5070 uint64_t sz; 5071 int rc; 5072 }; 5073 5074 static void 5075 _spdk_bs_resize_unfreeze_cpl(void *cb_arg, int rc) 5076 { 5077 struct spdk_bs_resize_ctx *ctx = (struct spdk_bs_resize_ctx *)cb_arg; 5078 5079 if (rc != 0) { 5080 SPDK_ERRLOG("Unfreeze failed, rc=%d\n", rc); 5081 } 5082 5083 if (ctx->rc != 0) { 5084 SPDK_ERRLOG("Unfreeze failed, ctx->rc=%d\n", ctx->rc); 5085 rc = ctx->rc; 5086 } 5087 5088 ctx->blob->locked_operation_in_progress = false; 5089 5090 ctx->cb_fn(ctx->cb_arg, rc); 5091 free(ctx); 5092 } 5093 5094 static void 5095 _spdk_bs_resize_freeze_cpl(void *cb_arg, int rc) 5096 { 5097 struct spdk_bs_resize_ctx *ctx = (struct spdk_bs_resize_ctx *)cb_arg; 5098 5099 if (rc != 0) { 5100 ctx->blob->locked_operation_in_progress = false; 5101 ctx->cb_fn(ctx->cb_arg, rc); 5102 free(ctx); 5103 return; 5104 } 5105 5106 ctx->rc = _spdk_blob_resize(ctx->blob, ctx->sz); 5107 5108 _spdk_blob_unfreeze_io(ctx->blob, _spdk_bs_resize_unfreeze_cpl, ctx); 5109 } 5110 5111 void 5112 spdk_blob_resize(struct spdk_blob *blob, uint64_t sz, spdk_blob_op_complete cb_fn, void *cb_arg) 5113 { 5114 struct spdk_bs_resize_ctx *ctx; 5115 5116 _spdk_blob_verify_md_op(blob); 5117 5118 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Resizing blob %lu to %lu clusters\n", blob->id, sz); 5119 5120 if (blob->md_ro) { 5121 cb_fn(cb_arg, -EPERM); 5122 return; 5123 } 5124 5125 if (sz == blob->active.num_clusters) { 5126 cb_fn(cb_arg, 0); 5127 return; 5128 } 5129 5130 if (blob->locked_operation_in_progress) { 5131 cb_fn(cb_arg, -EBUSY); 5132 return; 5133 } 5134 5135 ctx = calloc(1, sizeof(*ctx)); 5136 if (!ctx) { 5137 cb_fn(cb_arg, -ENOMEM); 5138 return; 5139 } 5140 5141 blob->locked_operation_in_progress = true; 5142 ctx->cb_fn = cb_fn; 5143 ctx->cb_arg = cb_arg; 5144 ctx->blob = blob; 5145 ctx->sz = sz; 5146 _spdk_blob_freeze_io(blob, _spdk_bs_resize_freeze_cpl, ctx); 5147 } 5148 5149 /* END spdk_blob_resize */ 5150 5151 5152 /* START spdk_bs_delete_blob */ 5153 5154 static void 5155 _spdk_bs_delete_close_cpl(void *cb_arg, int bserrno) 5156 { 5157 spdk_bs_sequence_t *seq = cb_arg; 5158 5159 spdk_bs_sequence_finish(seq, bserrno); 5160 } 5161 5162 static void 5163 _spdk_bs_delete_persist_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 5164 { 5165 struct spdk_blob *blob = cb_arg; 5166 5167 if (bserrno != 0) { 5168 /* 5169 * We already removed this blob from the blobstore tailq, so 5170 * we need to free it here since this is the last reference 5171 * to it. 5172 */ 5173 _spdk_blob_free(blob); 5174 _spdk_bs_delete_close_cpl(seq, bserrno); 5175 return; 5176 } 5177 5178 /* 5179 * This will immediately decrement the ref_count and call 5180 * the completion routine since the metadata state is clean. 5181 * By calling spdk_blob_close, we reduce the number of call 5182 * points into code that touches the blob->open_ref count 5183 * and the blobstore's blob list. 5184 */ 5185 spdk_blob_close(blob, _spdk_bs_delete_close_cpl, seq); 5186 } 5187 5188 struct delete_snapshot_ctx { 5189 struct spdk_blob_list *parent_snapshot_entry; 5190 struct spdk_blob *snapshot; 5191 bool snapshot_md_ro; 5192 struct spdk_blob *clone; 5193 bool clone_md_ro; 5194 spdk_blob_op_with_handle_complete cb_fn; 5195 void *cb_arg; 5196 int bserrno; 5197 }; 5198 5199 static void 5200 _spdk_delete_blob_cleanup_finish(void *cb_arg, int bserrno) 5201 { 5202 struct delete_snapshot_ctx *ctx = cb_arg; 5203 5204 if (bserrno != 0) { 5205 SPDK_ERRLOG("Snapshot cleanup error %d\n", bserrno); 5206 } 5207 5208 assert(ctx != NULL); 5209 5210 if (bserrno != 0 && ctx->bserrno == 0) { 5211 ctx->bserrno = bserrno; 5212 } 5213 5214 ctx->cb_fn(ctx->cb_arg, ctx->snapshot, ctx->bserrno); 5215 free(ctx); 5216 } 5217 5218 static void 5219 _spdk_delete_snapshot_cleanup_snapshot(void *cb_arg, int bserrno) 5220 { 5221 struct delete_snapshot_ctx *ctx = cb_arg; 5222 5223 if (bserrno != 0) { 5224 ctx->bserrno = bserrno; 5225 SPDK_ERRLOG("Clone cleanup error %d\n", bserrno); 5226 } 5227 5228 /* open_ref == 1 menas that only deletion context has opened this snapshot 5229 * open_ref == 2 menas that clone has opened this snapshot as well, 5230 * so we have to add it back to the blobs list */ 5231 if (ctx->snapshot->open_ref == 2) { 5232 TAILQ_INSERT_HEAD(&ctx->snapshot->bs->blobs, ctx->snapshot, link); 5233 } 5234 5235 ctx->snapshot->locked_operation_in_progress = false; 5236 ctx->snapshot->md_ro = ctx->snapshot_md_ro; 5237 5238 spdk_blob_close(ctx->snapshot, _spdk_delete_blob_cleanup_finish, ctx); 5239 } 5240 5241 static void 5242 _spdk_delete_snapshot_cleanup_clone(void *cb_arg, int bserrno) 5243 { 5244 struct delete_snapshot_ctx *ctx = cb_arg; 5245 5246 ctx->clone->locked_operation_in_progress = false; 5247 ctx->clone->md_ro = ctx->clone_md_ro; 5248 5249 spdk_blob_close(ctx->clone, _spdk_delete_snapshot_cleanup_snapshot, ctx); 5250 } 5251 5252 static void 5253 _spdk_delete_snapshot_unfreeze_cpl(void *cb_arg, int bserrno) 5254 { 5255 struct delete_snapshot_ctx *ctx = cb_arg; 5256 5257 if (bserrno) { 5258 ctx->bserrno = bserrno; 5259 _spdk_delete_snapshot_cleanup_clone(ctx, 0); 5260 return; 5261 } 5262 5263 ctx->clone->locked_operation_in_progress = false; 5264 spdk_blob_close(ctx->clone, _spdk_delete_blob_cleanup_finish, ctx); 5265 } 5266 5267 static void 5268 _spdk_delete_snapshot_sync_snapshot_cpl(void *cb_arg, int bserrno) 5269 { 5270 struct delete_snapshot_ctx *ctx = cb_arg; 5271 struct spdk_blob_list *parent_snapshot_entry = NULL; 5272 struct spdk_blob_list *snapshot_entry = NULL; 5273 struct spdk_blob_list *clone_entry = NULL; 5274 struct spdk_blob_list *snapshot_clone_entry = NULL; 5275 5276 if (bserrno) { 5277 SPDK_ERRLOG("Failed to sync MD on blob\n"); 5278 ctx->bserrno = bserrno; 5279 _spdk_delete_snapshot_cleanup_clone(ctx, 0); 5280 return; 5281 } 5282 5283 /* Get snapshot entry for the snapshot we want to remove */ 5284 snapshot_entry = _spdk_bs_get_snapshot_entry(ctx->snapshot->bs, ctx->snapshot->id); 5285 5286 assert(snapshot_entry != NULL); 5287 5288 /* Remove clone entry in this snapshot (at this point there can be only one clone) */ 5289 clone_entry = TAILQ_FIRST(&snapshot_entry->clones); 5290 assert(clone_entry != NULL); 5291 TAILQ_REMOVE(&snapshot_entry->clones, clone_entry, link); 5292 snapshot_entry->clone_count--; 5293 assert(TAILQ_EMPTY(&snapshot_entry->clones)); 5294 5295 if (ctx->snapshot->parent_id != SPDK_BLOBID_INVALID) { 5296 /* This snapshot is at the same time a clone of another snapshot - we need to 5297 * update parent snapshot (remove current clone, add new one inherited from 5298 * the snapshot that is being removed) */ 5299 5300 /* Get snapshot entry for parent snapshot and clone entry within that snapshot for 5301 * snapshot that we are removing */ 5302 _spdk_blob_get_snapshot_and_clone_entries(ctx->snapshot, &parent_snapshot_entry, 5303 &snapshot_clone_entry); 5304 5305 /* Switch clone entry in parent snapshot */ 5306 TAILQ_INSERT_TAIL(&parent_snapshot_entry->clones, clone_entry, link); 5307 TAILQ_REMOVE(&parent_snapshot_entry->clones, snapshot_clone_entry, link); 5308 free(snapshot_clone_entry); 5309 } else { 5310 /* No parent snapshot - just remove clone entry */ 5311 free(clone_entry); 5312 } 5313 5314 /* Restore md_ro flags */ 5315 ctx->clone->md_ro = ctx->clone_md_ro; 5316 ctx->snapshot->md_ro = ctx->snapshot_md_ro; 5317 5318 _spdk_blob_unfreeze_io(ctx->clone, _spdk_delete_snapshot_unfreeze_cpl, ctx); 5319 } 5320 5321 static void 5322 _spdk_delete_snapshot_sync_clone_cpl(void *cb_arg, int bserrno) 5323 { 5324 struct delete_snapshot_ctx *ctx = cb_arg; 5325 uint64_t i; 5326 5327 ctx->snapshot->md_ro = false; 5328 5329 if (bserrno) { 5330 SPDK_ERRLOG("Failed to sync MD on clone\n"); 5331 ctx->bserrno = bserrno; 5332 5333 /* Restore snapshot to previous state */ 5334 bserrno = _spdk_blob_remove_xattr(ctx->snapshot, SNAPSHOT_PENDING_REMOVAL, true); 5335 if (bserrno != 0) { 5336 _spdk_delete_snapshot_cleanup_clone(ctx, bserrno); 5337 return; 5338 } 5339 5340 spdk_blob_sync_md(ctx->snapshot, _spdk_delete_snapshot_cleanup_clone, ctx); 5341 return; 5342 } 5343 5344 /* Clear cluster map entries for snapshot */ 5345 for (i = 0; i < ctx->snapshot->active.num_clusters && i < ctx->clone->active.num_clusters; i++) { 5346 if (ctx->clone->active.clusters[i] == ctx->snapshot->active.clusters[i]) { 5347 ctx->snapshot->active.clusters[i] = 0; 5348 } 5349 } 5350 5351 ctx->snapshot->state = SPDK_BLOB_STATE_DIRTY; 5352 5353 if (ctx->parent_snapshot_entry != NULL) { 5354 ctx->snapshot->back_bs_dev = NULL; 5355 } 5356 5357 spdk_blob_sync_md(ctx->snapshot, _spdk_delete_snapshot_sync_snapshot_cpl, ctx); 5358 } 5359 5360 static void 5361 _spdk_delete_snapshot_sync_snapshot_xattr_cpl(void *cb_arg, int bserrno) 5362 { 5363 struct delete_snapshot_ctx *ctx = cb_arg; 5364 uint64_t i; 5365 5366 /* Temporarily override md_ro flag for clone for MD modification */ 5367 ctx->clone_md_ro = ctx->clone->md_ro; 5368 ctx->clone->md_ro = false; 5369 5370 if (bserrno) { 5371 SPDK_ERRLOG("Failed to sync MD with xattr on blob\n"); 5372 ctx->bserrno = bserrno; 5373 _spdk_delete_snapshot_cleanup_clone(ctx, 0); 5374 return; 5375 } 5376 5377 /* Copy snapshot map to clone map (only unallocated clusters in clone) */ 5378 for (i = 0; i < ctx->snapshot->active.num_clusters && i < ctx->clone->active.num_clusters; i++) { 5379 if (ctx->clone->active.clusters[i] == 0) { 5380 ctx->clone->active.clusters[i] = ctx->snapshot->active.clusters[i]; 5381 } 5382 } 5383 5384 /* Delete old backing bs_dev from clone (related to snapshot that will be removed) */ 5385 ctx->clone->back_bs_dev->destroy(ctx->clone->back_bs_dev); 5386 5387 /* Set/remove snapshot xattr and switch parent ID and backing bs_dev on clone... */ 5388 if (ctx->parent_snapshot_entry != NULL) { 5389 /* ...to parent snapshot */ 5390 ctx->clone->parent_id = ctx->parent_snapshot_entry->id; 5391 ctx->clone->back_bs_dev = ctx->snapshot->back_bs_dev; 5392 _spdk_blob_set_xattr(ctx->clone, BLOB_SNAPSHOT, &ctx->parent_snapshot_entry->id, 5393 sizeof(spdk_blob_id), 5394 true); 5395 } else { 5396 /* ...to blobid invalid and zeroes dev */ 5397 ctx->clone->parent_id = SPDK_BLOBID_INVALID; 5398 ctx->clone->back_bs_dev = spdk_bs_create_zeroes_dev(); 5399 _spdk_blob_remove_xattr(ctx->clone, BLOB_SNAPSHOT, true); 5400 } 5401 5402 spdk_blob_sync_md(ctx->clone, _spdk_delete_snapshot_sync_clone_cpl, ctx); 5403 } 5404 5405 static void 5406 _spdk_delete_snapshot_freeze_io_cb(void *cb_arg, int bserrno) 5407 { 5408 struct delete_snapshot_ctx *ctx = cb_arg; 5409 5410 if (bserrno) { 5411 SPDK_ERRLOG("Failed to freeze I/O on clone\n"); 5412 ctx->bserrno = bserrno; 5413 _spdk_delete_snapshot_cleanup_clone(ctx, 0); 5414 return; 5415 } 5416 5417 /* Temporarily override md_ro flag for snapshot for MD modification */ 5418 ctx->snapshot_md_ro = ctx->snapshot->md_ro; 5419 ctx->snapshot->md_ro = false; 5420 5421 /* Mark blob as pending for removal for power failure safety, use clone id for recovery */ 5422 ctx->bserrno = _spdk_blob_set_xattr(ctx->snapshot, SNAPSHOT_PENDING_REMOVAL, &ctx->clone->id, 5423 sizeof(spdk_blob_id), true); 5424 if (ctx->bserrno != 0) { 5425 _spdk_delete_snapshot_cleanup_clone(ctx, 0); 5426 return; 5427 } 5428 5429 spdk_blob_sync_md(ctx->snapshot, _spdk_delete_snapshot_sync_snapshot_xattr_cpl, ctx); 5430 } 5431 5432 static void 5433 _spdk_delete_snapshot_open_clone_cb(void *cb_arg, struct spdk_blob *clone, int bserrno) 5434 { 5435 struct delete_snapshot_ctx *ctx = cb_arg; 5436 5437 if (bserrno) { 5438 SPDK_ERRLOG("Failed to open clone\n"); 5439 ctx->bserrno = bserrno; 5440 _spdk_delete_snapshot_cleanup_snapshot(ctx, 0); 5441 return; 5442 } 5443 5444 ctx->clone = clone; 5445 5446 if (clone->locked_operation_in_progress) { 5447 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Cannot remove blob - another operation in progress on its clone\n"); 5448 ctx->bserrno = -EBUSY; 5449 spdk_blob_close(ctx->clone, _spdk_delete_snapshot_cleanup_snapshot, ctx); 5450 return; 5451 } 5452 5453 clone->locked_operation_in_progress = true; 5454 5455 _spdk_blob_freeze_io(clone, _spdk_delete_snapshot_freeze_io_cb, ctx); 5456 } 5457 5458 static void 5459 _spdk_update_clone_on_snapshot_deletion(struct spdk_blob *snapshot, struct delete_snapshot_ctx *ctx) 5460 { 5461 struct spdk_blob_list *snapshot_entry = NULL; 5462 struct spdk_blob_list *clone_entry = NULL; 5463 struct spdk_blob_list *snapshot_clone_entry = NULL; 5464 5465 /* Get snapshot entry for the snapshot we want to remove */ 5466 snapshot_entry = _spdk_bs_get_snapshot_entry(snapshot->bs, snapshot->id); 5467 5468 assert(snapshot_entry != NULL); 5469 5470 /* Get clone of the snapshot (at this point there can be only one clone) */ 5471 clone_entry = TAILQ_FIRST(&snapshot_entry->clones); 5472 assert(snapshot_entry->clone_count == 1); 5473 assert(clone_entry != NULL); 5474 5475 /* Get snapshot entry for parent snapshot and clone entry within that snapshot for 5476 * snapshot that we are removing */ 5477 _spdk_blob_get_snapshot_and_clone_entries(snapshot, &ctx->parent_snapshot_entry, 5478 &snapshot_clone_entry); 5479 5480 spdk_bs_open_blob(snapshot->bs, clone_entry->id, _spdk_delete_snapshot_open_clone_cb, ctx); 5481 } 5482 5483 static void 5484 _spdk_bs_delete_blob_finish(void *cb_arg, struct spdk_blob *blob, int bserrno) 5485 { 5486 spdk_bs_sequence_t *seq = cb_arg; 5487 struct spdk_blob_list *snapshot_entry = NULL; 5488 uint32_t page_num; 5489 5490 if (bserrno) { 5491 SPDK_ERRLOG("Failed to remove blob\n"); 5492 spdk_bs_sequence_finish(seq, bserrno); 5493 return; 5494 } 5495 5496 /* Remove snapshot from the list */ 5497 snapshot_entry = _spdk_bs_get_snapshot_entry(blob->bs, blob->id); 5498 if (snapshot_entry != NULL) { 5499 TAILQ_REMOVE(&blob->bs->snapshots, snapshot_entry, link); 5500 free(snapshot_entry); 5501 } 5502 5503 page_num = _spdk_bs_blobid_to_page(blob->id); 5504 spdk_bit_array_clear(blob->bs->used_blobids, page_num); 5505 blob->state = SPDK_BLOB_STATE_DIRTY; 5506 blob->active.num_pages = 0; 5507 _spdk_blob_resize(blob, 0); 5508 5509 _spdk_blob_persist(seq, blob, _spdk_bs_delete_persist_cpl, blob); 5510 } 5511 5512 static int 5513 _spdk_bs_is_blob_deletable(struct spdk_blob *blob, bool *update_clone) 5514 { 5515 struct spdk_blob_list *snapshot_entry = NULL; 5516 struct spdk_blob_list *clone_entry = NULL; 5517 struct spdk_blob *clone = NULL; 5518 bool has_one_clone = false; 5519 5520 /* Check if this is a snapshot with clones */ 5521 snapshot_entry = _spdk_bs_get_snapshot_entry(blob->bs, blob->id); 5522 if (snapshot_entry != NULL) { 5523 if (snapshot_entry->clone_count > 1) { 5524 SPDK_ERRLOG("Cannot remove snapshot with more than one clone\n"); 5525 return -EBUSY; 5526 } else if (snapshot_entry->clone_count == 1) { 5527 has_one_clone = true; 5528 } 5529 } 5530 5531 /* Check if someone has this blob open (besides this delete context): 5532 * - open_ref = 1 - only this context opened blob, so it is ok to remove it 5533 * - open_ref <= 2 && has_one_clone = true - clone is holding snapshot 5534 * and that is ok, because we will update it accordingly */ 5535 if (blob->open_ref <= 2 && has_one_clone) { 5536 clone_entry = TAILQ_FIRST(&snapshot_entry->clones); 5537 assert(clone_entry != NULL); 5538 clone = _spdk_blob_lookup(blob->bs, clone_entry->id); 5539 5540 if (blob->open_ref == 2 && clone == NULL) { 5541 /* Clone is closed and someone else opened this blob */ 5542 SPDK_ERRLOG("Cannot remove snapshot because it is open\n"); 5543 return -EBUSY; 5544 } 5545 5546 *update_clone = true; 5547 return 0; 5548 } 5549 5550 if (blob->open_ref > 1) { 5551 SPDK_ERRLOG("Cannot remove snapshot because it is open\n"); 5552 return -EBUSY; 5553 } 5554 5555 assert(has_one_clone == false); 5556 *update_clone = false; 5557 return 0; 5558 } 5559 5560 static void 5561 _spdk_bs_delete_enomem_close_cpl(void *cb_arg, int bserrno) 5562 { 5563 spdk_bs_sequence_t *seq = cb_arg; 5564 5565 spdk_bs_sequence_finish(seq, -ENOMEM); 5566 } 5567 5568 static void 5569 _spdk_bs_delete_open_cpl(void *cb_arg, struct spdk_blob *blob, int bserrno) 5570 { 5571 spdk_bs_sequence_t *seq = cb_arg; 5572 struct delete_snapshot_ctx *ctx; 5573 bool update_clone = false; 5574 5575 if (bserrno != 0) { 5576 spdk_bs_sequence_finish(seq, bserrno); 5577 return; 5578 } 5579 5580 _spdk_blob_verify_md_op(blob); 5581 5582 ctx = calloc(1, sizeof(*ctx)); 5583 if (ctx == NULL) { 5584 spdk_blob_close(blob, _spdk_bs_delete_enomem_close_cpl, seq); 5585 return; 5586 } 5587 5588 ctx->snapshot = blob; 5589 ctx->cb_fn = _spdk_bs_delete_blob_finish; 5590 ctx->cb_arg = seq; 5591 5592 /* Check if blob can be removed and if it is a snapshot with clone on top of it */ 5593 ctx->bserrno = _spdk_bs_is_blob_deletable(blob, &update_clone); 5594 if (ctx->bserrno) { 5595 spdk_blob_close(blob, _spdk_delete_blob_cleanup_finish, ctx); 5596 return; 5597 } 5598 5599 if (blob->locked_operation_in_progress) { 5600 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Cannot remove blob - another operation in progress\n"); 5601 ctx->bserrno = -EBUSY; 5602 spdk_blob_close(blob, _spdk_delete_blob_cleanup_finish, ctx); 5603 return; 5604 } 5605 5606 blob->locked_operation_in_progress = true; 5607 5608 /* 5609 * Remove the blob from the blob_store list now, to ensure it does not 5610 * get returned after this point by _spdk_blob_lookup(). 5611 */ 5612 TAILQ_REMOVE(&blob->bs->blobs, blob, link); 5613 5614 if (update_clone) { 5615 /* This blob is a snapshot with active clone - update clone first */ 5616 _spdk_update_clone_on_snapshot_deletion(blob, ctx); 5617 } else { 5618 /* This blob does not have any clones - just remove it */ 5619 _spdk_bs_blob_list_remove(blob); 5620 _spdk_bs_delete_blob_finish(seq, blob, 0); 5621 free(ctx); 5622 } 5623 } 5624 5625 void 5626 spdk_bs_delete_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 5627 spdk_blob_op_complete cb_fn, void *cb_arg) 5628 { 5629 struct spdk_bs_cpl cpl; 5630 spdk_bs_sequence_t *seq; 5631 5632 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Deleting blob %lu\n", blobid); 5633 5634 assert(spdk_get_thread() == bs->md_thread); 5635 5636 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 5637 cpl.u.blob_basic.cb_fn = cb_fn; 5638 cpl.u.blob_basic.cb_arg = cb_arg; 5639 5640 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 5641 if (!seq) { 5642 cb_fn(cb_arg, -ENOMEM); 5643 return; 5644 } 5645 5646 spdk_bs_open_blob(bs, blobid, _spdk_bs_delete_open_cpl, seq); 5647 } 5648 5649 /* END spdk_bs_delete_blob */ 5650 5651 /* START spdk_bs_open_blob */ 5652 5653 static void 5654 _spdk_bs_open_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 5655 { 5656 struct spdk_blob *blob = cb_arg; 5657 5658 if (bserrno != 0) { 5659 _spdk_blob_free(blob); 5660 seq->cpl.u.blob_handle.blob = NULL; 5661 spdk_bs_sequence_finish(seq, bserrno); 5662 return; 5663 } 5664 5665 blob->open_ref++; 5666 5667 TAILQ_INSERT_HEAD(&blob->bs->blobs, blob, link); 5668 5669 spdk_bs_sequence_finish(seq, bserrno); 5670 } 5671 5672 static void _spdk_bs_open_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 5673 struct spdk_blob_open_opts *opts, spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 5674 { 5675 struct spdk_blob *blob; 5676 struct spdk_bs_cpl cpl; 5677 struct spdk_blob_open_opts opts_default; 5678 spdk_bs_sequence_t *seq; 5679 uint32_t page_num; 5680 5681 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Opening blob %lu\n", blobid); 5682 assert(spdk_get_thread() == bs->md_thread); 5683 5684 page_num = _spdk_bs_blobid_to_page(blobid); 5685 if (spdk_bit_array_get(bs->used_blobids, page_num) == false) { 5686 /* Invalid blobid */ 5687 cb_fn(cb_arg, NULL, -ENOENT); 5688 return; 5689 } 5690 5691 blob = _spdk_blob_lookup(bs, blobid); 5692 if (blob) { 5693 blob->open_ref++; 5694 cb_fn(cb_arg, blob, 0); 5695 return; 5696 } 5697 5698 blob = _spdk_blob_alloc(bs, blobid); 5699 if (!blob) { 5700 cb_fn(cb_arg, NULL, -ENOMEM); 5701 return; 5702 } 5703 5704 if (!opts) { 5705 spdk_blob_open_opts_init(&opts_default); 5706 opts = &opts_default; 5707 } 5708 5709 blob->clear_method = opts->clear_method; 5710 5711 cpl.type = SPDK_BS_CPL_TYPE_BLOB_HANDLE; 5712 cpl.u.blob_handle.cb_fn = cb_fn; 5713 cpl.u.blob_handle.cb_arg = cb_arg; 5714 cpl.u.blob_handle.blob = blob; 5715 5716 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 5717 if (!seq) { 5718 _spdk_blob_free(blob); 5719 cb_fn(cb_arg, NULL, -ENOMEM); 5720 return; 5721 } 5722 5723 _spdk_blob_load(seq, blob, _spdk_bs_open_blob_cpl, blob); 5724 } 5725 5726 void spdk_bs_open_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 5727 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 5728 { 5729 _spdk_bs_open_blob(bs, blobid, NULL, cb_fn, cb_arg); 5730 } 5731 5732 void spdk_bs_open_blob_ext(struct spdk_blob_store *bs, spdk_blob_id blobid, 5733 struct spdk_blob_open_opts *opts, spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 5734 { 5735 _spdk_bs_open_blob(bs, blobid, opts, cb_fn, cb_arg); 5736 } 5737 5738 /* END spdk_bs_open_blob */ 5739 5740 /* START spdk_blob_set_read_only */ 5741 int spdk_blob_set_read_only(struct spdk_blob *blob) 5742 { 5743 _spdk_blob_verify_md_op(blob); 5744 5745 blob->data_ro_flags |= SPDK_BLOB_READ_ONLY; 5746 5747 blob->state = SPDK_BLOB_STATE_DIRTY; 5748 return 0; 5749 } 5750 /* END spdk_blob_set_read_only */ 5751 5752 /* START spdk_blob_sync_md */ 5753 5754 static void 5755 _spdk_blob_sync_md_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 5756 { 5757 struct spdk_blob *blob = cb_arg; 5758 5759 if (bserrno == 0 && (blob->data_ro_flags & SPDK_BLOB_READ_ONLY)) { 5760 blob->data_ro = true; 5761 blob->md_ro = true; 5762 } 5763 5764 spdk_bs_sequence_finish(seq, bserrno); 5765 } 5766 5767 static void 5768 _spdk_blob_sync_md(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 5769 { 5770 struct spdk_bs_cpl cpl; 5771 spdk_bs_sequence_t *seq; 5772 5773 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 5774 cpl.u.blob_basic.cb_fn = cb_fn; 5775 cpl.u.blob_basic.cb_arg = cb_arg; 5776 5777 seq = spdk_bs_sequence_start(blob->bs->md_channel, &cpl); 5778 if (!seq) { 5779 cb_fn(cb_arg, -ENOMEM); 5780 return; 5781 } 5782 5783 _spdk_blob_persist(seq, blob, _spdk_blob_sync_md_cpl, blob); 5784 } 5785 5786 void 5787 spdk_blob_sync_md(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 5788 { 5789 _spdk_blob_verify_md_op(blob); 5790 5791 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Syncing blob %lu\n", blob->id); 5792 5793 if (blob->md_ro) { 5794 assert(blob->state == SPDK_BLOB_STATE_CLEAN); 5795 cb_fn(cb_arg, 0); 5796 return; 5797 } 5798 5799 _spdk_blob_sync_md(blob, cb_fn, cb_arg); 5800 } 5801 5802 /* END spdk_blob_sync_md */ 5803 5804 struct spdk_blob_insert_cluster_ctx { 5805 struct spdk_thread *thread; 5806 struct spdk_blob *blob; 5807 uint32_t cluster_num; /* cluster index in blob */ 5808 uint32_t cluster; /* cluster on disk */ 5809 int rc; 5810 spdk_blob_op_complete cb_fn; 5811 void *cb_arg; 5812 }; 5813 5814 static void 5815 _spdk_blob_insert_cluster_msg_cpl(void *arg) 5816 { 5817 struct spdk_blob_insert_cluster_ctx *ctx = arg; 5818 5819 ctx->cb_fn(ctx->cb_arg, ctx->rc); 5820 free(ctx); 5821 } 5822 5823 static void 5824 _spdk_blob_insert_cluster_msg_cb(void *arg, int bserrno) 5825 { 5826 struct spdk_blob_insert_cluster_ctx *ctx = arg; 5827 5828 ctx->rc = bserrno; 5829 spdk_thread_send_msg(ctx->thread, _spdk_blob_insert_cluster_msg_cpl, ctx); 5830 } 5831 5832 static void 5833 _spdk_blob_insert_cluster_msg(void *arg) 5834 { 5835 struct spdk_blob_insert_cluster_ctx *ctx = arg; 5836 5837 ctx->rc = _spdk_blob_insert_cluster(ctx->blob, ctx->cluster_num, ctx->cluster); 5838 if (ctx->rc != 0) { 5839 spdk_thread_send_msg(ctx->thread, _spdk_blob_insert_cluster_msg_cpl, ctx); 5840 return; 5841 } 5842 5843 ctx->blob->state = SPDK_BLOB_STATE_DIRTY; 5844 _spdk_blob_sync_md(ctx->blob, _spdk_blob_insert_cluster_msg_cb, ctx); 5845 } 5846 5847 static void 5848 _spdk_blob_insert_cluster_on_md_thread(struct spdk_blob *blob, uint32_t cluster_num, 5849 uint64_t cluster, spdk_blob_op_complete cb_fn, void *cb_arg) 5850 { 5851 struct spdk_blob_insert_cluster_ctx *ctx; 5852 5853 ctx = calloc(1, sizeof(*ctx)); 5854 if (ctx == NULL) { 5855 cb_fn(cb_arg, -ENOMEM); 5856 return; 5857 } 5858 5859 ctx->thread = spdk_get_thread(); 5860 ctx->blob = blob; 5861 ctx->cluster_num = cluster_num; 5862 ctx->cluster = cluster; 5863 ctx->cb_fn = cb_fn; 5864 ctx->cb_arg = cb_arg; 5865 5866 spdk_thread_send_msg(blob->bs->md_thread, _spdk_blob_insert_cluster_msg, ctx); 5867 } 5868 5869 /* START spdk_blob_close */ 5870 5871 static void 5872 _spdk_blob_close_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 5873 { 5874 struct spdk_blob *blob = cb_arg; 5875 5876 if (bserrno == 0) { 5877 blob->open_ref--; 5878 if (blob->open_ref == 0) { 5879 /* 5880 * Blobs with active.num_pages == 0 are deleted blobs. 5881 * these blobs are removed from the blob_store list 5882 * when the deletion process starts - so don't try to 5883 * remove them again. 5884 */ 5885 if (blob->active.num_pages > 0) { 5886 TAILQ_REMOVE(&blob->bs->blobs, blob, link); 5887 } 5888 _spdk_blob_free(blob); 5889 } 5890 } 5891 5892 spdk_bs_sequence_finish(seq, bserrno); 5893 } 5894 5895 void spdk_blob_close(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 5896 { 5897 struct spdk_bs_cpl cpl; 5898 spdk_bs_sequence_t *seq; 5899 5900 _spdk_blob_verify_md_op(blob); 5901 5902 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Closing blob %lu\n", blob->id); 5903 5904 if (blob->open_ref == 0) { 5905 cb_fn(cb_arg, -EBADF); 5906 return; 5907 } 5908 5909 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 5910 cpl.u.blob_basic.cb_fn = cb_fn; 5911 cpl.u.blob_basic.cb_arg = cb_arg; 5912 5913 seq = spdk_bs_sequence_start(blob->bs->md_channel, &cpl); 5914 if (!seq) { 5915 cb_fn(cb_arg, -ENOMEM); 5916 return; 5917 } 5918 5919 /* Sync metadata */ 5920 _spdk_blob_persist(seq, blob, _spdk_blob_close_cpl, blob); 5921 } 5922 5923 /* END spdk_blob_close */ 5924 5925 struct spdk_io_channel *spdk_bs_alloc_io_channel(struct spdk_blob_store *bs) 5926 { 5927 return spdk_get_io_channel(bs); 5928 } 5929 5930 void spdk_bs_free_io_channel(struct spdk_io_channel *channel) 5931 { 5932 spdk_put_io_channel(channel); 5933 } 5934 5935 void spdk_blob_io_unmap(struct spdk_blob *blob, struct spdk_io_channel *channel, 5936 uint64_t offset, uint64_t length, spdk_blob_op_complete cb_fn, void *cb_arg) 5937 { 5938 _spdk_blob_request_submit_op(blob, channel, NULL, offset, length, cb_fn, cb_arg, 5939 SPDK_BLOB_UNMAP); 5940 } 5941 5942 void spdk_blob_io_write_zeroes(struct spdk_blob *blob, struct spdk_io_channel *channel, 5943 uint64_t offset, uint64_t length, spdk_blob_op_complete cb_fn, void *cb_arg) 5944 { 5945 _spdk_blob_request_submit_op(blob, channel, NULL, offset, length, cb_fn, cb_arg, 5946 SPDK_BLOB_WRITE_ZEROES); 5947 } 5948 5949 void spdk_blob_io_write(struct spdk_blob *blob, struct spdk_io_channel *channel, 5950 void *payload, uint64_t offset, uint64_t length, 5951 spdk_blob_op_complete cb_fn, void *cb_arg) 5952 { 5953 _spdk_blob_request_submit_op(blob, channel, payload, offset, length, cb_fn, cb_arg, 5954 SPDK_BLOB_WRITE); 5955 } 5956 5957 void spdk_blob_io_read(struct spdk_blob *blob, struct spdk_io_channel *channel, 5958 void *payload, uint64_t offset, uint64_t length, 5959 spdk_blob_op_complete cb_fn, void *cb_arg) 5960 { 5961 _spdk_blob_request_submit_op(blob, channel, payload, offset, length, cb_fn, cb_arg, 5962 SPDK_BLOB_READ); 5963 } 5964 5965 void spdk_blob_io_writev(struct spdk_blob *blob, struct spdk_io_channel *channel, 5966 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 5967 spdk_blob_op_complete cb_fn, void *cb_arg) 5968 { 5969 _spdk_blob_request_submit_rw_iov(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg, false); 5970 } 5971 5972 void spdk_blob_io_readv(struct spdk_blob *blob, struct spdk_io_channel *channel, 5973 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 5974 spdk_blob_op_complete cb_fn, void *cb_arg) 5975 { 5976 _spdk_blob_request_submit_rw_iov(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg, true); 5977 } 5978 5979 struct spdk_bs_iter_ctx { 5980 int64_t page_num; 5981 struct spdk_blob_store *bs; 5982 5983 spdk_blob_op_with_handle_complete cb_fn; 5984 void *cb_arg; 5985 }; 5986 5987 static void 5988 _spdk_bs_iter_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 5989 { 5990 struct spdk_bs_iter_ctx *ctx = cb_arg; 5991 struct spdk_blob_store *bs = ctx->bs; 5992 spdk_blob_id id; 5993 5994 if (bserrno == 0) { 5995 ctx->cb_fn(ctx->cb_arg, _blob, bserrno); 5996 free(ctx); 5997 return; 5998 } 5999 6000 ctx->page_num++; 6001 ctx->page_num = spdk_bit_array_find_first_set(bs->used_blobids, ctx->page_num); 6002 if (ctx->page_num >= spdk_bit_array_capacity(bs->used_blobids)) { 6003 ctx->cb_fn(ctx->cb_arg, NULL, -ENOENT); 6004 free(ctx); 6005 return; 6006 } 6007 6008 id = _spdk_bs_page_to_blobid(ctx->page_num); 6009 6010 spdk_bs_open_blob(bs, id, _spdk_bs_iter_cpl, ctx); 6011 } 6012 6013 void 6014 spdk_bs_iter_first(struct spdk_blob_store *bs, 6015 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 6016 { 6017 struct spdk_bs_iter_ctx *ctx; 6018 6019 ctx = calloc(1, sizeof(*ctx)); 6020 if (!ctx) { 6021 cb_fn(cb_arg, NULL, -ENOMEM); 6022 return; 6023 } 6024 6025 ctx->page_num = -1; 6026 ctx->bs = bs; 6027 ctx->cb_fn = cb_fn; 6028 ctx->cb_arg = cb_arg; 6029 6030 _spdk_bs_iter_cpl(ctx, NULL, -1); 6031 } 6032 6033 static void 6034 _spdk_bs_iter_close_cpl(void *cb_arg, int bserrno) 6035 { 6036 struct spdk_bs_iter_ctx *ctx = cb_arg; 6037 6038 _spdk_bs_iter_cpl(ctx, NULL, -1); 6039 } 6040 6041 void 6042 spdk_bs_iter_next(struct spdk_blob_store *bs, struct spdk_blob *blob, 6043 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 6044 { 6045 struct spdk_bs_iter_ctx *ctx; 6046 6047 assert(blob != NULL); 6048 6049 ctx = calloc(1, sizeof(*ctx)); 6050 if (!ctx) { 6051 cb_fn(cb_arg, NULL, -ENOMEM); 6052 return; 6053 } 6054 6055 ctx->page_num = _spdk_bs_blobid_to_page(blob->id); 6056 ctx->bs = bs; 6057 ctx->cb_fn = cb_fn; 6058 ctx->cb_arg = cb_arg; 6059 6060 /* Close the existing blob */ 6061 spdk_blob_close(blob, _spdk_bs_iter_close_cpl, ctx); 6062 } 6063 6064 static int 6065 _spdk_blob_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 6066 uint16_t value_len, bool internal) 6067 { 6068 struct spdk_xattr_tailq *xattrs; 6069 struct spdk_xattr *xattr; 6070 size_t desc_size; 6071 6072 _spdk_blob_verify_md_op(blob); 6073 6074 if (blob->md_ro) { 6075 return -EPERM; 6076 } 6077 6078 desc_size = sizeof(struct spdk_blob_md_descriptor_xattr) + strlen(name) + value_len; 6079 if (desc_size > SPDK_BS_MAX_DESC_SIZE) { 6080 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Xattr '%s' of size %ld does not fix into single page %ld\n", name, 6081 desc_size, SPDK_BS_MAX_DESC_SIZE); 6082 return -ENOMEM; 6083 } 6084 6085 if (internal) { 6086 xattrs = &blob->xattrs_internal; 6087 blob->invalid_flags |= SPDK_BLOB_INTERNAL_XATTR; 6088 } else { 6089 xattrs = &blob->xattrs; 6090 } 6091 6092 TAILQ_FOREACH(xattr, xattrs, link) { 6093 if (!strcmp(name, xattr->name)) { 6094 free(xattr->value); 6095 xattr->value_len = value_len; 6096 xattr->value = malloc(value_len); 6097 memcpy(xattr->value, value, value_len); 6098 6099 blob->state = SPDK_BLOB_STATE_DIRTY; 6100 6101 return 0; 6102 } 6103 } 6104 6105 xattr = calloc(1, sizeof(*xattr)); 6106 if (!xattr) { 6107 return -ENOMEM; 6108 } 6109 xattr->name = strdup(name); 6110 xattr->value_len = value_len; 6111 xattr->value = malloc(value_len); 6112 memcpy(xattr->value, value, value_len); 6113 TAILQ_INSERT_TAIL(xattrs, xattr, link); 6114 6115 blob->state = SPDK_BLOB_STATE_DIRTY; 6116 6117 return 0; 6118 } 6119 6120 int 6121 spdk_blob_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 6122 uint16_t value_len) 6123 { 6124 return _spdk_blob_set_xattr(blob, name, value, value_len, false); 6125 } 6126 6127 static int 6128 _spdk_blob_remove_xattr(struct spdk_blob *blob, const char *name, bool internal) 6129 { 6130 struct spdk_xattr_tailq *xattrs; 6131 struct spdk_xattr *xattr; 6132 6133 _spdk_blob_verify_md_op(blob); 6134 6135 if (blob->md_ro) { 6136 return -EPERM; 6137 } 6138 xattrs = internal ? &blob->xattrs_internal : &blob->xattrs; 6139 6140 TAILQ_FOREACH(xattr, xattrs, link) { 6141 if (!strcmp(name, xattr->name)) { 6142 TAILQ_REMOVE(xattrs, xattr, link); 6143 free(xattr->value); 6144 free(xattr->name); 6145 free(xattr); 6146 6147 if (internal && TAILQ_EMPTY(&blob->xattrs_internal)) { 6148 blob->invalid_flags &= ~SPDK_BLOB_INTERNAL_XATTR; 6149 } 6150 blob->state = SPDK_BLOB_STATE_DIRTY; 6151 6152 return 0; 6153 } 6154 } 6155 6156 return -ENOENT; 6157 } 6158 6159 int 6160 spdk_blob_remove_xattr(struct spdk_blob *blob, const char *name) 6161 { 6162 return _spdk_blob_remove_xattr(blob, name, false); 6163 } 6164 6165 static int 6166 _spdk_blob_get_xattr_value(struct spdk_blob *blob, const char *name, 6167 const void **value, size_t *value_len, bool internal) 6168 { 6169 struct spdk_xattr *xattr; 6170 struct spdk_xattr_tailq *xattrs; 6171 6172 xattrs = internal ? &blob->xattrs_internal : &blob->xattrs; 6173 6174 TAILQ_FOREACH(xattr, xattrs, link) { 6175 if (!strcmp(name, xattr->name)) { 6176 *value = xattr->value; 6177 *value_len = xattr->value_len; 6178 return 0; 6179 } 6180 } 6181 return -ENOENT; 6182 } 6183 6184 int 6185 spdk_blob_get_xattr_value(struct spdk_blob *blob, const char *name, 6186 const void **value, size_t *value_len) 6187 { 6188 _spdk_blob_verify_md_op(blob); 6189 6190 return _spdk_blob_get_xattr_value(blob, name, value, value_len, false); 6191 } 6192 6193 struct spdk_xattr_names { 6194 uint32_t count; 6195 const char *names[0]; 6196 }; 6197 6198 static int 6199 _spdk_blob_get_xattr_names(struct spdk_xattr_tailq *xattrs, struct spdk_xattr_names **names) 6200 { 6201 struct spdk_xattr *xattr; 6202 int count = 0; 6203 6204 TAILQ_FOREACH(xattr, xattrs, link) { 6205 count++; 6206 } 6207 6208 *names = calloc(1, sizeof(struct spdk_xattr_names) + count * sizeof(char *)); 6209 if (*names == NULL) { 6210 return -ENOMEM; 6211 } 6212 6213 TAILQ_FOREACH(xattr, xattrs, link) { 6214 (*names)->names[(*names)->count++] = xattr->name; 6215 } 6216 6217 return 0; 6218 } 6219 6220 int 6221 spdk_blob_get_xattr_names(struct spdk_blob *blob, struct spdk_xattr_names **names) 6222 { 6223 _spdk_blob_verify_md_op(blob); 6224 6225 return _spdk_blob_get_xattr_names(&blob->xattrs, names); 6226 } 6227 6228 uint32_t 6229 spdk_xattr_names_get_count(struct spdk_xattr_names *names) 6230 { 6231 assert(names != NULL); 6232 6233 return names->count; 6234 } 6235 6236 const char * 6237 spdk_xattr_names_get_name(struct spdk_xattr_names *names, uint32_t index) 6238 { 6239 if (index >= names->count) { 6240 return NULL; 6241 } 6242 6243 return names->names[index]; 6244 } 6245 6246 void 6247 spdk_xattr_names_free(struct spdk_xattr_names *names) 6248 { 6249 free(names); 6250 } 6251 6252 struct spdk_bs_type 6253 spdk_bs_get_bstype(struct spdk_blob_store *bs) 6254 { 6255 return bs->bstype; 6256 } 6257 6258 void 6259 spdk_bs_set_bstype(struct spdk_blob_store *bs, struct spdk_bs_type bstype) 6260 { 6261 memcpy(&bs->bstype, &bstype, sizeof(bstype)); 6262 } 6263 6264 bool 6265 spdk_blob_is_read_only(struct spdk_blob *blob) 6266 { 6267 assert(blob != NULL); 6268 return (blob->data_ro || blob->md_ro); 6269 } 6270 6271 bool 6272 spdk_blob_is_snapshot(struct spdk_blob *blob) 6273 { 6274 struct spdk_blob_list *snapshot_entry; 6275 6276 assert(blob != NULL); 6277 6278 snapshot_entry = _spdk_bs_get_snapshot_entry(blob->bs, blob->id); 6279 if (snapshot_entry == NULL) { 6280 return false; 6281 } 6282 6283 return true; 6284 } 6285 6286 bool 6287 spdk_blob_is_clone(struct spdk_blob *blob) 6288 { 6289 assert(blob != NULL); 6290 6291 if (blob->parent_id != SPDK_BLOBID_INVALID) { 6292 assert(spdk_blob_is_thin_provisioned(blob)); 6293 return true; 6294 } 6295 6296 return false; 6297 } 6298 6299 bool 6300 spdk_blob_is_thin_provisioned(struct spdk_blob *blob) 6301 { 6302 assert(blob != NULL); 6303 return !!(blob->invalid_flags & SPDK_BLOB_THIN_PROV); 6304 } 6305 6306 spdk_blob_id 6307 spdk_blob_get_parent_snapshot(struct spdk_blob_store *bs, spdk_blob_id blob_id) 6308 { 6309 struct spdk_blob_list *snapshot_entry = NULL; 6310 struct spdk_blob_list *clone_entry = NULL; 6311 6312 TAILQ_FOREACH(snapshot_entry, &bs->snapshots, link) { 6313 TAILQ_FOREACH(clone_entry, &snapshot_entry->clones, link) { 6314 if (clone_entry->id == blob_id) { 6315 return snapshot_entry->id; 6316 } 6317 } 6318 } 6319 6320 return SPDK_BLOBID_INVALID; 6321 } 6322 6323 int 6324 spdk_blob_get_clones(struct spdk_blob_store *bs, spdk_blob_id blobid, spdk_blob_id *ids, 6325 size_t *count) 6326 { 6327 struct spdk_blob_list *snapshot_entry, *clone_entry; 6328 size_t n; 6329 6330 snapshot_entry = _spdk_bs_get_snapshot_entry(bs, blobid); 6331 if (snapshot_entry == NULL) { 6332 *count = 0; 6333 return 0; 6334 } 6335 6336 if (ids == NULL || *count < snapshot_entry->clone_count) { 6337 *count = snapshot_entry->clone_count; 6338 return -ENOMEM; 6339 } 6340 *count = snapshot_entry->clone_count; 6341 6342 n = 0; 6343 TAILQ_FOREACH(clone_entry, &snapshot_entry->clones, link) { 6344 ids[n++] = clone_entry->id; 6345 } 6346 6347 return 0; 6348 } 6349 6350 SPDK_LOG_REGISTER_COMPONENT("blob", SPDK_LOG_BLOB) 6351