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 45 #include "spdk_internal/assert.h" 46 #include "spdk_internal/log.h" 47 48 #include "blobstore.h" 49 50 #define BLOB_CRC32C_INITIAL 0xffffffffUL 51 52 static int spdk_bs_register_md_thread(struct spdk_blob_store *bs); 53 static int spdk_bs_unregister_md_thread(struct spdk_blob_store *bs); 54 static void _spdk_blob_close_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno); 55 static void _spdk_blob_insert_cluster_on_md_thread(struct spdk_blob *blob, uint32_t cluster_num, 56 uint64_t cluster, spdk_blob_op_complete cb_fn, void *cb_arg); 57 58 static int _spdk_blob_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 59 uint16_t value_len, bool internal); 60 static int _spdk_blob_get_xattr_value(struct spdk_blob *blob, const char *name, 61 const void **value, size_t *value_len, bool internal); 62 static int _spdk_blob_remove_xattr(struct spdk_blob *blob, const char *name, bool internal); 63 64 static void 65 _spdk_blob_verify_md_op(struct spdk_blob *blob) 66 { 67 assert(blob != NULL); 68 assert(spdk_get_thread() == blob->bs->md_thread); 69 assert(blob->state != SPDK_BLOB_STATE_LOADING); 70 } 71 72 static struct spdk_blob_list * 73 _spdk_bs_get_snapshot_entry(struct spdk_blob_store *bs, spdk_blob_id blobid) 74 { 75 struct spdk_blob_list *snapshot_entry = NULL; 76 77 TAILQ_FOREACH(snapshot_entry, &bs->snapshots, link) { 78 if (snapshot_entry->id == blobid) { 79 break; 80 } 81 } 82 83 return snapshot_entry; 84 } 85 86 static void 87 _spdk_bs_claim_cluster(struct spdk_blob_store *bs, uint32_t cluster_num) 88 { 89 assert(cluster_num < spdk_bit_array_capacity(bs->used_clusters)); 90 assert(spdk_bit_array_get(bs->used_clusters, cluster_num) == false); 91 assert(bs->num_free_clusters > 0); 92 93 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Claiming cluster %u\n", cluster_num); 94 95 spdk_bit_array_set(bs->used_clusters, cluster_num); 96 bs->num_free_clusters--; 97 } 98 99 static int 100 _spdk_blob_insert_cluster(struct spdk_blob *blob, uint32_t cluster_num, uint64_t cluster) 101 { 102 uint64_t *cluster_lba = &blob->active.clusters[cluster_num]; 103 104 _spdk_blob_verify_md_op(blob); 105 106 if (*cluster_lba != 0) { 107 return -EEXIST; 108 } 109 110 *cluster_lba = _spdk_bs_cluster_to_lba(blob->bs, cluster); 111 return 0; 112 } 113 114 static int 115 _spdk_bs_allocate_cluster(struct spdk_blob *blob, uint32_t cluster_num, 116 uint64_t *lowest_free_cluster, bool update_map) 117 { 118 pthread_mutex_lock(&blob->bs->used_clusters_mutex); 119 *lowest_free_cluster = spdk_bit_array_find_first_clear(blob->bs->used_clusters, 120 *lowest_free_cluster); 121 if (*lowest_free_cluster == UINT32_MAX) { 122 /* No more free clusters. Cannot satisfy the request */ 123 pthread_mutex_unlock(&blob->bs->used_clusters_mutex); 124 return -ENOSPC; 125 } 126 127 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Claiming cluster %lu for blob %lu\n", *lowest_free_cluster, blob->id); 128 _spdk_bs_claim_cluster(blob->bs, *lowest_free_cluster); 129 pthread_mutex_unlock(&blob->bs->used_clusters_mutex); 130 131 if (update_map) { 132 _spdk_blob_insert_cluster(blob, cluster_num, *lowest_free_cluster); 133 } 134 135 return 0; 136 } 137 138 static void 139 _spdk_bs_release_cluster(struct spdk_blob_store *bs, uint32_t cluster_num) 140 { 141 assert(cluster_num < spdk_bit_array_capacity(bs->used_clusters)); 142 assert(spdk_bit_array_get(bs->used_clusters, cluster_num) == true); 143 assert(bs->num_free_clusters < bs->total_clusters); 144 145 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Releasing cluster %u\n", cluster_num); 146 147 pthread_mutex_lock(&bs->used_clusters_mutex); 148 spdk_bit_array_clear(bs->used_clusters, cluster_num); 149 bs->num_free_clusters++; 150 pthread_mutex_unlock(&bs->used_clusters_mutex); 151 } 152 153 static void 154 _spdk_blob_xattrs_init(struct spdk_blob_xattr_opts *xattrs) 155 { 156 xattrs->count = 0; 157 xattrs->names = NULL; 158 xattrs->ctx = NULL; 159 xattrs->get_value = NULL; 160 } 161 162 void 163 spdk_blob_opts_init(struct spdk_blob_opts *opts) 164 { 165 opts->num_clusters = 0; 166 opts->thin_provision = false; 167 _spdk_blob_xattrs_init(&opts->xattrs); 168 } 169 170 void 171 spdk_blob_open_opts_init(struct spdk_blob_open_opts *opts) 172 { 173 opts->clear_method = BLOB_CLEAR_WITH_UNMAP; 174 } 175 176 static struct spdk_blob * 177 _spdk_blob_alloc(struct spdk_blob_store *bs, spdk_blob_id id) 178 { 179 struct spdk_blob *blob; 180 181 blob = calloc(1, sizeof(*blob)); 182 if (!blob) { 183 return NULL; 184 } 185 186 blob->id = id; 187 blob->bs = bs; 188 189 blob->parent_id = SPDK_BLOBID_INVALID; 190 191 blob->state = SPDK_BLOB_STATE_DIRTY; 192 blob->active.num_pages = 1; 193 blob->active.pages = calloc(1, sizeof(*blob->active.pages)); 194 if (!blob->active.pages) { 195 free(blob); 196 return NULL; 197 } 198 199 blob->active.pages[0] = _spdk_bs_blobid_to_page(id); 200 201 TAILQ_INIT(&blob->xattrs); 202 TAILQ_INIT(&blob->xattrs_internal); 203 204 return blob; 205 } 206 207 static void 208 _spdk_xattrs_free(struct spdk_xattr_tailq *xattrs) 209 { 210 struct spdk_xattr *xattr, *xattr_tmp; 211 212 TAILQ_FOREACH_SAFE(xattr, xattrs, link, xattr_tmp) { 213 TAILQ_REMOVE(xattrs, xattr, link); 214 free(xattr->name); 215 free(xattr->value); 216 free(xattr); 217 } 218 } 219 220 static void 221 _spdk_blob_free(struct spdk_blob *blob) 222 { 223 assert(blob != NULL); 224 225 free(blob->active.clusters); 226 free(blob->clean.clusters); 227 free(blob->active.pages); 228 free(blob->clean.pages); 229 230 _spdk_xattrs_free(&blob->xattrs); 231 _spdk_xattrs_free(&blob->xattrs_internal); 232 233 if (blob->back_bs_dev) { 234 blob->back_bs_dev->destroy(blob->back_bs_dev); 235 } 236 237 free(blob); 238 } 239 240 struct freeze_io_ctx { 241 struct spdk_bs_cpl cpl; 242 struct spdk_blob *blob; 243 }; 244 245 static void 246 _spdk_blob_io_sync(struct spdk_io_channel_iter *i) 247 { 248 spdk_for_each_channel_continue(i, 0); 249 } 250 251 static void 252 _spdk_blob_execute_queued_io(struct spdk_io_channel_iter *i) 253 { 254 struct spdk_io_channel *_ch = spdk_io_channel_iter_get_channel(i); 255 struct spdk_bs_channel *ch = spdk_io_channel_get_ctx(_ch); 256 struct freeze_io_ctx *ctx = spdk_io_channel_iter_get_ctx(i); 257 struct spdk_bs_request_set *set; 258 struct spdk_bs_user_op_args *args; 259 spdk_bs_user_op_t *op, *tmp; 260 261 TAILQ_FOREACH_SAFE(op, &ch->queued_io, link, tmp) { 262 set = (struct spdk_bs_request_set *)op; 263 args = &set->u.user_op; 264 265 if (args->blob == ctx->blob) { 266 TAILQ_REMOVE(&ch->queued_io, op, link); 267 spdk_bs_user_op_execute(op); 268 } 269 } 270 271 spdk_for_each_channel_continue(i, 0); 272 } 273 274 static void 275 _spdk_blob_io_cpl(struct spdk_io_channel_iter *i, int status) 276 { 277 struct freeze_io_ctx *ctx = spdk_io_channel_iter_get_ctx(i); 278 279 ctx->cpl.u.blob_basic.cb_fn(ctx->cpl.u.blob_basic.cb_arg, 0); 280 281 free(ctx); 282 } 283 284 static void 285 _spdk_blob_freeze_io(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 286 { 287 struct freeze_io_ctx *ctx; 288 289 ctx = calloc(1, sizeof(*ctx)); 290 if (!ctx) { 291 cb_fn(cb_arg, -ENOMEM); 292 return; 293 } 294 295 ctx->cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 296 ctx->cpl.u.blob_basic.cb_fn = cb_fn; 297 ctx->cpl.u.blob_basic.cb_arg = cb_arg; 298 ctx->blob = blob; 299 300 /* Freeze I/O on blob */ 301 blob->frozen_refcnt++; 302 303 if (blob->frozen_refcnt == 1) { 304 spdk_for_each_channel(blob->bs, _spdk_blob_io_sync, ctx, _spdk_blob_io_cpl); 305 } else { 306 cb_fn(cb_arg, 0); 307 free(ctx); 308 } 309 } 310 311 static void 312 _spdk_blob_unfreeze_io(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 313 { 314 struct freeze_io_ctx *ctx; 315 316 ctx = calloc(1, sizeof(*ctx)); 317 if (!ctx) { 318 cb_fn(cb_arg, -ENOMEM); 319 return; 320 } 321 322 ctx->cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 323 ctx->cpl.u.blob_basic.cb_fn = cb_fn; 324 ctx->cpl.u.blob_basic.cb_arg = cb_arg; 325 ctx->blob = blob; 326 327 assert(blob->frozen_refcnt > 0); 328 329 blob->frozen_refcnt--; 330 331 if (blob->frozen_refcnt == 0) { 332 spdk_for_each_channel(blob->bs, _spdk_blob_execute_queued_io, ctx, _spdk_blob_io_cpl); 333 } else { 334 cb_fn(cb_arg, 0); 335 free(ctx); 336 } 337 } 338 339 static int 340 _spdk_blob_mark_clean(struct spdk_blob *blob) 341 { 342 uint64_t *clusters = NULL; 343 uint32_t *pages = NULL; 344 345 assert(blob != NULL); 346 347 if (blob->active.num_clusters) { 348 assert(blob->active.clusters); 349 clusters = calloc(blob->active.num_clusters, sizeof(*blob->active.clusters)); 350 if (!clusters) { 351 return -ENOMEM; 352 } 353 memcpy(clusters, blob->active.clusters, blob->active.num_clusters * sizeof(*clusters)); 354 } 355 356 if (blob->active.num_pages) { 357 assert(blob->active.pages); 358 pages = calloc(blob->active.num_pages, sizeof(*blob->active.pages)); 359 if (!pages) { 360 free(clusters); 361 return -ENOMEM; 362 } 363 memcpy(pages, blob->active.pages, blob->active.num_pages * sizeof(*pages)); 364 } 365 366 free(blob->clean.clusters); 367 free(blob->clean.pages); 368 369 blob->clean.num_clusters = blob->active.num_clusters; 370 blob->clean.clusters = blob->active.clusters; 371 blob->clean.num_pages = blob->active.num_pages; 372 blob->clean.pages = blob->active.pages; 373 374 blob->active.clusters = clusters; 375 blob->active.pages = pages; 376 377 /* If the metadata was dirtied again while the metadata was being written to disk, 378 * we do not want to revert the DIRTY state back to CLEAN here. 379 */ 380 if (blob->state == SPDK_BLOB_STATE_LOADING) { 381 blob->state = SPDK_BLOB_STATE_CLEAN; 382 } 383 384 return 0; 385 } 386 387 static int 388 _spdk_blob_deserialize_xattr(struct spdk_blob *blob, 389 struct spdk_blob_md_descriptor_xattr *desc_xattr, bool internal) 390 { 391 struct spdk_xattr *xattr; 392 393 if (desc_xattr->length != sizeof(desc_xattr->name_length) + 394 sizeof(desc_xattr->value_length) + 395 desc_xattr->name_length + desc_xattr->value_length) { 396 return -EINVAL; 397 } 398 399 xattr = calloc(1, sizeof(*xattr)); 400 if (xattr == NULL) { 401 return -ENOMEM; 402 } 403 404 xattr->name = malloc(desc_xattr->name_length + 1); 405 if (xattr->name == NULL) { 406 free(xattr); 407 return -ENOMEM; 408 } 409 memcpy(xattr->name, desc_xattr->name, desc_xattr->name_length); 410 xattr->name[desc_xattr->name_length] = '\0'; 411 412 xattr->value = malloc(desc_xattr->value_length); 413 if (xattr->value == NULL) { 414 free(xattr->name); 415 free(xattr); 416 return -ENOMEM; 417 } 418 xattr->value_len = desc_xattr->value_length; 419 memcpy(xattr->value, 420 (void *)((uintptr_t)desc_xattr->name + desc_xattr->name_length), 421 desc_xattr->value_length); 422 423 TAILQ_INSERT_TAIL(internal ? &blob->xattrs_internal : &blob->xattrs, xattr, link); 424 425 return 0; 426 } 427 428 429 static int 430 _spdk_blob_parse_page(const struct spdk_blob_md_page *page, struct spdk_blob *blob) 431 { 432 struct spdk_blob_md_descriptor *desc; 433 size_t cur_desc = 0; 434 void *tmp; 435 436 desc = (struct spdk_blob_md_descriptor *)page->descriptors; 437 while (cur_desc < sizeof(page->descriptors)) { 438 if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_PADDING) { 439 if (desc->length == 0) { 440 /* If padding and length are 0, this terminates the page */ 441 break; 442 } 443 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_FLAGS) { 444 struct spdk_blob_md_descriptor_flags *desc_flags; 445 446 desc_flags = (struct spdk_blob_md_descriptor_flags *)desc; 447 448 if (desc_flags->length != sizeof(*desc_flags) - sizeof(*desc)) { 449 return -EINVAL; 450 } 451 452 if ((desc_flags->invalid_flags | SPDK_BLOB_INVALID_FLAGS_MASK) != 453 SPDK_BLOB_INVALID_FLAGS_MASK) { 454 return -EINVAL; 455 } 456 457 if ((desc_flags->data_ro_flags | SPDK_BLOB_DATA_RO_FLAGS_MASK) != 458 SPDK_BLOB_DATA_RO_FLAGS_MASK) { 459 blob->data_ro = true; 460 blob->md_ro = true; 461 } 462 463 if ((desc_flags->md_ro_flags | SPDK_BLOB_MD_RO_FLAGS_MASK) != 464 SPDK_BLOB_MD_RO_FLAGS_MASK) { 465 blob->md_ro = true; 466 } 467 468 if ((desc_flags->data_ro_flags & SPDK_BLOB_READ_ONLY)) { 469 blob->data_ro = true; 470 blob->md_ro = true; 471 } 472 473 blob->invalid_flags = desc_flags->invalid_flags; 474 blob->data_ro_flags = desc_flags->data_ro_flags; 475 blob->md_ro_flags = desc_flags->md_ro_flags; 476 477 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_EXTENT) { 478 struct spdk_blob_md_descriptor_extent *desc_extent; 479 unsigned int i, j; 480 unsigned int cluster_count = blob->active.num_clusters; 481 482 desc_extent = (struct spdk_blob_md_descriptor_extent *)desc; 483 484 if (desc_extent->length == 0 || 485 (desc_extent->length % sizeof(desc_extent->extents[0]) != 0)) { 486 return -EINVAL; 487 } 488 489 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 490 for (j = 0; j < desc_extent->extents[i].length; j++) { 491 if (desc_extent->extents[i].cluster_idx != 0) { 492 if (!spdk_bit_array_get(blob->bs->used_clusters, 493 desc_extent->extents[i].cluster_idx + j)) { 494 return -EINVAL; 495 } 496 } 497 cluster_count++; 498 } 499 } 500 501 if (cluster_count == 0) { 502 return -EINVAL; 503 } 504 tmp = realloc(blob->active.clusters, cluster_count * sizeof(uint64_t)); 505 if (tmp == NULL) { 506 return -ENOMEM; 507 } 508 blob->active.clusters = tmp; 509 blob->active.cluster_array_size = cluster_count; 510 511 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 512 for (j = 0; j < desc_extent->extents[i].length; j++) { 513 if (desc_extent->extents[i].cluster_idx != 0) { 514 blob->active.clusters[blob->active.num_clusters++] = _spdk_bs_cluster_to_lba(blob->bs, 515 desc_extent->extents[i].cluster_idx + j); 516 } else if (spdk_blob_is_thin_provisioned(blob)) { 517 blob->active.clusters[blob->active.num_clusters++] = 0; 518 } else { 519 return -EINVAL; 520 } 521 } 522 } 523 524 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR) { 525 int rc; 526 527 rc = _spdk_blob_deserialize_xattr(blob, 528 (struct spdk_blob_md_descriptor_xattr *) desc, false); 529 if (rc != 0) { 530 return rc; 531 } 532 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR_INTERNAL) { 533 int rc; 534 535 rc = _spdk_blob_deserialize_xattr(blob, 536 (struct spdk_blob_md_descriptor_xattr *) desc, true); 537 if (rc != 0) { 538 return rc; 539 } 540 } else { 541 /* Unrecognized descriptor type. Do not fail - just continue to the 542 * next descriptor. If this descriptor is associated with some feature 543 * defined in a newer version of blobstore, that version of blobstore 544 * should create and set an associated feature flag to specify if this 545 * blob can be loaded or not. 546 */ 547 } 548 549 /* Advance to the next descriptor */ 550 cur_desc += sizeof(*desc) + desc->length; 551 if (cur_desc + sizeof(*desc) > sizeof(page->descriptors)) { 552 break; 553 } 554 desc = (struct spdk_blob_md_descriptor *)((uintptr_t)page->descriptors + cur_desc); 555 } 556 557 return 0; 558 } 559 560 static int 561 _spdk_blob_parse(const struct spdk_blob_md_page *pages, uint32_t page_count, 562 struct spdk_blob *blob) 563 { 564 const struct spdk_blob_md_page *page; 565 uint32_t i; 566 int rc; 567 568 assert(page_count > 0); 569 assert(pages[0].sequence_num == 0); 570 assert(blob != NULL); 571 assert(blob->state == SPDK_BLOB_STATE_LOADING); 572 assert(blob->active.clusters == NULL); 573 574 /* The blobid provided doesn't match what's in the MD, this can 575 * happen for example if a bogus blobid is passed in through open. 576 */ 577 if (blob->id != pages[0].id) { 578 SPDK_ERRLOG("Blobid (%lu) doesn't match what's in metadata (%lu)\n", 579 blob->id, pages[0].id); 580 return -ENOENT; 581 } 582 583 for (i = 0; i < page_count; i++) { 584 page = &pages[i]; 585 586 assert(page->id == blob->id); 587 assert(page->sequence_num == i); 588 589 rc = _spdk_blob_parse_page(page, blob); 590 if (rc != 0) { 591 return rc; 592 } 593 } 594 595 return 0; 596 } 597 598 static int 599 _spdk_blob_serialize_add_page(const struct spdk_blob *blob, 600 struct spdk_blob_md_page **pages, 601 uint32_t *page_count, 602 struct spdk_blob_md_page **last_page) 603 { 604 struct spdk_blob_md_page *page; 605 606 assert(pages != NULL); 607 assert(page_count != NULL); 608 609 if (*page_count == 0) { 610 assert(*pages == NULL); 611 *page_count = 1; 612 *pages = spdk_malloc(SPDK_BS_PAGE_SIZE, SPDK_BS_PAGE_SIZE, 613 NULL, SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 614 } else { 615 assert(*pages != NULL); 616 (*page_count)++; 617 *pages = spdk_realloc(*pages, 618 SPDK_BS_PAGE_SIZE * (*page_count), 619 SPDK_BS_PAGE_SIZE); 620 } 621 622 if (*pages == NULL) { 623 *page_count = 0; 624 *last_page = NULL; 625 return -ENOMEM; 626 } 627 628 page = &(*pages)[*page_count - 1]; 629 memset(page, 0, sizeof(*page)); 630 page->id = blob->id; 631 page->sequence_num = *page_count - 1; 632 page->next = SPDK_INVALID_MD_PAGE; 633 *last_page = page; 634 635 return 0; 636 } 637 638 /* Transform the in-memory representation 'xattr' into an on-disk xattr descriptor. 639 * Update required_sz on both success and failure. 640 * 641 */ 642 static int 643 _spdk_blob_serialize_xattr(const struct spdk_xattr *xattr, 644 uint8_t *buf, size_t buf_sz, 645 size_t *required_sz, bool internal) 646 { 647 struct spdk_blob_md_descriptor_xattr *desc; 648 649 *required_sz = sizeof(struct spdk_blob_md_descriptor_xattr) + 650 strlen(xattr->name) + 651 xattr->value_len; 652 653 if (buf_sz < *required_sz) { 654 return -1; 655 } 656 657 desc = (struct spdk_blob_md_descriptor_xattr *)buf; 658 659 desc->type = internal ? SPDK_MD_DESCRIPTOR_TYPE_XATTR_INTERNAL : SPDK_MD_DESCRIPTOR_TYPE_XATTR; 660 desc->length = sizeof(desc->name_length) + 661 sizeof(desc->value_length) + 662 strlen(xattr->name) + 663 xattr->value_len; 664 desc->name_length = strlen(xattr->name); 665 desc->value_length = xattr->value_len; 666 667 memcpy(desc->name, xattr->name, desc->name_length); 668 memcpy((void *)((uintptr_t)desc->name + desc->name_length), 669 xattr->value, 670 desc->value_length); 671 672 return 0; 673 } 674 675 static void 676 _spdk_blob_serialize_extent(const struct spdk_blob *blob, 677 uint64_t start_cluster, uint64_t *next_cluster, 678 uint8_t *buf, size_t buf_sz) 679 { 680 struct spdk_blob_md_descriptor_extent *desc; 681 size_t cur_sz; 682 uint64_t i, extent_idx; 683 uint64_t lba, lba_per_cluster, lba_count; 684 685 /* The buffer must have room for at least one extent */ 686 cur_sz = sizeof(struct spdk_blob_md_descriptor) + sizeof(desc->extents[0]); 687 if (buf_sz < cur_sz) { 688 *next_cluster = start_cluster; 689 return; 690 } 691 692 desc = (struct spdk_blob_md_descriptor_extent *)buf; 693 desc->type = SPDK_MD_DESCRIPTOR_TYPE_EXTENT; 694 695 lba_per_cluster = _spdk_bs_cluster_to_lba(blob->bs, 1); 696 697 lba = blob->active.clusters[start_cluster]; 698 lba_count = lba_per_cluster; 699 extent_idx = 0; 700 for (i = start_cluster + 1; i < blob->active.num_clusters; i++) { 701 if ((lba + lba_count) == blob->active.clusters[i]) { 702 lba_count += lba_per_cluster; 703 continue; 704 } else if (lba == 0 && blob->active.clusters[i] == 0) { 705 lba_count += lba_per_cluster; 706 continue; 707 } 708 desc->extents[extent_idx].cluster_idx = lba / lba_per_cluster; 709 desc->extents[extent_idx].length = lba_count / lba_per_cluster; 710 extent_idx++; 711 712 cur_sz += sizeof(desc->extents[extent_idx]); 713 714 if (buf_sz < cur_sz) { 715 /* If we ran out of buffer space, return */ 716 desc->length = sizeof(desc->extents[0]) * extent_idx; 717 *next_cluster = i; 718 return; 719 } 720 721 lba = blob->active.clusters[i]; 722 lba_count = lba_per_cluster; 723 } 724 725 desc->extents[extent_idx].cluster_idx = lba / lba_per_cluster; 726 desc->extents[extent_idx].length = lba_count / lba_per_cluster; 727 extent_idx++; 728 729 desc->length = sizeof(desc->extents[0]) * extent_idx; 730 *next_cluster = blob->active.num_clusters; 731 732 return; 733 } 734 735 static void 736 _spdk_blob_serialize_flags(const struct spdk_blob *blob, 737 uint8_t *buf, size_t *buf_sz) 738 { 739 struct spdk_blob_md_descriptor_flags *desc; 740 741 /* 742 * Flags get serialized first, so we should always have room for the flags 743 * descriptor. 744 */ 745 assert(*buf_sz >= sizeof(*desc)); 746 747 desc = (struct spdk_blob_md_descriptor_flags *)buf; 748 desc->type = SPDK_MD_DESCRIPTOR_TYPE_FLAGS; 749 desc->length = sizeof(*desc) - sizeof(struct spdk_blob_md_descriptor); 750 desc->invalid_flags = blob->invalid_flags; 751 desc->data_ro_flags = blob->data_ro_flags; 752 desc->md_ro_flags = blob->md_ro_flags; 753 754 *buf_sz -= sizeof(*desc); 755 } 756 757 static int 758 _spdk_blob_serialize_xattrs(const struct spdk_blob *blob, 759 const struct spdk_xattr_tailq *xattrs, bool internal, 760 struct spdk_blob_md_page **pages, 761 struct spdk_blob_md_page *cur_page, 762 uint32_t *page_count, uint8_t **buf, 763 size_t *remaining_sz) 764 { 765 const struct spdk_xattr *xattr; 766 int rc; 767 768 TAILQ_FOREACH(xattr, xattrs, link) { 769 size_t required_sz = 0; 770 771 rc = _spdk_blob_serialize_xattr(xattr, 772 *buf, *remaining_sz, 773 &required_sz, internal); 774 if (rc < 0) { 775 /* Need to add a new page to the chain */ 776 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, 777 &cur_page); 778 if (rc < 0) { 779 spdk_free(*pages); 780 *pages = NULL; 781 *page_count = 0; 782 return rc; 783 } 784 785 *buf = (uint8_t *)cur_page->descriptors; 786 *remaining_sz = sizeof(cur_page->descriptors); 787 788 /* Try again */ 789 required_sz = 0; 790 rc = _spdk_blob_serialize_xattr(xattr, 791 *buf, *remaining_sz, 792 &required_sz, internal); 793 794 if (rc < 0) { 795 spdk_free(*pages); 796 *pages = NULL; 797 *page_count = 0; 798 return rc; 799 } 800 } 801 802 *remaining_sz -= required_sz; 803 *buf += required_sz; 804 } 805 806 return 0; 807 } 808 809 static int 810 _spdk_blob_serialize(const struct spdk_blob *blob, struct spdk_blob_md_page **pages, 811 uint32_t *page_count) 812 { 813 struct spdk_blob_md_page *cur_page; 814 int rc; 815 uint8_t *buf; 816 size_t remaining_sz; 817 uint64_t last_cluster; 818 819 assert(pages != NULL); 820 assert(page_count != NULL); 821 assert(blob != NULL); 822 assert(blob->state == SPDK_BLOB_STATE_DIRTY); 823 824 *pages = NULL; 825 *page_count = 0; 826 827 /* A blob always has at least 1 page, even if it has no descriptors */ 828 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, &cur_page); 829 if (rc < 0) { 830 return rc; 831 } 832 833 buf = (uint8_t *)cur_page->descriptors; 834 remaining_sz = sizeof(cur_page->descriptors); 835 836 /* Serialize flags */ 837 _spdk_blob_serialize_flags(blob, buf, &remaining_sz); 838 buf += sizeof(struct spdk_blob_md_descriptor_flags); 839 840 /* Serialize xattrs */ 841 rc = _spdk_blob_serialize_xattrs(blob, &blob->xattrs, false, 842 pages, cur_page, page_count, &buf, &remaining_sz); 843 if (rc < 0) { 844 return rc; 845 } 846 847 /* Serialize internal xattrs */ 848 rc = _spdk_blob_serialize_xattrs(blob, &blob->xattrs_internal, true, 849 pages, cur_page, page_count, &buf, &remaining_sz); 850 if (rc < 0) { 851 return rc; 852 } 853 854 /* Serialize extents */ 855 last_cluster = 0; 856 while (last_cluster < blob->active.num_clusters) { 857 _spdk_blob_serialize_extent(blob, last_cluster, &last_cluster, 858 buf, remaining_sz); 859 860 if (last_cluster == blob->active.num_clusters) { 861 break; 862 } 863 864 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, 865 &cur_page); 866 if (rc < 0) { 867 return rc; 868 } 869 870 buf = (uint8_t *)cur_page->descriptors; 871 remaining_sz = sizeof(cur_page->descriptors); 872 } 873 874 return 0; 875 } 876 877 struct spdk_blob_load_ctx { 878 struct spdk_blob *blob; 879 880 struct spdk_blob_md_page *pages; 881 uint32_t num_pages; 882 spdk_bs_sequence_t *seq; 883 884 spdk_bs_sequence_cpl cb_fn; 885 void *cb_arg; 886 }; 887 888 static uint32_t 889 _spdk_blob_md_page_calc_crc(void *page) 890 { 891 uint32_t crc; 892 893 crc = BLOB_CRC32C_INITIAL; 894 crc = spdk_crc32c_update(page, SPDK_BS_PAGE_SIZE - 4, crc); 895 crc ^= BLOB_CRC32C_INITIAL; 896 897 return crc; 898 899 } 900 901 static void 902 _spdk_blob_load_final(void *cb_arg, int bserrno) 903 { 904 struct spdk_blob_load_ctx *ctx = cb_arg; 905 struct spdk_blob *blob = ctx->blob; 906 907 _spdk_blob_mark_clean(blob); 908 909 ctx->cb_fn(ctx->seq, ctx->cb_arg, bserrno); 910 911 /* Free the memory */ 912 spdk_free(ctx->pages); 913 free(ctx); 914 } 915 916 static void 917 _spdk_blob_load_snapshot_cpl(void *cb_arg, struct spdk_blob *snapshot, int bserrno) 918 { 919 struct spdk_blob_load_ctx *ctx = cb_arg; 920 struct spdk_blob *blob = ctx->blob; 921 922 if (bserrno != 0) { 923 goto error; 924 } 925 926 blob->back_bs_dev = spdk_bs_create_blob_bs_dev(snapshot); 927 928 if (blob->back_bs_dev == NULL) { 929 bserrno = -ENOMEM; 930 goto error; 931 } 932 933 _spdk_blob_load_final(ctx, bserrno); 934 return; 935 936 error: 937 SPDK_ERRLOG("Snapshot fail\n"); 938 _spdk_blob_free(blob); 939 ctx->cb_fn(ctx->seq, NULL, bserrno); 940 spdk_free(ctx->pages); 941 free(ctx); 942 } 943 944 static void 945 _spdk_blob_load_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 946 { 947 struct spdk_blob_load_ctx *ctx = cb_arg; 948 struct spdk_blob *blob = ctx->blob; 949 struct spdk_blob_md_page *page; 950 const void *value; 951 size_t len; 952 int rc; 953 uint32_t crc; 954 955 if (bserrno) { 956 SPDK_ERRLOG("Metadata page read failed: %d\n", bserrno); 957 _spdk_blob_free(blob); 958 ctx->cb_fn(seq, NULL, bserrno); 959 spdk_free(ctx->pages); 960 free(ctx); 961 return; 962 } 963 964 page = &ctx->pages[ctx->num_pages - 1]; 965 crc = _spdk_blob_md_page_calc_crc(page); 966 if (crc != page->crc) { 967 SPDK_ERRLOG("Metadata page %d crc mismatch\n", ctx->num_pages); 968 _spdk_blob_free(blob); 969 ctx->cb_fn(seq, NULL, -EINVAL); 970 spdk_free(ctx->pages); 971 free(ctx); 972 return; 973 } 974 975 if (page->next != SPDK_INVALID_MD_PAGE) { 976 uint32_t next_page = page->next; 977 uint64_t next_lba = _spdk_bs_page_to_lba(blob->bs, blob->bs->md_start + next_page); 978 979 980 assert(next_lba < (blob->bs->md_start + blob->bs->md_len)); 981 982 /* Read the next page */ 983 ctx->num_pages++; 984 ctx->pages = spdk_realloc(ctx->pages, (sizeof(*page) * ctx->num_pages), 985 sizeof(*page)); 986 if (ctx->pages == NULL) { 987 ctx->cb_fn(seq, ctx->cb_arg, -ENOMEM); 988 free(ctx); 989 return; 990 } 991 992 spdk_bs_sequence_read_dev(seq, &ctx->pages[ctx->num_pages - 1], 993 next_lba, 994 _spdk_bs_byte_to_lba(blob->bs, sizeof(*page)), 995 _spdk_blob_load_cpl, ctx); 996 return; 997 } 998 999 /* Parse the pages */ 1000 rc = _spdk_blob_parse(ctx->pages, ctx->num_pages, blob); 1001 if (rc) { 1002 _spdk_blob_free(blob); 1003 ctx->cb_fn(seq, NULL, rc); 1004 spdk_free(ctx->pages); 1005 free(ctx); 1006 return; 1007 } 1008 ctx->seq = seq; 1009 1010 1011 if (spdk_blob_is_thin_provisioned(blob)) { 1012 rc = _spdk_blob_get_xattr_value(blob, BLOB_SNAPSHOT, &value, &len, true); 1013 if (rc == 0) { 1014 if (len != sizeof(spdk_blob_id)) { 1015 _spdk_blob_free(blob); 1016 ctx->cb_fn(seq, NULL, -EINVAL); 1017 spdk_free(ctx->pages); 1018 free(ctx); 1019 return; 1020 } 1021 /* open snapshot blob and continue in the callback function */ 1022 blob->parent_id = *(spdk_blob_id *)value; 1023 spdk_bs_open_blob(blob->bs, blob->parent_id, 1024 _spdk_blob_load_snapshot_cpl, ctx); 1025 return; 1026 } else { 1027 /* add zeroes_dev for thin provisioned blob */ 1028 blob->back_bs_dev = spdk_bs_create_zeroes_dev(); 1029 } 1030 } else { 1031 /* standard blob */ 1032 blob->back_bs_dev = NULL; 1033 } 1034 _spdk_blob_load_final(ctx, bserrno); 1035 } 1036 1037 /* Load a blob from disk given a blobid */ 1038 static void 1039 _spdk_blob_load(spdk_bs_sequence_t *seq, struct spdk_blob *blob, 1040 spdk_bs_sequence_cpl cb_fn, void *cb_arg) 1041 { 1042 struct spdk_blob_load_ctx *ctx; 1043 struct spdk_blob_store *bs; 1044 uint32_t page_num; 1045 uint64_t lba; 1046 1047 _spdk_blob_verify_md_op(blob); 1048 1049 bs = blob->bs; 1050 1051 ctx = calloc(1, sizeof(*ctx)); 1052 if (!ctx) { 1053 cb_fn(seq, cb_arg, -ENOMEM); 1054 return; 1055 } 1056 1057 ctx->blob = blob; 1058 ctx->pages = spdk_realloc(ctx->pages, SPDK_BS_PAGE_SIZE, SPDK_BS_PAGE_SIZE); 1059 if (!ctx->pages) { 1060 free(ctx); 1061 cb_fn(seq, cb_arg, -ENOMEM); 1062 return; 1063 } 1064 ctx->num_pages = 1; 1065 ctx->cb_fn = cb_fn; 1066 ctx->cb_arg = cb_arg; 1067 1068 page_num = _spdk_bs_blobid_to_page(blob->id); 1069 lba = _spdk_bs_page_to_lba(blob->bs, bs->md_start + page_num); 1070 1071 blob->state = SPDK_BLOB_STATE_LOADING; 1072 1073 spdk_bs_sequence_read_dev(seq, &ctx->pages[0], lba, 1074 _spdk_bs_byte_to_lba(bs, SPDK_BS_PAGE_SIZE), 1075 _spdk_blob_load_cpl, ctx); 1076 } 1077 1078 struct spdk_blob_persist_ctx { 1079 struct spdk_blob *blob; 1080 1081 struct spdk_bs_super_block *super; 1082 1083 struct spdk_blob_md_page *pages; 1084 1085 uint64_t idx; 1086 1087 spdk_bs_sequence_t *seq; 1088 spdk_bs_sequence_cpl cb_fn; 1089 void *cb_arg; 1090 }; 1091 1092 static void 1093 spdk_bs_batch_clear_dev(struct spdk_blob_persist_ctx *ctx, spdk_bs_batch_t *batch, uint64_t lba, 1094 uint32_t lba_count) 1095 { 1096 if (ctx->blob->clear_method == BLOB_CLEAR_WITH_DEFAULT || 1097 ctx->blob->clear_method == BLOB_CLEAR_WITH_UNMAP) { 1098 spdk_bs_batch_unmap_dev(batch, lba, lba_count); 1099 } else if (ctx->blob->clear_method == BLOB_CLEAR_WITH_WRITE_ZEROES) { 1100 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1101 } 1102 } 1103 1104 static void 1105 _spdk_blob_persist_complete(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1106 { 1107 struct spdk_blob_persist_ctx *ctx = cb_arg; 1108 struct spdk_blob *blob = ctx->blob; 1109 1110 if (bserrno == 0) { 1111 _spdk_blob_mark_clean(blob); 1112 } 1113 1114 /* Call user callback */ 1115 ctx->cb_fn(seq, ctx->cb_arg, bserrno); 1116 1117 /* Free the memory */ 1118 spdk_free(ctx->pages); 1119 free(ctx); 1120 } 1121 1122 static void 1123 _spdk_blob_persist_clear_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1124 { 1125 struct spdk_blob_persist_ctx *ctx = cb_arg; 1126 struct spdk_blob *blob = ctx->blob; 1127 struct spdk_blob_store *bs = blob->bs; 1128 void *tmp; 1129 size_t i; 1130 1131 /* Release all clusters that were truncated */ 1132 for (i = blob->active.num_clusters; i < blob->active.cluster_array_size; i++) { 1133 uint32_t cluster_num = _spdk_bs_lba_to_cluster(bs, blob->active.clusters[i]); 1134 1135 /* Nothing to release if it was not allocated */ 1136 if (blob->active.clusters[i] != 0) { 1137 _spdk_bs_release_cluster(bs, cluster_num); 1138 } 1139 } 1140 1141 if (blob->active.num_clusters == 0) { 1142 free(blob->active.clusters); 1143 blob->active.clusters = NULL; 1144 blob->active.cluster_array_size = 0; 1145 } else if (blob->active.num_clusters != blob->active.cluster_array_size) { 1146 tmp = realloc(blob->active.clusters, sizeof(uint64_t) * blob->active.num_clusters); 1147 assert(tmp != NULL); 1148 blob->active.clusters = tmp; 1149 blob->active.cluster_array_size = blob->active.num_clusters; 1150 } 1151 1152 _spdk_blob_persist_complete(seq, ctx, bserrno); 1153 } 1154 1155 static void 1156 _spdk_blob_persist_clear_clusters(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1157 { 1158 struct spdk_blob_persist_ctx *ctx = cb_arg; 1159 struct spdk_blob *blob = ctx->blob; 1160 struct spdk_blob_store *bs = blob->bs; 1161 spdk_bs_batch_t *batch; 1162 size_t i; 1163 uint64_t lba; 1164 uint32_t lba_count; 1165 1166 /* Clusters don't move around in blobs. The list shrinks or grows 1167 * at the end, but no changes ever occur in the middle of the list. 1168 */ 1169 1170 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_clear_clusters_cpl, ctx); 1171 1172 /* Clear all clusters that were truncated */ 1173 lba = 0; 1174 lba_count = 0; 1175 for (i = blob->active.num_clusters; i < blob->active.cluster_array_size; i++) { 1176 uint64_t next_lba = blob->active.clusters[i]; 1177 uint32_t next_lba_count = _spdk_bs_cluster_to_lba(bs, 1); 1178 1179 if (next_lba > 0 && (lba + lba_count) == next_lba) { 1180 /* This cluster is contiguous with the previous one. */ 1181 lba_count += next_lba_count; 1182 continue; 1183 } 1184 1185 /* This cluster is not contiguous with the previous one. */ 1186 1187 /* If a run of LBAs previously existing, clear them now */ 1188 if (lba_count > 0) { 1189 spdk_bs_batch_clear_dev(ctx, batch, lba, lba_count); 1190 } 1191 1192 /* Start building the next batch */ 1193 lba = next_lba; 1194 if (next_lba > 0) { 1195 lba_count = next_lba_count; 1196 } else { 1197 lba_count = 0; 1198 } 1199 } 1200 1201 /* If we ended with a contiguous set of LBAs, clear them now */ 1202 if (lba_count > 0) { 1203 spdk_bs_batch_clear_dev(ctx, batch, lba, lba_count); 1204 } 1205 1206 spdk_bs_batch_close(batch); 1207 } 1208 1209 static void 1210 _spdk_blob_persist_zero_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1211 { 1212 struct spdk_blob_persist_ctx *ctx = cb_arg; 1213 struct spdk_blob *blob = ctx->blob; 1214 struct spdk_blob_store *bs = blob->bs; 1215 size_t i; 1216 1217 /* This loop starts at 1 because the first page is special and handled 1218 * below. The pages (except the first) are never written in place, 1219 * so any pages in the clean list must be zeroed. 1220 */ 1221 for (i = 1; i < blob->clean.num_pages; i++) { 1222 spdk_bit_array_clear(bs->used_md_pages, blob->clean.pages[i]); 1223 } 1224 1225 if (blob->active.num_pages == 0) { 1226 uint32_t page_num; 1227 1228 page_num = _spdk_bs_blobid_to_page(blob->id); 1229 spdk_bit_array_clear(bs->used_md_pages, page_num); 1230 } 1231 1232 /* Move on to clearing clusters */ 1233 _spdk_blob_persist_clear_clusters(seq, ctx, 0); 1234 } 1235 1236 static void 1237 _spdk_blob_persist_zero_pages(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1238 { 1239 struct spdk_blob_persist_ctx *ctx = cb_arg; 1240 struct spdk_blob *blob = ctx->blob; 1241 struct spdk_blob_store *bs = blob->bs; 1242 uint64_t lba; 1243 uint32_t lba_count; 1244 spdk_bs_batch_t *batch; 1245 size_t i; 1246 1247 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_zero_pages_cpl, ctx); 1248 1249 lba_count = _spdk_bs_byte_to_lba(bs, SPDK_BS_PAGE_SIZE); 1250 1251 /* This loop starts at 1 because the first page is special and handled 1252 * below. The pages (except the first) are never written in place, 1253 * so any pages in the clean list must be zeroed. 1254 */ 1255 for (i = 1; i < blob->clean.num_pages; i++) { 1256 lba = _spdk_bs_page_to_lba(bs, bs->md_start + blob->clean.pages[i]); 1257 1258 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1259 } 1260 1261 /* The first page will only be zeroed if this is a delete. */ 1262 if (blob->active.num_pages == 0) { 1263 uint32_t page_num; 1264 1265 /* The first page in the metadata goes where the blobid indicates */ 1266 page_num = _spdk_bs_blobid_to_page(blob->id); 1267 lba = _spdk_bs_page_to_lba(bs, bs->md_start + page_num); 1268 1269 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1270 } 1271 1272 spdk_bs_batch_close(batch); 1273 } 1274 1275 static void 1276 _spdk_blob_persist_write_page_root(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1277 { 1278 struct spdk_blob_persist_ctx *ctx = cb_arg; 1279 struct spdk_blob *blob = ctx->blob; 1280 struct spdk_blob_store *bs = blob->bs; 1281 uint64_t lba; 1282 uint32_t lba_count; 1283 struct spdk_blob_md_page *page; 1284 1285 if (blob->active.num_pages == 0) { 1286 /* Move on to the next step */ 1287 _spdk_blob_persist_zero_pages(seq, ctx, 0); 1288 return; 1289 } 1290 1291 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(*page)); 1292 1293 page = &ctx->pages[0]; 1294 /* The first page in the metadata goes where the blobid indicates */ 1295 lba = _spdk_bs_page_to_lba(bs, bs->md_start + _spdk_bs_blobid_to_page(blob->id)); 1296 1297 spdk_bs_sequence_write_dev(seq, page, lba, lba_count, 1298 _spdk_blob_persist_zero_pages, ctx); 1299 } 1300 1301 static void 1302 _spdk_blob_persist_write_page_chain(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1303 { 1304 struct spdk_blob_persist_ctx *ctx = cb_arg; 1305 struct spdk_blob *blob = ctx->blob; 1306 struct spdk_blob_store *bs = blob->bs; 1307 uint64_t lba; 1308 uint32_t lba_count; 1309 struct spdk_blob_md_page *page; 1310 spdk_bs_batch_t *batch; 1311 size_t i; 1312 1313 /* Clusters don't move around in blobs. The list shrinks or grows 1314 * at the end, but no changes ever occur in the middle of the list. 1315 */ 1316 1317 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(*page)); 1318 1319 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_write_page_root, ctx); 1320 1321 /* This starts at 1. The root page is not written until 1322 * all of the others are finished 1323 */ 1324 for (i = 1; i < blob->active.num_pages; i++) { 1325 page = &ctx->pages[i]; 1326 assert(page->sequence_num == i); 1327 1328 lba = _spdk_bs_page_to_lba(bs, bs->md_start + blob->active.pages[i]); 1329 1330 spdk_bs_batch_write_dev(batch, page, lba, lba_count); 1331 } 1332 1333 spdk_bs_batch_close(batch); 1334 } 1335 1336 static int 1337 _spdk_blob_resize(struct spdk_blob *blob, uint64_t sz) 1338 { 1339 uint64_t i; 1340 uint64_t *tmp; 1341 uint64_t lfc; /* lowest free cluster */ 1342 uint64_t num_clusters; 1343 struct spdk_blob_store *bs; 1344 1345 bs = blob->bs; 1346 1347 _spdk_blob_verify_md_op(blob); 1348 1349 if (blob->active.num_clusters == sz) { 1350 return 0; 1351 } 1352 1353 if (blob->active.num_clusters < blob->active.cluster_array_size) { 1354 /* If this blob was resized to be larger, then smaller, then 1355 * larger without syncing, then the cluster array already 1356 * contains spare assigned clusters we can use. 1357 */ 1358 num_clusters = spdk_min(blob->active.cluster_array_size, 1359 sz); 1360 } else { 1361 num_clusters = blob->active.num_clusters; 1362 } 1363 1364 /* Do two passes - one to verify that we can obtain enough clusters 1365 * and another to actually claim them. 1366 */ 1367 1368 if (spdk_blob_is_thin_provisioned(blob) == false) { 1369 lfc = 0; 1370 for (i = num_clusters; i < sz; i++) { 1371 lfc = spdk_bit_array_find_first_clear(bs->used_clusters, lfc); 1372 if (lfc == UINT32_MAX) { 1373 /* No more free clusters. Cannot satisfy the request */ 1374 return -ENOSPC; 1375 } 1376 lfc++; 1377 } 1378 } 1379 1380 if (sz > num_clusters) { 1381 /* Expand the cluster array if necessary. 1382 * We only shrink the array when persisting. 1383 */ 1384 tmp = realloc(blob->active.clusters, sizeof(uint64_t) * sz); 1385 if (sz > 0 && tmp == NULL) { 1386 return -ENOMEM; 1387 } 1388 memset(tmp + blob->active.cluster_array_size, 0, 1389 sizeof(uint64_t) * (sz - blob->active.cluster_array_size)); 1390 blob->active.clusters = tmp; 1391 blob->active.cluster_array_size = sz; 1392 } 1393 1394 blob->state = SPDK_BLOB_STATE_DIRTY; 1395 1396 if (spdk_blob_is_thin_provisioned(blob) == false) { 1397 lfc = 0; 1398 for (i = num_clusters; i < sz; i++) { 1399 _spdk_bs_allocate_cluster(blob, i, &lfc, true); 1400 lfc++; 1401 } 1402 } 1403 1404 blob->active.num_clusters = sz; 1405 1406 return 0; 1407 } 1408 1409 static void 1410 _spdk_blob_persist_start(struct spdk_blob_persist_ctx *ctx) 1411 { 1412 spdk_bs_sequence_t *seq = ctx->seq; 1413 struct spdk_blob *blob = ctx->blob; 1414 struct spdk_blob_store *bs = blob->bs; 1415 uint64_t i; 1416 uint32_t page_num; 1417 void *tmp; 1418 int rc; 1419 1420 if (blob->active.num_pages == 0) { 1421 /* This is the signal that the blob should be deleted. 1422 * Immediately jump to the clean up routine. */ 1423 assert(blob->clean.num_pages > 0); 1424 ctx->idx = blob->clean.num_pages - 1; 1425 blob->state = SPDK_BLOB_STATE_CLEAN; 1426 _spdk_blob_persist_zero_pages(seq, ctx, 0); 1427 return; 1428 1429 } 1430 1431 /* Generate the new metadata */ 1432 rc = _spdk_blob_serialize(blob, &ctx->pages, &blob->active.num_pages); 1433 if (rc < 0) { 1434 _spdk_blob_persist_complete(seq, ctx, rc); 1435 return; 1436 } 1437 1438 assert(blob->active.num_pages >= 1); 1439 1440 /* Resize the cache of page indices */ 1441 tmp = realloc(blob->active.pages, blob->active.num_pages * sizeof(*blob->active.pages)); 1442 if (!tmp) { 1443 _spdk_blob_persist_complete(seq, ctx, -ENOMEM); 1444 return; 1445 } 1446 blob->active.pages = tmp; 1447 1448 /* Assign this metadata to pages. This requires two passes - 1449 * one to verify that there are enough pages and a second 1450 * to actually claim them. */ 1451 page_num = 0; 1452 /* Note that this loop starts at one. The first page location is fixed by the blobid. */ 1453 for (i = 1; i < blob->active.num_pages; i++) { 1454 page_num = spdk_bit_array_find_first_clear(bs->used_md_pages, page_num); 1455 if (page_num == UINT32_MAX) { 1456 _spdk_blob_persist_complete(seq, ctx, -ENOMEM); 1457 return; 1458 } 1459 page_num++; 1460 } 1461 1462 page_num = 0; 1463 blob->active.pages[0] = _spdk_bs_blobid_to_page(blob->id); 1464 for (i = 1; i < blob->active.num_pages; i++) { 1465 page_num = spdk_bit_array_find_first_clear(bs->used_md_pages, page_num); 1466 ctx->pages[i - 1].next = page_num; 1467 /* Now that previous metadata page is complete, calculate the crc for it. */ 1468 ctx->pages[i - 1].crc = _spdk_blob_md_page_calc_crc(&ctx->pages[i - 1]); 1469 blob->active.pages[i] = page_num; 1470 spdk_bit_array_set(bs->used_md_pages, page_num); 1471 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Claiming page %u for blob %lu\n", page_num, blob->id); 1472 page_num++; 1473 } 1474 ctx->pages[i - 1].crc = _spdk_blob_md_page_calc_crc(&ctx->pages[i - 1]); 1475 /* Start writing the metadata from last page to first */ 1476 ctx->idx = blob->active.num_pages - 1; 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 for %s\n", __func__); 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 }; 2548 2549 static void 2550 _spdk_bs_load_ctx_fail(spdk_bs_sequence_t *seq, struct spdk_bs_load_ctx *ctx, int bserrno) 2551 { 2552 assert(bserrno != 0); 2553 2554 spdk_free(ctx->super); 2555 spdk_bs_sequence_finish(seq, bserrno); 2556 _spdk_bs_free(ctx->bs); 2557 free(ctx); 2558 } 2559 2560 static void 2561 _spdk_bs_set_mask(struct spdk_bit_array *array, struct spdk_bs_md_mask *mask) 2562 { 2563 uint32_t i = 0; 2564 2565 while (true) { 2566 i = spdk_bit_array_find_first_set(array, i); 2567 if (i >= mask->length) { 2568 break; 2569 } 2570 mask->mask[i / 8] |= 1U << (i % 8); 2571 i++; 2572 } 2573 } 2574 2575 static int 2576 _spdk_bs_load_mask(struct spdk_bit_array **array_ptr, struct spdk_bs_md_mask *mask) 2577 { 2578 struct spdk_bit_array *array; 2579 uint32_t i; 2580 2581 if (spdk_bit_array_resize(array_ptr, mask->length) < 0) { 2582 return -ENOMEM; 2583 } 2584 2585 array = *array_ptr; 2586 for (i = 0; i < mask->length; i++) { 2587 if (mask->mask[i / 8] & (1U << (i % 8))) { 2588 spdk_bit_array_set(array, i); 2589 } 2590 } 2591 2592 return 0; 2593 } 2594 2595 static void 2596 _spdk_bs_write_super(spdk_bs_sequence_t *seq, struct spdk_blob_store *bs, 2597 struct spdk_bs_super_block *super, spdk_bs_sequence_cpl cb_fn, void *cb_arg) 2598 { 2599 /* Update the values in the super block */ 2600 super->super_blob = bs->super_blob; 2601 memcpy(&super->bstype, &bs->bstype, sizeof(bs->bstype)); 2602 super->crc = _spdk_blob_md_page_calc_crc(super); 2603 spdk_bs_sequence_write_dev(seq, super, _spdk_bs_page_to_lba(bs, 0), 2604 _spdk_bs_byte_to_lba(bs, sizeof(*super)), 2605 cb_fn, cb_arg); 2606 } 2607 2608 static void 2609 _spdk_bs_write_used_clusters(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 2610 { 2611 struct spdk_bs_load_ctx *ctx = arg; 2612 uint64_t mask_size, lba, lba_count; 2613 2614 /* Write out the used clusters mask */ 2615 mask_size = ctx->super->used_cluster_mask_len * SPDK_BS_PAGE_SIZE; 2616 ctx->mask = spdk_zmalloc(mask_size, 0x1000, NULL, 2617 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 2618 if (!ctx->mask) { 2619 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2620 return; 2621 } 2622 2623 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_CLUSTERS; 2624 ctx->mask->length = ctx->bs->total_clusters; 2625 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_clusters)); 2626 2627 _spdk_bs_set_mask(ctx->bs->used_clusters, ctx->mask); 2628 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_start); 2629 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_len); 2630 spdk_bs_sequence_write_dev(seq, ctx->mask, lba, lba_count, cb_fn, arg); 2631 } 2632 2633 static void 2634 _spdk_bs_write_used_md(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 2635 { 2636 struct spdk_bs_load_ctx *ctx = arg; 2637 uint64_t mask_size, lba, lba_count; 2638 2639 if (seq->bserrno) { 2640 _spdk_bs_load_ctx_fail(seq, ctx, seq->bserrno); 2641 return; 2642 } 2643 2644 mask_size = ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE; 2645 ctx->mask = spdk_zmalloc(mask_size, 0x1000, NULL, 2646 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 2647 if (!ctx->mask) { 2648 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2649 return; 2650 } 2651 2652 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_PAGES; 2653 ctx->mask->length = ctx->super->md_len; 2654 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_md_pages)); 2655 2656 _spdk_bs_set_mask(ctx->bs->used_md_pages, ctx->mask); 2657 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_start); 2658 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_len); 2659 spdk_bs_sequence_write_dev(seq, ctx->mask, lba, lba_count, cb_fn, arg); 2660 } 2661 2662 static void 2663 _spdk_bs_write_used_blobids(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 2664 { 2665 struct spdk_bs_load_ctx *ctx = arg; 2666 uint64_t mask_size, lba, lba_count; 2667 2668 if (ctx->super->used_blobid_mask_len == 0) { 2669 /* 2670 * This is a pre-v3 on-disk format where the blobid mask does not get 2671 * written to disk. 2672 */ 2673 cb_fn(seq, arg, 0); 2674 return; 2675 } 2676 2677 mask_size = ctx->super->used_blobid_mask_len * SPDK_BS_PAGE_SIZE; 2678 ctx->mask = spdk_zmalloc(mask_size, 0x1000, NULL, SPDK_ENV_SOCKET_ID_ANY, 2679 SPDK_MALLOC_DMA); 2680 if (!ctx->mask) { 2681 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2682 return; 2683 } 2684 2685 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_BLOBIDS; 2686 ctx->mask->length = ctx->super->md_len; 2687 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_blobids)); 2688 2689 _spdk_bs_set_mask(ctx->bs->used_blobids, ctx->mask); 2690 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_start); 2691 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_len); 2692 spdk_bs_sequence_write_dev(seq, ctx->mask, lba, lba_count, cb_fn, arg); 2693 } 2694 2695 static void 2696 _spdk_bs_load_iter(void *arg, struct spdk_blob *blob, int bserrno) 2697 { 2698 struct spdk_bs_load_ctx *ctx = arg; 2699 2700 if (bserrno == 0) { 2701 if (ctx->iter_cb_fn) { 2702 ctx->iter_cb_fn(ctx->iter_cb_arg, blob, 0); 2703 } 2704 _spdk_bs_blob_list_add(blob); 2705 spdk_bs_iter_next(ctx->bs, blob, _spdk_bs_load_iter, ctx); 2706 return; 2707 } 2708 2709 if (bserrno == -ENOENT) { 2710 bserrno = 0; 2711 } else { 2712 /* 2713 * This case needs to be looked at further. Same problem 2714 * exists with applications that rely on explicit blob 2715 * iteration. We should just skip the blob that failed 2716 * to load and continue on to the next one. 2717 */ 2718 SPDK_ERRLOG("Error in iterating blobs\n"); 2719 } 2720 2721 ctx->iter_cb_fn = NULL; 2722 2723 spdk_free(ctx->super); 2724 spdk_free(ctx->mask); 2725 spdk_bs_sequence_finish(ctx->seq, bserrno); 2726 free(ctx); 2727 } 2728 2729 static void 2730 _spdk_bs_load_complete(spdk_bs_sequence_t *seq, struct spdk_bs_load_ctx *ctx, int bserrno) 2731 { 2732 ctx->seq = seq; 2733 spdk_bs_iter_first(ctx->bs, _spdk_bs_load_iter, ctx); 2734 } 2735 2736 static void 2737 _spdk_bs_load_used_blobids_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2738 { 2739 struct spdk_bs_load_ctx *ctx = cb_arg; 2740 int rc; 2741 2742 /* The type must be correct */ 2743 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_BLOBIDS); 2744 2745 /* The length of the mask (in bits) must not be greater than 2746 * the length of the buffer (converted to bits) */ 2747 assert(ctx->mask->length <= (ctx->super->used_blobid_mask_len * SPDK_BS_PAGE_SIZE * 8)); 2748 2749 /* The length of the mask must be exactly equal to the size 2750 * (in pages) of the metadata region */ 2751 assert(ctx->mask->length == ctx->super->md_len); 2752 2753 rc = _spdk_bs_load_mask(&ctx->bs->used_blobids, ctx->mask); 2754 if (rc < 0) { 2755 spdk_free(ctx->mask); 2756 _spdk_bs_load_ctx_fail(seq, ctx, rc); 2757 return; 2758 } 2759 2760 _spdk_bs_load_complete(seq, ctx, bserrno); 2761 } 2762 2763 static void 2764 _spdk_bs_load_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2765 { 2766 struct spdk_bs_load_ctx *ctx = cb_arg; 2767 uint64_t lba, lba_count, mask_size; 2768 int rc; 2769 2770 /* The type must be correct */ 2771 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_CLUSTERS); 2772 /* The length of the mask (in bits) must not be greater than the length of the buffer (converted to bits) */ 2773 assert(ctx->mask->length <= (ctx->super->used_cluster_mask_len * sizeof( 2774 struct spdk_blob_md_page) * 8)); 2775 /* The length of the mask must be exactly equal to the total number of clusters */ 2776 assert(ctx->mask->length == ctx->bs->total_clusters); 2777 2778 rc = _spdk_bs_load_mask(&ctx->bs->used_clusters, ctx->mask); 2779 if (rc < 0) { 2780 spdk_free(ctx->mask); 2781 _spdk_bs_load_ctx_fail(seq, ctx, rc); 2782 return; 2783 } 2784 2785 ctx->bs->num_free_clusters = spdk_bit_array_count_clear(ctx->bs->used_clusters); 2786 assert(ctx->bs->num_free_clusters <= ctx->bs->total_clusters); 2787 2788 spdk_free(ctx->mask); 2789 2790 /* Read the used blobids mask */ 2791 mask_size = ctx->super->used_blobid_mask_len * SPDK_BS_PAGE_SIZE; 2792 ctx->mask = spdk_zmalloc(mask_size, 0x1000, NULL, SPDK_ENV_SOCKET_ID_ANY, 2793 SPDK_MALLOC_DMA); 2794 if (!ctx->mask) { 2795 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2796 return; 2797 } 2798 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_start); 2799 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_len); 2800 spdk_bs_sequence_read_dev(seq, ctx->mask, lba, lba_count, 2801 _spdk_bs_load_used_blobids_cpl, ctx); 2802 } 2803 2804 static void 2805 _spdk_bs_load_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2806 { 2807 struct spdk_bs_load_ctx *ctx = cb_arg; 2808 uint64_t lba, lba_count, mask_size; 2809 int rc; 2810 2811 /* The type must be correct */ 2812 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_PAGES); 2813 /* The length of the mask (in bits) must not be greater than the length of the buffer (converted to bits) */ 2814 assert(ctx->mask->length <= (ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE * 2815 8)); 2816 /* The length of the mask must be exactly equal to the size (in pages) of the metadata region */ 2817 assert(ctx->mask->length == ctx->super->md_len); 2818 2819 rc = _spdk_bs_load_mask(&ctx->bs->used_md_pages, ctx->mask); 2820 if (rc < 0) { 2821 spdk_free(ctx->mask); 2822 _spdk_bs_load_ctx_fail(seq, ctx, rc); 2823 return; 2824 } 2825 2826 spdk_free(ctx->mask); 2827 2828 /* Read the used clusters mask */ 2829 mask_size = ctx->super->used_cluster_mask_len * SPDK_BS_PAGE_SIZE; 2830 ctx->mask = spdk_zmalloc(mask_size, 0x1000, NULL, SPDK_ENV_SOCKET_ID_ANY, 2831 SPDK_MALLOC_DMA); 2832 if (!ctx->mask) { 2833 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2834 return; 2835 } 2836 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_start); 2837 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_len); 2838 spdk_bs_sequence_read_dev(seq, ctx->mask, lba, lba_count, 2839 _spdk_bs_load_used_clusters_cpl, ctx); 2840 } 2841 2842 static void 2843 _spdk_bs_load_read_used_pages(spdk_bs_sequence_t *seq, void *cb_arg) 2844 { 2845 struct spdk_bs_load_ctx *ctx = cb_arg; 2846 uint64_t lba, lba_count, mask_size; 2847 2848 /* Read the used pages mask */ 2849 mask_size = ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE; 2850 ctx->mask = spdk_zmalloc(mask_size, 0x1000, NULL, 2851 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 2852 if (!ctx->mask) { 2853 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2854 return; 2855 } 2856 2857 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_start); 2858 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_len); 2859 spdk_bs_sequence_read_dev(seq, ctx->mask, lba, lba_count, 2860 _spdk_bs_load_used_pages_cpl, ctx); 2861 } 2862 2863 static int 2864 _spdk_bs_load_replay_md_parse_page(const struct spdk_blob_md_page *page, struct spdk_blob_store *bs) 2865 { 2866 struct spdk_blob_md_descriptor *desc; 2867 size_t cur_desc = 0; 2868 2869 desc = (struct spdk_blob_md_descriptor *)page->descriptors; 2870 while (cur_desc < sizeof(page->descriptors)) { 2871 if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_PADDING) { 2872 if (desc->length == 0) { 2873 /* If padding and length are 0, this terminates the page */ 2874 break; 2875 } 2876 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_EXTENT) { 2877 struct spdk_blob_md_descriptor_extent *desc_extent; 2878 unsigned int i, j; 2879 unsigned int cluster_count = 0; 2880 uint32_t cluster_idx; 2881 2882 desc_extent = (struct spdk_blob_md_descriptor_extent *)desc; 2883 2884 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 2885 for (j = 0; j < desc_extent->extents[i].length; j++) { 2886 cluster_idx = desc_extent->extents[i].cluster_idx; 2887 /* 2888 * cluster_idx = 0 means an unallocated cluster - don't mark that 2889 * in the used cluster map. 2890 */ 2891 if (cluster_idx != 0) { 2892 spdk_bit_array_set(bs->used_clusters, cluster_idx + j); 2893 if (bs->num_free_clusters == 0) { 2894 return -ENOSPC; 2895 } 2896 bs->num_free_clusters--; 2897 } 2898 cluster_count++; 2899 } 2900 } 2901 if (cluster_count == 0) { 2902 return -EINVAL; 2903 } 2904 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR) { 2905 /* Skip this item */ 2906 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR_INTERNAL) { 2907 /* Skip this item */ 2908 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_FLAGS) { 2909 /* Skip this item */ 2910 } else { 2911 /* Error */ 2912 return -EINVAL; 2913 } 2914 /* Advance to the next descriptor */ 2915 cur_desc += sizeof(*desc) + desc->length; 2916 if (cur_desc + sizeof(*desc) > sizeof(page->descriptors)) { 2917 break; 2918 } 2919 desc = (struct spdk_blob_md_descriptor *)((uintptr_t)page->descriptors + cur_desc); 2920 } 2921 return 0; 2922 } 2923 2924 static bool _spdk_bs_load_cur_md_page_valid(struct spdk_bs_load_ctx *ctx) 2925 { 2926 uint32_t crc; 2927 2928 crc = _spdk_blob_md_page_calc_crc(ctx->page); 2929 if (crc != ctx->page->crc) { 2930 return false; 2931 } 2932 2933 if (_spdk_bs_page_to_blobid(ctx->cur_page) != ctx->page->id) { 2934 return false; 2935 } 2936 return true; 2937 } 2938 2939 static void 2940 _spdk_bs_load_replay_cur_md_page(spdk_bs_sequence_t *seq, void *cb_arg); 2941 2942 static void 2943 _spdk_bs_load_write_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2944 { 2945 struct spdk_bs_load_ctx *ctx = cb_arg; 2946 2947 _spdk_bs_load_complete(seq, ctx, bserrno); 2948 } 2949 2950 static void 2951 _spdk_bs_load_write_used_blobids_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2952 { 2953 struct spdk_bs_load_ctx *ctx = cb_arg; 2954 2955 spdk_free(ctx->mask); 2956 ctx->mask = NULL; 2957 2958 _spdk_bs_write_used_clusters(seq, cb_arg, _spdk_bs_load_write_used_clusters_cpl); 2959 } 2960 2961 static void 2962 _spdk_bs_load_write_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2963 { 2964 struct spdk_bs_load_ctx *ctx = cb_arg; 2965 2966 spdk_free(ctx->mask); 2967 ctx->mask = NULL; 2968 2969 _spdk_bs_write_used_blobids(seq, cb_arg, _spdk_bs_load_write_used_blobids_cpl); 2970 } 2971 2972 static void 2973 _spdk_bs_load_write_used_md(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2974 { 2975 _spdk_bs_write_used_md(seq, cb_arg, _spdk_bs_load_write_used_pages_cpl); 2976 } 2977 2978 static void 2979 _spdk_bs_load_replay_md_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2980 { 2981 struct spdk_bs_load_ctx *ctx = cb_arg; 2982 uint64_t num_md_clusters; 2983 uint64_t i; 2984 uint32_t page_num; 2985 2986 if (bserrno != 0) { 2987 _spdk_bs_load_ctx_fail(seq, ctx, bserrno); 2988 return; 2989 } 2990 2991 page_num = ctx->cur_page; 2992 if (_spdk_bs_load_cur_md_page_valid(ctx) == true) { 2993 if (ctx->page->sequence_num == 0 || ctx->in_page_chain == true) { 2994 spdk_bit_array_set(ctx->bs->used_md_pages, page_num); 2995 if (ctx->page->sequence_num == 0) { 2996 spdk_bit_array_set(ctx->bs->used_blobids, page_num); 2997 } 2998 if (_spdk_bs_load_replay_md_parse_page(ctx->page, ctx->bs)) { 2999 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 3000 return; 3001 } 3002 if (ctx->page->next != SPDK_INVALID_MD_PAGE) { 3003 ctx->in_page_chain = true; 3004 ctx->cur_page = ctx->page->next; 3005 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 3006 return; 3007 } 3008 } 3009 } 3010 3011 ctx->in_page_chain = false; 3012 3013 do { 3014 ctx->page_index++; 3015 } while (spdk_bit_array_get(ctx->bs->used_md_pages, ctx->page_index) == true); 3016 3017 if (ctx->page_index < ctx->super->md_len) { 3018 ctx->cur_page = ctx->page_index; 3019 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 3020 } else { 3021 /* Claim all of the clusters used by the metadata */ 3022 num_md_clusters = spdk_divide_round_up(ctx->super->md_len, ctx->bs->pages_per_cluster); 3023 for (i = 0; i < num_md_clusters; i++) { 3024 _spdk_bs_claim_cluster(ctx->bs, i); 3025 } 3026 spdk_free(ctx->page); 3027 _spdk_bs_load_write_used_md(seq, ctx, bserrno); 3028 } 3029 } 3030 3031 static void 3032 _spdk_bs_load_replay_cur_md_page(spdk_bs_sequence_t *seq, void *cb_arg) 3033 { 3034 struct spdk_bs_load_ctx *ctx = cb_arg; 3035 uint64_t lba; 3036 3037 assert(ctx->cur_page < ctx->super->md_len); 3038 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->md_start + ctx->cur_page); 3039 spdk_bs_sequence_read_dev(seq, ctx->page, lba, 3040 _spdk_bs_byte_to_lba(ctx->bs, SPDK_BS_PAGE_SIZE), 3041 _spdk_bs_load_replay_md_cpl, ctx); 3042 } 3043 3044 static void 3045 _spdk_bs_load_replay_md(spdk_bs_sequence_t *seq, void *cb_arg) 3046 { 3047 struct spdk_bs_load_ctx *ctx = cb_arg; 3048 3049 ctx->page_index = 0; 3050 ctx->cur_page = 0; 3051 ctx->page = spdk_zmalloc(SPDK_BS_PAGE_SIZE, SPDK_BS_PAGE_SIZE, 3052 NULL, SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 3053 if (!ctx->page) { 3054 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 3055 return; 3056 } 3057 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 3058 } 3059 3060 static void 3061 _spdk_bs_recover(spdk_bs_sequence_t *seq, void *cb_arg) 3062 { 3063 struct spdk_bs_load_ctx *ctx = cb_arg; 3064 int rc; 3065 3066 rc = spdk_bit_array_resize(&ctx->bs->used_md_pages, ctx->super->md_len); 3067 if (rc < 0) { 3068 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 3069 return; 3070 } 3071 3072 rc = spdk_bit_array_resize(&ctx->bs->used_blobids, ctx->super->md_len); 3073 if (rc < 0) { 3074 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 3075 return; 3076 } 3077 3078 rc = spdk_bit_array_resize(&ctx->bs->used_clusters, ctx->bs->total_clusters); 3079 if (rc < 0) { 3080 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 3081 return; 3082 } 3083 3084 ctx->bs->num_free_clusters = ctx->bs->total_clusters; 3085 _spdk_bs_load_replay_md(seq, cb_arg); 3086 } 3087 3088 static void 3089 _spdk_bs_load_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3090 { 3091 struct spdk_bs_load_ctx *ctx = cb_arg; 3092 uint32_t crc; 3093 int rc; 3094 static const char zeros[SPDK_BLOBSTORE_TYPE_LENGTH]; 3095 3096 if (ctx->super->version > SPDK_BS_VERSION || 3097 ctx->super->version < SPDK_BS_INITIAL_VERSION) { 3098 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 3099 return; 3100 } 3101 3102 if (memcmp(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 3103 sizeof(ctx->super->signature)) != 0) { 3104 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 3105 return; 3106 } 3107 3108 crc = _spdk_blob_md_page_calc_crc(ctx->super); 3109 if (crc != ctx->super->crc) { 3110 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 3111 return; 3112 } 3113 3114 if (memcmp(&ctx->bs->bstype, &ctx->super->bstype, SPDK_BLOBSTORE_TYPE_LENGTH) == 0) { 3115 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Bstype matched - loading blobstore\n"); 3116 } else if (memcmp(&ctx->bs->bstype, zeros, SPDK_BLOBSTORE_TYPE_LENGTH) == 0) { 3117 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Bstype wildcard used - loading blobstore regardless bstype\n"); 3118 } else { 3119 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Unexpected bstype\n"); 3120 SPDK_LOGDUMP(SPDK_LOG_BLOB, "Expected:", ctx->bs->bstype.bstype, SPDK_BLOBSTORE_TYPE_LENGTH); 3121 SPDK_LOGDUMP(SPDK_LOG_BLOB, "Found:", ctx->super->bstype.bstype, SPDK_BLOBSTORE_TYPE_LENGTH); 3122 _spdk_bs_load_ctx_fail(seq, ctx, -ENXIO); 3123 return; 3124 } 3125 3126 if (ctx->super->size > ctx->bs->dev->blockcnt * ctx->bs->dev->blocklen) { 3127 SPDK_NOTICELOG("Size mismatch, dev size: %lu, blobstore size: %lu\n", 3128 ctx->bs->dev->blockcnt * ctx->bs->dev->blocklen, ctx->super->size); 3129 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 3130 return; 3131 } 3132 3133 if (ctx->super->size == 0) { 3134 ctx->super->size = ctx->bs->dev->blockcnt * ctx->bs->dev->blocklen; 3135 } 3136 3137 if (ctx->super->io_unit_size == 0) { 3138 ctx->super->io_unit_size = SPDK_BS_PAGE_SIZE; 3139 } 3140 3141 /* Parse the super block */ 3142 ctx->bs->clean = 1; 3143 ctx->bs->cluster_sz = ctx->super->cluster_size; 3144 ctx->bs->total_clusters = ctx->super->size / ctx->super->cluster_size; 3145 ctx->bs->pages_per_cluster = ctx->bs->cluster_sz / SPDK_BS_PAGE_SIZE; 3146 ctx->bs->io_unit_size = ctx->super->io_unit_size; 3147 rc = spdk_bit_array_resize(&ctx->bs->used_clusters, ctx->bs->total_clusters); 3148 if (rc < 0) { 3149 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 3150 return; 3151 } 3152 ctx->bs->md_start = ctx->super->md_start; 3153 ctx->bs->md_len = ctx->super->md_len; 3154 ctx->bs->total_data_clusters = ctx->bs->total_clusters - spdk_divide_round_up( 3155 ctx->bs->md_start + ctx->bs->md_len, ctx->bs->pages_per_cluster); 3156 ctx->bs->super_blob = ctx->super->super_blob; 3157 memcpy(&ctx->bs->bstype, &ctx->super->bstype, sizeof(ctx->super->bstype)); 3158 3159 if (ctx->super->used_blobid_mask_len == 0 || ctx->super->clean == 0) { 3160 _spdk_bs_recover(seq, ctx); 3161 } else { 3162 _spdk_bs_load_read_used_pages(seq, ctx); 3163 } 3164 } 3165 3166 void 3167 spdk_bs_load(struct spdk_bs_dev *dev, struct spdk_bs_opts *o, 3168 spdk_bs_op_with_handle_complete cb_fn, void *cb_arg) 3169 { 3170 struct spdk_blob_store *bs; 3171 struct spdk_bs_cpl cpl; 3172 spdk_bs_sequence_t *seq; 3173 struct spdk_bs_load_ctx *ctx; 3174 struct spdk_bs_opts opts = {}; 3175 int err; 3176 3177 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Loading blobstore from dev %p\n", dev); 3178 3179 if ((SPDK_BS_PAGE_SIZE % dev->blocklen) != 0) { 3180 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "unsupported dev block length of %d\n", dev->blocklen); 3181 dev->destroy(dev); 3182 cb_fn(cb_arg, NULL, -EINVAL); 3183 return; 3184 } 3185 3186 if (o) { 3187 opts = *o; 3188 } else { 3189 spdk_bs_opts_init(&opts); 3190 } 3191 3192 if (opts.max_md_ops == 0 || opts.max_channel_ops == 0) { 3193 dev->destroy(dev); 3194 cb_fn(cb_arg, NULL, -EINVAL); 3195 return; 3196 } 3197 3198 err = _spdk_bs_alloc(dev, &opts, &bs); 3199 if (err) { 3200 dev->destroy(dev); 3201 cb_fn(cb_arg, NULL, err); 3202 return; 3203 } 3204 3205 ctx = calloc(1, sizeof(*ctx)); 3206 if (!ctx) { 3207 _spdk_bs_free(bs); 3208 cb_fn(cb_arg, NULL, -ENOMEM); 3209 return; 3210 } 3211 3212 ctx->bs = bs; 3213 ctx->iter_cb_fn = opts.iter_cb_fn; 3214 ctx->iter_cb_arg = opts.iter_cb_arg; 3215 3216 /* Allocate memory for the super block */ 3217 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL, 3218 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 3219 if (!ctx->super) { 3220 free(ctx); 3221 _spdk_bs_free(bs); 3222 cb_fn(cb_arg, NULL, -ENOMEM); 3223 return; 3224 } 3225 3226 cpl.type = SPDK_BS_CPL_TYPE_BS_HANDLE; 3227 cpl.u.bs_handle.cb_fn = cb_fn; 3228 cpl.u.bs_handle.cb_arg = cb_arg; 3229 cpl.u.bs_handle.bs = bs; 3230 3231 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3232 if (!seq) { 3233 spdk_free(ctx->super); 3234 free(ctx); 3235 _spdk_bs_free(bs); 3236 cb_fn(cb_arg, NULL, -ENOMEM); 3237 return; 3238 } 3239 3240 /* Read the super block */ 3241 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 3242 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 3243 _spdk_bs_load_super_cpl, ctx); 3244 } 3245 3246 /* END spdk_bs_load */ 3247 3248 /* START spdk_bs_dump */ 3249 3250 struct spdk_bs_dump_ctx { 3251 struct spdk_blob_store *bs; 3252 struct spdk_bs_super_block *super; 3253 uint32_t cur_page; 3254 struct spdk_blob_md_page *page; 3255 spdk_bs_sequence_t *seq; 3256 FILE *fp; 3257 spdk_bs_dump_print_xattr print_xattr_fn; 3258 char xattr_name[4096]; 3259 }; 3260 3261 static void 3262 _spdk_bs_dump_finish(spdk_bs_sequence_t *seq, struct spdk_bs_dump_ctx *ctx, int bserrno) 3263 { 3264 spdk_free(ctx->super); 3265 3266 /* 3267 * We need to defer calling spdk_bs_call_cpl() until after 3268 * dev destruction, so tuck these away for later use. 3269 */ 3270 ctx->bs->unload_err = bserrno; 3271 memcpy(&ctx->bs->unload_cpl, &seq->cpl, sizeof(struct spdk_bs_cpl)); 3272 seq->cpl.type = SPDK_BS_CPL_TYPE_NONE; 3273 3274 spdk_bs_sequence_finish(seq, 0); 3275 _spdk_bs_free(ctx->bs); 3276 free(ctx); 3277 } 3278 3279 static void _spdk_bs_dump_read_md_page(spdk_bs_sequence_t *seq, void *cb_arg); 3280 3281 static void 3282 _spdk_bs_dump_print_md_page(struct spdk_bs_dump_ctx *ctx) 3283 { 3284 uint32_t page_idx = ctx->cur_page; 3285 struct spdk_blob_md_page *page = ctx->page; 3286 struct spdk_blob_md_descriptor *desc; 3287 size_t cur_desc = 0; 3288 uint32_t crc; 3289 3290 fprintf(ctx->fp, "=========\n"); 3291 fprintf(ctx->fp, "Metadata Page Index: %" PRIu32 " (0x%" PRIx32 ")\n", page_idx, page_idx); 3292 fprintf(ctx->fp, "Blob ID: 0x%" PRIx64 "\n", page->id); 3293 3294 crc = _spdk_blob_md_page_calc_crc(page); 3295 fprintf(ctx->fp, "CRC: 0x%" PRIx32 " (%s)\n", page->crc, crc == page->crc ? "OK" : "Mismatch"); 3296 3297 desc = (struct spdk_blob_md_descriptor *)page->descriptors; 3298 while (cur_desc < sizeof(page->descriptors)) { 3299 if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_PADDING) { 3300 if (desc->length == 0) { 3301 /* If padding and length are 0, this terminates the page */ 3302 break; 3303 } 3304 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_EXTENT) { 3305 struct spdk_blob_md_descriptor_extent *desc_extent; 3306 unsigned int i; 3307 3308 desc_extent = (struct spdk_blob_md_descriptor_extent *)desc; 3309 3310 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 3311 if (desc_extent->extents[i].cluster_idx != 0) { 3312 fprintf(ctx->fp, "Allocated Extent - Start: %" PRIu32, 3313 desc_extent->extents[i].cluster_idx); 3314 } else { 3315 fprintf(ctx->fp, "Unallocated Extent - "); 3316 } 3317 fprintf(ctx->fp, " Length: %" PRIu32, desc_extent->extents[i].length); 3318 fprintf(ctx->fp, "\n"); 3319 } 3320 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR) { 3321 struct spdk_blob_md_descriptor_xattr *desc_xattr; 3322 uint32_t i; 3323 3324 desc_xattr = (struct spdk_blob_md_descriptor_xattr *)desc; 3325 3326 if (desc_xattr->length != 3327 sizeof(desc_xattr->name_length) + sizeof(desc_xattr->value_length) + 3328 desc_xattr->name_length + desc_xattr->value_length) { 3329 } 3330 3331 memcpy(ctx->xattr_name, desc_xattr->name, desc_xattr->name_length); 3332 ctx->xattr_name[desc_xattr->name_length] = '\0'; 3333 fprintf(ctx->fp, "XATTR: name = \"%s\"\n", ctx->xattr_name); 3334 fprintf(ctx->fp, " value = \""); 3335 ctx->print_xattr_fn(ctx->fp, ctx->super->bstype.bstype, ctx->xattr_name, 3336 (void *)((uintptr_t)desc_xattr->name + desc_xattr->name_length), 3337 desc_xattr->value_length); 3338 fprintf(ctx->fp, "\"\n"); 3339 for (i = 0; i < desc_xattr->value_length; i++) { 3340 if (i % 16 == 0) { 3341 fprintf(ctx->fp, " "); 3342 } 3343 fprintf(ctx->fp, "%02" PRIx8 " ", *((uint8_t *)desc_xattr->name + desc_xattr->name_length + i)); 3344 if ((i + 1) % 16 == 0) { 3345 fprintf(ctx->fp, "\n"); 3346 } 3347 } 3348 if (i % 16 != 0) { 3349 fprintf(ctx->fp, "\n"); 3350 } 3351 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR_INTERNAL) { 3352 /* TODO */ 3353 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_FLAGS) { 3354 /* TODO */ 3355 } else { 3356 /* Error */ 3357 } 3358 /* Advance to the next descriptor */ 3359 cur_desc += sizeof(*desc) + desc->length; 3360 if (cur_desc + sizeof(*desc) > sizeof(page->descriptors)) { 3361 break; 3362 } 3363 desc = (struct spdk_blob_md_descriptor *)((uintptr_t)page->descriptors + cur_desc); 3364 } 3365 } 3366 3367 static void 3368 _spdk_bs_dump_read_md_page_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3369 { 3370 struct spdk_bs_dump_ctx *ctx = cb_arg; 3371 3372 if (bserrno != 0) { 3373 _spdk_bs_dump_finish(seq, ctx, bserrno); 3374 return; 3375 } 3376 3377 if (ctx->page->id != 0) { 3378 _spdk_bs_dump_print_md_page(ctx); 3379 } 3380 3381 ctx->cur_page++; 3382 3383 if (ctx->cur_page < ctx->super->md_len) { 3384 _spdk_bs_dump_read_md_page(seq, cb_arg); 3385 } else { 3386 spdk_free(ctx->page); 3387 _spdk_bs_dump_finish(seq, ctx, 0); 3388 } 3389 } 3390 3391 static void 3392 _spdk_bs_dump_read_md_page(spdk_bs_sequence_t *seq, void *cb_arg) 3393 { 3394 struct spdk_bs_dump_ctx *ctx = cb_arg; 3395 uint64_t lba; 3396 3397 assert(ctx->cur_page < ctx->super->md_len); 3398 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->md_start + ctx->cur_page); 3399 spdk_bs_sequence_read_dev(seq, ctx->page, lba, 3400 _spdk_bs_byte_to_lba(ctx->bs, SPDK_BS_PAGE_SIZE), 3401 _spdk_bs_dump_read_md_page_cpl, ctx); 3402 } 3403 3404 static void 3405 _spdk_bs_dump_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3406 { 3407 struct spdk_bs_dump_ctx *ctx = cb_arg; 3408 3409 fprintf(ctx->fp, "Signature: \"%.8s\" ", ctx->super->signature); 3410 if (memcmp(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 3411 sizeof(ctx->super->signature)) != 0) { 3412 fprintf(ctx->fp, "(Mismatch)\n"); 3413 _spdk_bs_dump_finish(seq, ctx, bserrno); 3414 return; 3415 } else { 3416 fprintf(ctx->fp, "(OK)\n"); 3417 } 3418 fprintf(ctx->fp, "Version: %" PRIu32 "\n", ctx->super->version); 3419 fprintf(ctx->fp, "CRC: 0x%x (%s)\n", ctx->super->crc, 3420 (ctx->super->crc == _spdk_blob_md_page_calc_crc(ctx->super)) ? "OK" : "Mismatch"); 3421 fprintf(ctx->fp, "Blobstore Type: %.*s\n", SPDK_BLOBSTORE_TYPE_LENGTH, ctx->super->bstype.bstype); 3422 fprintf(ctx->fp, "Cluster Size: %" PRIu32 "\n", ctx->super->cluster_size); 3423 fprintf(ctx->fp, "Super Blob ID: "); 3424 if (ctx->super->super_blob == SPDK_BLOBID_INVALID) { 3425 fprintf(ctx->fp, "(None)\n"); 3426 } else { 3427 fprintf(ctx->fp, "%" PRIu64 "\n", ctx->super->super_blob); 3428 } 3429 fprintf(ctx->fp, "Clean: %" PRIu32 "\n", ctx->super->clean); 3430 fprintf(ctx->fp, "Used Metadata Page Mask Start: %" PRIu32 "\n", ctx->super->used_page_mask_start); 3431 fprintf(ctx->fp, "Used Metadata Page Mask Length: %" PRIu32 "\n", ctx->super->used_page_mask_len); 3432 fprintf(ctx->fp, "Used Cluster Mask Start: %" PRIu32 "\n", ctx->super->used_cluster_mask_start); 3433 fprintf(ctx->fp, "Used Cluster Mask Length: %" PRIu32 "\n", ctx->super->used_cluster_mask_len); 3434 fprintf(ctx->fp, "Used Blob ID Mask Start: %" PRIu32 "\n", ctx->super->used_blobid_mask_start); 3435 fprintf(ctx->fp, "Used Blob ID Mask Length: %" PRIu32 "\n", ctx->super->used_blobid_mask_len); 3436 fprintf(ctx->fp, "Metadata Start: %" PRIu32 "\n", ctx->super->md_start); 3437 fprintf(ctx->fp, "Metadata Length: %" PRIu32 "\n", ctx->super->md_len); 3438 3439 ctx->cur_page = 0; 3440 ctx->page = spdk_zmalloc(SPDK_BS_PAGE_SIZE, SPDK_BS_PAGE_SIZE, 3441 NULL, SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 3442 if (!ctx->page) { 3443 _spdk_bs_dump_finish(seq, ctx, -ENOMEM); 3444 return; 3445 } 3446 _spdk_bs_dump_read_md_page(seq, cb_arg); 3447 } 3448 3449 void 3450 spdk_bs_dump(struct spdk_bs_dev *dev, FILE *fp, spdk_bs_dump_print_xattr print_xattr_fn, 3451 spdk_bs_op_complete cb_fn, void *cb_arg) 3452 { 3453 struct spdk_blob_store *bs; 3454 struct spdk_bs_cpl cpl; 3455 spdk_bs_sequence_t *seq; 3456 struct spdk_bs_dump_ctx *ctx; 3457 struct spdk_bs_opts opts = {}; 3458 int err; 3459 3460 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Dumping blobstore from dev %p\n", dev); 3461 3462 spdk_bs_opts_init(&opts); 3463 3464 err = _spdk_bs_alloc(dev, &opts, &bs); 3465 if (err) { 3466 dev->destroy(dev); 3467 cb_fn(cb_arg, err); 3468 return; 3469 } 3470 3471 ctx = calloc(1, sizeof(*ctx)); 3472 if (!ctx) { 3473 _spdk_bs_free(bs); 3474 cb_fn(cb_arg, -ENOMEM); 3475 return; 3476 } 3477 3478 ctx->bs = bs; 3479 ctx->fp = fp; 3480 ctx->print_xattr_fn = print_xattr_fn; 3481 3482 /* Allocate memory for the super block */ 3483 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL, 3484 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 3485 if (!ctx->super) { 3486 free(ctx); 3487 _spdk_bs_free(bs); 3488 cb_fn(cb_arg, -ENOMEM); 3489 return; 3490 } 3491 3492 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 3493 cpl.u.bs_basic.cb_fn = cb_fn; 3494 cpl.u.bs_basic.cb_arg = cb_arg; 3495 3496 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3497 if (!seq) { 3498 spdk_free(ctx->super); 3499 free(ctx); 3500 _spdk_bs_free(bs); 3501 cb_fn(cb_arg, -ENOMEM); 3502 return; 3503 } 3504 3505 /* Read the super block */ 3506 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 3507 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 3508 _spdk_bs_dump_super_cpl, ctx); 3509 } 3510 3511 /* END spdk_bs_dump */ 3512 3513 /* START spdk_bs_init */ 3514 3515 struct spdk_bs_init_ctx { 3516 struct spdk_blob_store *bs; 3517 struct spdk_bs_super_block *super; 3518 }; 3519 3520 static void 3521 _spdk_bs_init_persist_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3522 { 3523 struct spdk_bs_init_ctx *ctx = cb_arg; 3524 3525 spdk_free(ctx->super); 3526 free(ctx); 3527 3528 spdk_bs_sequence_finish(seq, bserrno); 3529 } 3530 3531 static void 3532 _spdk_bs_init_trim_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3533 { 3534 struct spdk_bs_init_ctx *ctx = cb_arg; 3535 3536 /* Write super block */ 3537 spdk_bs_sequence_write_dev(seq, ctx->super, _spdk_bs_page_to_lba(ctx->bs, 0), 3538 _spdk_bs_byte_to_lba(ctx->bs, sizeof(*ctx->super)), 3539 _spdk_bs_init_persist_super_cpl, ctx); 3540 } 3541 3542 void 3543 spdk_bs_init(struct spdk_bs_dev *dev, struct spdk_bs_opts *o, 3544 spdk_bs_op_with_handle_complete cb_fn, void *cb_arg) 3545 { 3546 struct spdk_bs_init_ctx *ctx; 3547 struct spdk_blob_store *bs; 3548 struct spdk_bs_cpl cpl; 3549 spdk_bs_sequence_t *seq; 3550 spdk_bs_batch_t *batch; 3551 uint64_t num_md_lba; 3552 uint64_t num_md_pages; 3553 uint64_t num_md_clusters; 3554 uint32_t i; 3555 struct spdk_bs_opts opts = {}; 3556 int rc; 3557 3558 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Initializing blobstore on dev %p\n", dev); 3559 3560 if ((SPDK_BS_PAGE_SIZE % dev->blocklen) != 0) { 3561 SPDK_ERRLOG("unsupported dev block length of %d\n", 3562 dev->blocklen); 3563 dev->destroy(dev); 3564 cb_fn(cb_arg, NULL, -EINVAL); 3565 return; 3566 } 3567 3568 if (o) { 3569 opts = *o; 3570 } else { 3571 spdk_bs_opts_init(&opts); 3572 } 3573 3574 if (_spdk_bs_opts_verify(&opts) != 0) { 3575 dev->destroy(dev); 3576 cb_fn(cb_arg, NULL, -EINVAL); 3577 return; 3578 } 3579 3580 rc = _spdk_bs_alloc(dev, &opts, &bs); 3581 if (rc) { 3582 dev->destroy(dev); 3583 cb_fn(cb_arg, NULL, rc); 3584 return; 3585 } 3586 3587 if (opts.num_md_pages == SPDK_BLOB_OPTS_NUM_MD_PAGES) { 3588 /* By default, allocate 1 page per cluster. 3589 * Technically, this over-allocates metadata 3590 * because more metadata will reduce the number 3591 * of usable clusters. This can be addressed with 3592 * more complex math in the future. 3593 */ 3594 bs->md_len = bs->total_clusters; 3595 } else { 3596 bs->md_len = opts.num_md_pages; 3597 } 3598 rc = spdk_bit_array_resize(&bs->used_md_pages, bs->md_len); 3599 if (rc < 0) { 3600 _spdk_bs_free(bs); 3601 cb_fn(cb_arg, NULL, -ENOMEM); 3602 return; 3603 } 3604 3605 rc = spdk_bit_array_resize(&bs->used_blobids, bs->md_len); 3606 if (rc < 0) { 3607 _spdk_bs_free(bs); 3608 cb_fn(cb_arg, NULL, -ENOMEM); 3609 return; 3610 } 3611 3612 ctx = calloc(1, sizeof(*ctx)); 3613 if (!ctx) { 3614 _spdk_bs_free(bs); 3615 cb_fn(cb_arg, NULL, -ENOMEM); 3616 return; 3617 } 3618 3619 ctx->bs = bs; 3620 3621 /* Allocate memory for the super block */ 3622 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL, 3623 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 3624 if (!ctx->super) { 3625 free(ctx); 3626 _spdk_bs_free(bs); 3627 cb_fn(cb_arg, NULL, -ENOMEM); 3628 return; 3629 } 3630 memcpy(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 3631 sizeof(ctx->super->signature)); 3632 ctx->super->version = SPDK_BS_VERSION; 3633 ctx->super->length = sizeof(*ctx->super); 3634 ctx->super->super_blob = bs->super_blob; 3635 ctx->super->clean = 0; 3636 ctx->super->cluster_size = bs->cluster_sz; 3637 ctx->super->io_unit_size = bs->io_unit_size; 3638 memcpy(&ctx->super->bstype, &bs->bstype, sizeof(bs->bstype)); 3639 3640 /* Calculate how many pages the metadata consumes at the front 3641 * of the disk. 3642 */ 3643 3644 /* The super block uses 1 page */ 3645 num_md_pages = 1; 3646 3647 /* The used_md_pages mask requires 1 bit per metadata page, rounded 3648 * up to the nearest page, plus a header. 3649 */ 3650 ctx->super->used_page_mask_start = num_md_pages; 3651 ctx->super->used_page_mask_len = spdk_divide_round_up(sizeof(struct spdk_bs_md_mask) + 3652 spdk_divide_round_up(bs->md_len, 8), 3653 SPDK_BS_PAGE_SIZE); 3654 num_md_pages += ctx->super->used_page_mask_len; 3655 3656 /* The used_clusters mask requires 1 bit per cluster, rounded 3657 * up to the nearest page, plus a header. 3658 */ 3659 ctx->super->used_cluster_mask_start = num_md_pages; 3660 ctx->super->used_cluster_mask_len = spdk_divide_round_up(sizeof(struct spdk_bs_md_mask) + 3661 spdk_divide_round_up(bs->total_clusters, 8), 3662 SPDK_BS_PAGE_SIZE); 3663 num_md_pages += ctx->super->used_cluster_mask_len; 3664 3665 /* The used_blobids mask requires 1 bit per metadata page, rounded 3666 * up to the nearest page, plus a header. 3667 */ 3668 ctx->super->used_blobid_mask_start = num_md_pages; 3669 ctx->super->used_blobid_mask_len = spdk_divide_round_up(sizeof(struct spdk_bs_md_mask) + 3670 spdk_divide_round_up(bs->md_len, 8), 3671 SPDK_BS_PAGE_SIZE); 3672 num_md_pages += ctx->super->used_blobid_mask_len; 3673 3674 /* The metadata region size was chosen above */ 3675 ctx->super->md_start = bs->md_start = num_md_pages; 3676 ctx->super->md_len = bs->md_len; 3677 num_md_pages += bs->md_len; 3678 3679 num_md_lba = _spdk_bs_page_to_lba(bs, num_md_pages); 3680 3681 ctx->super->size = dev->blockcnt * dev->blocklen; 3682 3683 ctx->super->crc = _spdk_blob_md_page_calc_crc(ctx->super); 3684 3685 num_md_clusters = spdk_divide_round_up(num_md_pages, bs->pages_per_cluster); 3686 if (num_md_clusters > bs->total_clusters) { 3687 SPDK_ERRLOG("Blobstore metadata cannot use more clusters than is available, " 3688 "please decrease number of pages reserved for metadata " 3689 "or increase cluster size.\n"); 3690 spdk_free(ctx->super); 3691 free(ctx); 3692 _spdk_bs_free(bs); 3693 cb_fn(cb_arg, NULL, -ENOMEM); 3694 return; 3695 } 3696 /* Claim all of the clusters used by the metadata */ 3697 for (i = 0; i < num_md_clusters; i++) { 3698 _spdk_bs_claim_cluster(bs, i); 3699 } 3700 3701 bs->total_data_clusters = bs->num_free_clusters; 3702 3703 cpl.type = SPDK_BS_CPL_TYPE_BS_HANDLE; 3704 cpl.u.bs_handle.cb_fn = cb_fn; 3705 cpl.u.bs_handle.cb_arg = cb_arg; 3706 cpl.u.bs_handle.bs = bs; 3707 3708 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3709 if (!seq) { 3710 spdk_free(ctx->super); 3711 free(ctx); 3712 _spdk_bs_free(bs); 3713 cb_fn(cb_arg, NULL, -ENOMEM); 3714 return; 3715 } 3716 3717 batch = spdk_bs_sequence_to_batch(seq, _spdk_bs_init_trim_cpl, ctx); 3718 3719 /* Clear metadata space */ 3720 spdk_bs_batch_write_zeroes_dev(batch, 0, num_md_lba); 3721 3722 if (opts.clear_method == BS_CLEAR_WITH_UNMAP) { 3723 /* Trim data clusters */ 3724 spdk_bs_batch_unmap_dev(batch, num_md_lba, ctx->bs->dev->blockcnt - num_md_lba); 3725 } else if (opts.clear_method == BS_CLEAR_WITH_WRITE_ZEROES) { 3726 /* Write_zeroes to data clusters */ 3727 spdk_bs_batch_write_zeroes_dev(batch, num_md_lba, ctx->bs->dev->blockcnt - num_md_lba); 3728 } 3729 3730 spdk_bs_batch_close(batch); 3731 } 3732 3733 /* END spdk_bs_init */ 3734 3735 /* START spdk_bs_destroy */ 3736 3737 static void 3738 _spdk_bs_destroy_trim_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3739 { 3740 struct spdk_bs_init_ctx *ctx = cb_arg; 3741 struct spdk_blob_store *bs = ctx->bs; 3742 3743 /* 3744 * We need to defer calling spdk_bs_call_cpl() until after 3745 * dev destruction, so tuck these away for later use. 3746 */ 3747 bs->unload_err = bserrno; 3748 memcpy(&bs->unload_cpl, &seq->cpl, sizeof(struct spdk_bs_cpl)); 3749 seq->cpl.type = SPDK_BS_CPL_TYPE_NONE; 3750 3751 spdk_bs_sequence_finish(seq, bserrno); 3752 3753 _spdk_bs_free(bs); 3754 free(ctx); 3755 } 3756 3757 void 3758 spdk_bs_destroy(struct spdk_blob_store *bs, spdk_bs_op_complete cb_fn, 3759 void *cb_arg) 3760 { 3761 struct spdk_bs_cpl cpl; 3762 spdk_bs_sequence_t *seq; 3763 struct spdk_bs_init_ctx *ctx; 3764 3765 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Destroying blobstore\n"); 3766 3767 if (!TAILQ_EMPTY(&bs->blobs)) { 3768 SPDK_ERRLOG("Blobstore still has open blobs\n"); 3769 cb_fn(cb_arg, -EBUSY); 3770 return; 3771 } 3772 3773 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 3774 cpl.u.bs_basic.cb_fn = cb_fn; 3775 cpl.u.bs_basic.cb_arg = cb_arg; 3776 3777 ctx = calloc(1, sizeof(*ctx)); 3778 if (!ctx) { 3779 cb_fn(cb_arg, -ENOMEM); 3780 return; 3781 } 3782 3783 ctx->bs = bs; 3784 3785 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3786 if (!seq) { 3787 free(ctx); 3788 cb_fn(cb_arg, -ENOMEM); 3789 return; 3790 } 3791 3792 /* Write zeroes to the super block */ 3793 spdk_bs_sequence_write_zeroes_dev(seq, 3794 _spdk_bs_page_to_lba(bs, 0), 3795 _spdk_bs_byte_to_lba(bs, sizeof(struct spdk_bs_super_block)), 3796 _spdk_bs_destroy_trim_cpl, ctx); 3797 } 3798 3799 /* END spdk_bs_destroy */ 3800 3801 /* START spdk_bs_unload */ 3802 3803 static void 3804 _spdk_bs_unload_write_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3805 { 3806 struct spdk_bs_load_ctx *ctx = cb_arg; 3807 3808 spdk_free(ctx->super); 3809 3810 /* 3811 * We need to defer calling spdk_bs_call_cpl() until after 3812 * dev destruction, so tuck these away for later use. 3813 */ 3814 ctx->bs->unload_err = bserrno; 3815 memcpy(&ctx->bs->unload_cpl, &seq->cpl, sizeof(struct spdk_bs_cpl)); 3816 seq->cpl.type = SPDK_BS_CPL_TYPE_NONE; 3817 3818 spdk_bs_sequence_finish(seq, bserrno); 3819 3820 _spdk_bs_free(ctx->bs); 3821 free(ctx); 3822 } 3823 3824 static void 3825 _spdk_bs_unload_write_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3826 { 3827 struct spdk_bs_load_ctx *ctx = cb_arg; 3828 3829 spdk_free(ctx->mask); 3830 ctx->super->clean = 1; 3831 3832 _spdk_bs_write_super(seq, ctx->bs, ctx->super, _spdk_bs_unload_write_super_cpl, ctx); 3833 } 3834 3835 static void 3836 _spdk_bs_unload_write_used_blobids_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3837 { 3838 struct spdk_bs_load_ctx *ctx = cb_arg; 3839 3840 spdk_free(ctx->mask); 3841 ctx->mask = NULL; 3842 3843 _spdk_bs_write_used_clusters(seq, cb_arg, _spdk_bs_unload_write_used_clusters_cpl); 3844 } 3845 3846 static void 3847 _spdk_bs_unload_write_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3848 { 3849 struct spdk_bs_load_ctx *ctx = cb_arg; 3850 3851 spdk_free(ctx->mask); 3852 ctx->mask = NULL; 3853 3854 _spdk_bs_write_used_blobids(seq, cb_arg, _spdk_bs_unload_write_used_blobids_cpl); 3855 } 3856 3857 static void 3858 _spdk_bs_unload_read_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3859 { 3860 _spdk_bs_write_used_md(seq, cb_arg, _spdk_bs_unload_write_used_pages_cpl); 3861 } 3862 3863 void 3864 spdk_bs_unload(struct spdk_blob_store *bs, spdk_bs_op_complete cb_fn, void *cb_arg) 3865 { 3866 struct spdk_bs_cpl cpl; 3867 spdk_bs_sequence_t *seq; 3868 struct spdk_bs_load_ctx *ctx; 3869 3870 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Syncing blobstore\n"); 3871 3872 if (!TAILQ_EMPTY(&bs->blobs)) { 3873 SPDK_ERRLOG("Blobstore still has open blobs\n"); 3874 cb_fn(cb_arg, -EBUSY); 3875 return; 3876 } 3877 3878 ctx = calloc(1, sizeof(*ctx)); 3879 if (!ctx) { 3880 cb_fn(cb_arg, -ENOMEM); 3881 return; 3882 } 3883 3884 ctx->bs = bs; 3885 3886 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL, 3887 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 3888 if (!ctx->super) { 3889 free(ctx); 3890 cb_fn(cb_arg, -ENOMEM); 3891 return; 3892 } 3893 3894 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 3895 cpl.u.bs_basic.cb_fn = cb_fn; 3896 cpl.u.bs_basic.cb_arg = cb_arg; 3897 3898 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3899 if (!seq) { 3900 spdk_free(ctx->super); 3901 free(ctx); 3902 cb_fn(cb_arg, -ENOMEM); 3903 return; 3904 } 3905 3906 /* Read super block */ 3907 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 3908 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 3909 _spdk_bs_unload_read_super_cpl, ctx); 3910 } 3911 3912 /* END spdk_bs_unload */ 3913 3914 /* START spdk_bs_set_super */ 3915 3916 struct spdk_bs_set_super_ctx { 3917 struct spdk_blob_store *bs; 3918 struct spdk_bs_super_block *super; 3919 }; 3920 3921 static void 3922 _spdk_bs_set_super_write_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3923 { 3924 struct spdk_bs_set_super_ctx *ctx = cb_arg; 3925 3926 if (bserrno != 0) { 3927 SPDK_ERRLOG("Unable to write to super block of blobstore\n"); 3928 } 3929 3930 spdk_free(ctx->super); 3931 3932 spdk_bs_sequence_finish(seq, bserrno); 3933 3934 free(ctx); 3935 } 3936 3937 static void 3938 _spdk_bs_set_super_read_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3939 { 3940 struct spdk_bs_set_super_ctx *ctx = cb_arg; 3941 3942 if (bserrno != 0) { 3943 SPDK_ERRLOG("Unable to read super block of blobstore\n"); 3944 spdk_free(ctx->super); 3945 spdk_bs_sequence_finish(seq, bserrno); 3946 free(ctx); 3947 return; 3948 } 3949 3950 _spdk_bs_write_super(seq, ctx->bs, ctx->super, _spdk_bs_set_super_write_cpl, ctx); 3951 } 3952 3953 void 3954 spdk_bs_set_super(struct spdk_blob_store *bs, spdk_blob_id blobid, 3955 spdk_bs_op_complete cb_fn, void *cb_arg) 3956 { 3957 struct spdk_bs_cpl cpl; 3958 spdk_bs_sequence_t *seq; 3959 struct spdk_bs_set_super_ctx *ctx; 3960 3961 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Setting super blob id on blobstore\n"); 3962 3963 ctx = calloc(1, sizeof(*ctx)); 3964 if (!ctx) { 3965 cb_fn(cb_arg, -ENOMEM); 3966 return; 3967 } 3968 3969 ctx->bs = bs; 3970 3971 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL, 3972 SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA); 3973 if (!ctx->super) { 3974 free(ctx); 3975 cb_fn(cb_arg, -ENOMEM); 3976 return; 3977 } 3978 3979 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 3980 cpl.u.bs_basic.cb_fn = cb_fn; 3981 cpl.u.bs_basic.cb_arg = cb_arg; 3982 3983 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3984 if (!seq) { 3985 spdk_free(ctx->super); 3986 free(ctx); 3987 cb_fn(cb_arg, -ENOMEM); 3988 return; 3989 } 3990 3991 bs->super_blob = blobid; 3992 3993 /* Read super block */ 3994 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 3995 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 3996 _spdk_bs_set_super_read_cpl, ctx); 3997 } 3998 3999 /* END spdk_bs_set_super */ 4000 4001 void 4002 spdk_bs_get_super(struct spdk_blob_store *bs, 4003 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 4004 { 4005 if (bs->super_blob == SPDK_BLOBID_INVALID) { 4006 cb_fn(cb_arg, SPDK_BLOBID_INVALID, -ENOENT); 4007 } else { 4008 cb_fn(cb_arg, bs->super_blob, 0); 4009 } 4010 } 4011 4012 uint64_t 4013 spdk_bs_get_cluster_size(struct spdk_blob_store *bs) 4014 { 4015 return bs->cluster_sz; 4016 } 4017 4018 uint64_t 4019 spdk_bs_get_page_size(struct spdk_blob_store *bs) 4020 { 4021 return SPDK_BS_PAGE_SIZE; 4022 } 4023 4024 uint64_t 4025 spdk_bs_get_io_unit_size(struct spdk_blob_store *bs) 4026 { 4027 return bs->io_unit_size; 4028 } 4029 4030 uint64_t 4031 spdk_bs_free_cluster_count(struct spdk_blob_store *bs) 4032 { 4033 return bs->num_free_clusters; 4034 } 4035 4036 uint64_t 4037 spdk_bs_total_data_cluster_count(struct spdk_blob_store *bs) 4038 { 4039 return bs->total_data_clusters; 4040 } 4041 4042 static int 4043 spdk_bs_register_md_thread(struct spdk_blob_store *bs) 4044 { 4045 bs->md_channel = spdk_get_io_channel(bs); 4046 if (!bs->md_channel) { 4047 SPDK_ERRLOG("Failed to get IO channel.\n"); 4048 return -1; 4049 } 4050 4051 return 0; 4052 } 4053 4054 static int 4055 spdk_bs_unregister_md_thread(struct spdk_blob_store *bs) 4056 { 4057 spdk_put_io_channel(bs->md_channel); 4058 4059 return 0; 4060 } 4061 4062 spdk_blob_id spdk_blob_get_id(struct spdk_blob *blob) 4063 { 4064 assert(blob != NULL); 4065 4066 return blob->id; 4067 } 4068 4069 uint64_t spdk_blob_get_num_pages(struct spdk_blob *blob) 4070 { 4071 assert(blob != NULL); 4072 4073 return _spdk_bs_cluster_to_page(blob->bs, blob->active.num_clusters); 4074 } 4075 4076 uint64_t spdk_blob_get_num_io_units(struct spdk_blob *blob) 4077 { 4078 assert(blob != NULL); 4079 4080 return spdk_blob_get_num_pages(blob) * _spdk_bs_io_unit_per_page(blob->bs); 4081 } 4082 4083 uint64_t spdk_blob_get_num_clusters(struct spdk_blob *blob) 4084 { 4085 assert(blob != NULL); 4086 4087 return blob->active.num_clusters; 4088 } 4089 4090 /* START spdk_bs_create_blob */ 4091 4092 static void 4093 _spdk_bs_create_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 4094 { 4095 struct spdk_blob *blob = cb_arg; 4096 4097 _spdk_blob_free(blob); 4098 4099 spdk_bs_sequence_finish(seq, bserrno); 4100 } 4101 4102 static int 4103 _spdk_blob_set_xattrs(struct spdk_blob *blob, const struct spdk_blob_xattr_opts *xattrs, 4104 bool internal) 4105 { 4106 uint64_t i; 4107 size_t value_len = 0; 4108 int rc; 4109 const void *value = NULL; 4110 if (xattrs->count > 0 && xattrs->get_value == NULL) { 4111 return -EINVAL; 4112 } 4113 for (i = 0; i < xattrs->count; i++) { 4114 xattrs->get_value(xattrs->ctx, xattrs->names[i], &value, &value_len); 4115 if (value == NULL || value_len == 0) { 4116 return -EINVAL; 4117 } 4118 rc = _spdk_blob_set_xattr(blob, xattrs->names[i], value, value_len, internal); 4119 if (rc < 0) { 4120 return rc; 4121 } 4122 } 4123 return 0; 4124 } 4125 4126 static void 4127 _spdk_blob_set_thin_provision(struct spdk_blob *blob) 4128 { 4129 _spdk_blob_verify_md_op(blob); 4130 blob->invalid_flags |= SPDK_BLOB_THIN_PROV; 4131 blob->state = SPDK_BLOB_STATE_DIRTY; 4132 } 4133 4134 static void 4135 _spdk_bs_create_blob(struct spdk_blob_store *bs, 4136 const struct spdk_blob_opts *opts, 4137 const struct spdk_blob_xattr_opts *internal_xattrs, 4138 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 4139 { 4140 struct spdk_blob *blob; 4141 uint32_t page_idx; 4142 struct spdk_bs_cpl cpl; 4143 struct spdk_blob_opts opts_default; 4144 struct spdk_blob_xattr_opts internal_xattrs_default; 4145 spdk_bs_sequence_t *seq; 4146 spdk_blob_id id; 4147 int rc; 4148 4149 assert(spdk_get_thread() == bs->md_thread); 4150 4151 page_idx = spdk_bit_array_find_first_clear(bs->used_md_pages, 0); 4152 if (page_idx == UINT32_MAX) { 4153 cb_fn(cb_arg, 0, -ENOMEM); 4154 return; 4155 } 4156 spdk_bit_array_set(bs->used_blobids, page_idx); 4157 spdk_bit_array_set(bs->used_md_pages, page_idx); 4158 4159 id = _spdk_bs_page_to_blobid(page_idx); 4160 4161 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Creating blob with id %lu at page %u\n", id, page_idx); 4162 4163 blob = _spdk_blob_alloc(bs, id); 4164 if (!blob) { 4165 cb_fn(cb_arg, 0, -ENOMEM); 4166 return; 4167 } 4168 4169 if (!opts) { 4170 spdk_blob_opts_init(&opts_default); 4171 opts = &opts_default; 4172 } 4173 if (!internal_xattrs) { 4174 _spdk_blob_xattrs_init(&internal_xattrs_default); 4175 internal_xattrs = &internal_xattrs_default; 4176 } 4177 4178 rc = _spdk_blob_set_xattrs(blob, &opts->xattrs, false); 4179 if (rc < 0) { 4180 _spdk_blob_free(blob); 4181 cb_fn(cb_arg, 0, rc); 4182 return; 4183 } 4184 4185 rc = _spdk_blob_set_xattrs(blob, internal_xattrs, true); 4186 if (rc < 0) { 4187 _spdk_blob_free(blob); 4188 cb_fn(cb_arg, 0, rc); 4189 return; 4190 } 4191 4192 if (opts->thin_provision) { 4193 _spdk_blob_set_thin_provision(blob); 4194 } 4195 4196 rc = _spdk_blob_resize(blob, opts->num_clusters); 4197 if (rc < 0) { 4198 _spdk_blob_free(blob); 4199 cb_fn(cb_arg, 0, rc); 4200 return; 4201 } 4202 cpl.type = SPDK_BS_CPL_TYPE_BLOBID; 4203 cpl.u.blobid.cb_fn = cb_fn; 4204 cpl.u.blobid.cb_arg = cb_arg; 4205 cpl.u.blobid.blobid = blob->id; 4206 4207 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 4208 if (!seq) { 4209 _spdk_blob_free(blob); 4210 cb_fn(cb_arg, 0, -ENOMEM); 4211 return; 4212 } 4213 4214 _spdk_blob_persist(seq, blob, _spdk_bs_create_blob_cpl, blob); 4215 } 4216 4217 void spdk_bs_create_blob(struct spdk_blob_store *bs, 4218 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 4219 { 4220 _spdk_bs_create_blob(bs, NULL, NULL, cb_fn, cb_arg); 4221 } 4222 4223 void spdk_bs_create_blob_ext(struct spdk_blob_store *bs, const struct spdk_blob_opts *opts, 4224 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 4225 { 4226 _spdk_bs_create_blob(bs, opts, NULL, cb_fn, cb_arg); 4227 } 4228 4229 /* END spdk_bs_create_blob */ 4230 4231 /* START blob_cleanup */ 4232 4233 struct spdk_clone_snapshot_ctx { 4234 struct spdk_bs_cpl cpl; 4235 int bserrno; 4236 bool frozen; 4237 4238 struct spdk_io_channel *channel; 4239 4240 /* Current cluster for inflate operation */ 4241 uint64_t cluster; 4242 4243 /* For inflation force allocation of all unallocated clusters and remove 4244 * thin-provisioning. Otherwise only decouple parent and keep clone thin. */ 4245 bool allocate_all; 4246 4247 struct { 4248 spdk_blob_id id; 4249 struct spdk_blob *blob; 4250 } original; 4251 struct { 4252 spdk_blob_id id; 4253 struct spdk_blob *blob; 4254 } new; 4255 4256 /* xattrs specified for snapshot/clones only. They have no impact on 4257 * the original blobs xattrs. */ 4258 const struct spdk_blob_xattr_opts *xattrs; 4259 }; 4260 4261 static void 4262 _spdk_bs_clone_snapshot_cleanup_finish(void *cb_arg, int bserrno) 4263 { 4264 struct spdk_clone_snapshot_ctx *ctx = cb_arg; 4265 struct spdk_bs_cpl *cpl = &ctx->cpl; 4266 4267 if (bserrno != 0) { 4268 if (ctx->bserrno != 0) { 4269 SPDK_ERRLOG("Cleanup error %d\n", bserrno); 4270 } else { 4271 ctx->bserrno = bserrno; 4272 } 4273 } 4274 4275 switch (cpl->type) { 4276 case SPDK_BS_CPL_TYPE_BLOBID: 4277 cpl->u.blobid.cb_fn(cpl->u.blobid.cb_arg, cpl->u.blobid.blobid, ctx->bserrno); 4278 break; 4279 case SPDK_BS_CPL_TYPE_BLOB_BASIC: 4280 cpl->u.blob_basic.cb_fn(cpl->u.blob_basic.cb_arg, ctx->bserrno); 4281 break; 4282 default: 4283 SPDK_UNREACHABLE(); 4284 break; 4285 } 4286 4287 free(ctx); 4288 } 4289 4290 static void 4291 _spdk_bs_snapshot_unfreeze_cpl(void *cb_arg, int bserrno) 4292 { 4293 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4294 struct spdk_blob *origblob = ctx->original.blob; 4295 4296 if (bserrno != 0) { 4297 if (ctx->bserrno != 0) { 4298 SPDK_ERRLOG("Unfreeze error %d\n", bserrno); 4299 } else { 4300 ctx->bserrno = bserrno; 4301 } 4302 } 4303 4304 ctx->original.id = origblob->id; 4305 origblob->locked_operation_in_progress = false; 4306 4307 spdk_blob_close(origblob, _spdk_bs_clone_snapshot_cleanup_finish, ctx); 4308 } 4309 4310 static void 4311 _spdk_bs_clone_snapshot_origblob_cleanup(void *cb_arg, int bserrno) 4312 { 4313 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4314 struct spdk_blob *origblob = ctx->original.blob; 4315 4316 if (bserrno != 0) { 4317 if (ctx->bserrno != 0) { 4318 SPDK_ERRLOG("Cleanup error %d\n", bserrno); 4319 } else { 4320 ctx->bserrno = bserrno; 4321 } 4322 } 4323 4324 if (ctx->frozen) { 4325 /* Unfreeze any outstanding I/O */ 4326 _spdk_blob_unfreeze_io(origblob, _spdk_bs_snapshot_unfreeze_cpl, ctx); 4327 } else { 4328 _spdk_bs_snapshot_unfreeze_cpl(ctx, 0); 4329 } 4330 4331 } 4332 4333 static void 4334 _spdk_bs_clone_snapshot_newblob_cleanup(void *cb_arg, int bserrno) 4335 { 4336 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4337 struct spdk_blob *newblob = ctx->new.blob; 4338 4339 if (bserrno != 0) { 4340 if (ctx->bserrno != 0) { 4341 SPDK_ERRLOG("Cleanup error %d\n", bserrno); 4342 } else { 4343 ctx->bserrno = bserrno; 4344 } 4345 } 4346 4347 ctx->new.id = newblob->id; 4348 spdk_blob_close(newblob, _spdk_bs_clone_snapshot_origblob_cleanup, ctx); 4349 } 4350 4351 /* END blob_cleanup */ 4352 4353 /* START spdk_bs_create_snapshot */ 4354 4355 static void 4356 _spdk_bs_snapshot_swap_cluster_maps(struct spdk_blob *blob1, struct spdk_blob *blob2) 4357 { 4358 uint64_t *cluster_temp; 4359 4360 cluster_temp = blob1->active.clusters; 4361 blob1->active.clusters = blob2->active.clusters; 4362 blob2->active.clusters = cluster_temp; 4363 } 4364 4365 static void 4366 _spdk_bs_snapshot_origblob_sync_cpl(void *cb_arg, int bserrno) 4367 { 4368 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4369 struct spdk_blob *origblob = ctx->original.blob; 4370 struct spdk_blob *newblob = ctx->new.blob; 4371 4372 if (bserrno != 0) { 4373 _spdk_bs_snapshot_swap_cluster_maps(newblob, origblob); 4374 _spdk_bs_clone_snapshot_newblob_cleanup(ctx, bserrno); 4375 return; 4376 } 4377 4378 /* Remove metadata descriptor SNAPSHOT_IN_PROGRESS */ 4379 bserrno = _spdk_blob_remove_xattr(newblob, SNAPSHOT_IN_PROGRESS, true); 4380 if (bserrno != 0) { 4381 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4382 return; 4383 } 4384 4385 _spdk_bs_blob_list_add(ctx->original.blob); 4386 4387 spdk_blob_set_read_only(newblob); 4388 4389 /* sync snapshot metadata */ 4390 spdk_blob_sync_md(newblob, _spdk_bs_clone_snapshot_origblob_cleanup, cb_arg); 4391 } 4392 4393 static void 4394 _spdk_bs_snapshot_newblob_sync_cpl(void *cb_arg, int bserrno) 4395 { 4396 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4397 struct spdk_blob *origblob = ctx->original.blob; 4398 struct spdk_blob *newblob = ctx->new.blob; 4399 4400 if (bserrno != 0) { 4401 /* return cluster map back to original */ 4402 _spdk_bs_snapshot_swap_cluster_maps(newblob, origblob); 4403 _spdk_bs_clone_snapshot_newblob_cleanup(ctx, bserrno); 4404 return; 4405 } 4406 4407 /* Set internal xattr for snapshot id */ 4408 bserrno = _spdk_blob_set_xattr(origblob, BLOB_SNAPSHOT, &newblob->id, sizeof(spdk_blob_id), true); 4409 if (bserrno != 0) { 4410 /* return cluster map back to original */ 4411 _spdk_bs_snapshot_swap_cluster_maps(newblob, origblob); 4412 _spdk_bs_clone_snapshot_newblob_cleanup(ctx, bserrno); 4413 return; 4414 } 4415 4416 _spdk_bs_blob_list_remove(origblob); 4417 origblob->parent_id = newblob->id; 4418 4419 /* Create new back_bs_dev for snapshot */ 4420 origblob->back_bs_dev = spdk_bs_create_blob_bs_dev(newblob); 4421 if (origblob->back_bs_dev == NULL) { 4422 /* return cluster map back to original */ 4423 _spdk_bs_snapshot_swap_cluster_maps(newblob, origblob); 4424 _spdk_bs_clone_snapshot_newblob_cleanup(ctx, -EINVAL); 4425 return; 4426 } 4427 4428 /* set clone blob as thin provisioned */ 4429 _spdk_blob_set_thin_provision(origblob); 4430 4431 _spdk_bs_blob_list_add(newblob); 4432 4433 /* sync clone metadata */ 4434 spdk_blob_sync_md(origblob, _spdk_bs_snapshot_origblob_sync_cpl, ctx); 4435 } 4436 4437 static void 4438 _spdk_bs_snapshot_freeze_cpl(void *cb_arg, int rc) 4439 { 4440 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4441 struct spdk_blob *origblob = ctx->original.blob; 4442 struct spdk_blob *newblob = ctx->new.blob; 4443 int bserrno; 4444 4445 if (rc != 0) { 4446 _spdk_bs_clone_snapshot_newblob_cleanup(ctx, rc); 4447 return; 4448 } 4449 4450 ctx->frozen = true; 4451 4452 /* set new back_bs_dev for snapshot */ 4453 newblob->back_bs_dev = origblob->back_bs_dev; 4454 /* Set invalid flags from origblob */ 4455 newblob->invalid_flags = origblob->invalid_flags; 4456 4457 /* inherit parent from original blob if set */ 4458 newblob->parent_id = origblob->parent_id; 4459 if (origblob->parent_id != SPDK_BLOBID_INVALID) { 4460 /* Set internal xattr for snapshot id */ 4461 bserrno = _spdk_blob_set_xattr(newblob, BLOB_SNAPSHOT, 4462 &origblob->parent_id, sizeof(spdk_blob_id), true); 4463 if (bserrno != 0) { 4464 _spdk_bs_clone_snapshot_newblob_cleanup(ctx, bserrno); 4465 return; 4466 } 4467 } 4468 4469 /* swap cluster maps */ 4470 _spdk_bs_snapshot_swap_cluster_maps(newblob, origblob); 4471 4472 /* sync snapshot metadata */ 4473 spdk_blob_sync_md(newblob, _spdk_bs_snapshot_newblob_sync_cpl, ctx); 4474 } 4475 4476 static void 4477 _spdk_bs_snapshot_newblob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 4478 { 4479 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4480 struct spdk_blob *origblob = ctx->original.blob; 4481 struct spdk_blob *newblob = _blob; 4482 4483 if (bserrno != 0) { 4484 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4485 return; 4486 } 4487 4488 ctx->new.blob = newblob; 4489 4490 /* Zero out newblob cluster map */ 4491 memset(newblob->active.clusters, 0, 4492 newblob->active.num_clusters * sizeof(newblob->active.clusters)); 4493 4494 _spdk_blob_freeze_io(origblob, _spdk_bs_snapshot_freeze_cpl, ctx); 4495 } 4496 4497 static void 4498 _spdk_bs_snapshot_newblob_create_cpl(void *cb_arg, spdk_blob_id blobid, int bserrno) 4499 { 4500 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4501 struct spdk_blob *origblob = ctx->original.blob; 4502 4503 if (bserrno != 0) { 4504 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4505 return; 4506 } 4507 4508 ctx->new.id = blobid; 4509 ctx->cpl.u.blobid.blobid = blobid; 4510 4511 spdk_bs_open_blob(origblob->bs, ctx->new.id, _spdk_bs_snapshot_newblob_open_cpl, ctx); 4512 } 4513 4514 4515 static void 4516 _spdk_bs_xattr_snapshot(void *arg, const char *name, 4517 const void **value, size_t *value_len) 4518 { 4519 assert(strncmp(name, SNAPSHOT_IN_PROGRESS, sizeof(SNAPSHOT_IN_PROGRESS)) == 0); 4520 4521 struct spdk_blob *blob = (struct spdk_blob *)arg; 4522 *value = &blob->id; 4523 *value_len = sizeof(blob->id); 4524 } 4525 4526 static void 4527 _spdk_bs_snapshot_origblob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 4528 { 4529 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4530 struct spdk_blob_opts opts; 4531 struct spdk_blob_xattr_opts internal_xattrs; 4532 char *xattrs_names[] = { SNAPSHOT_IN_PROGRESS }; 4533 4534 if (bserrno != 0) { 4535 _spdk_bs_clone_snapshot_cleanup_finish(ctx, bserrno); 4536 return; 4537 } 4538 4539 ctx->original.blob = _blob; 4540 4541 if (_blob->data_ro || _blob->md_ro) { 4542 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Cannot create snapshot from read only blob with id %lu\n", 4543 _blob->id); 4544 ctx->bserrno = -EINVAL; 4545 spdk_blob_close(_blob, _spdk_bs_clone_snapshot_cleanup_finish, ctx); 4546 return; 4547 } 4548 4549 if (_blob->locked_operation_in_progress) { 4550 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Cannot create snapshot - another operation in progress\n"); 4551 ctx->bserrno = -EBUSY; 4552 spdk_blob_close(_blob, _spdk_bs_clone_snapshot_cleanup_finish, ctx); 4553 return; 4554 } 4555 4556 _blob->locked_operation_in_progress = true; 4557 4558 spdk_blob_opts_init(&opts); 4559 _spdk_blob_xattrs_init(&internal_xattrs); 4560 4561 /* Change the size of new blob to the same as in original blob, 4562 * but do not allocate clusters */ 4563 opts.thin_provision = true; 4564 opts.num_clusters = spdk_blob_get_num_clusters(_blob); 4565 4566 /* If there are any xattrs specified for snapshot, set them now */ 4567 if (ctx->xattrs) { 4568 memcpy(&opts.xattrs, ctx->xattrs, sizeof(*ctx->xattrs)); 4569 } 4570 /* Set internal xattr SNAPSHOT_IN_PROGRESS */ 4571 internal_xattrs.count = 1; 4572 internal_xattrs.ctx = _blob; 4573 internal_xattrs.names = xattrs_names; 4574 internal_xattrs.get_value = _spdk_bs_xattr_snapshot; 4575 4576 _spdk_bs_create_blob(_blob->bs, &opts, &internal_xattrs, 4577 _spdk_bs_snapshot_newblob_create_cpl, ctx); 4578 } 4579 4580 void spdk_bs_create_snapshot(struct spdk_blob_store *bs, spdk_blob_id blobid, 4581 const struct spdk_blob_xattr_opts *snapshot_xattrs, 4582 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 4583 { 4584 struct spdk_clone_snapshot_ctx *ctx = calloc(1, sizeof(*ctx)); 4585 4586 if (!ctx) { 4587 cb_fn(cb_arg, SPDK_BLOBID_INVALID, -ENOMEM); 4588 return; 4589 } 4590 ctx->cpl.type = SPDK_BS_CPL_TYPE_BLOBID; 4591 ctx->cpl.u.blobid.cb_fn = cb_fn; 4592 ctx->cpl.u.blobid.cb_arg = cb_arg; 4593 ctx->cpl.u.blobid.blobid = SPDK_BLOBID_INVALID; 4594 ctx->bserrno = 0; 4595 ctx->frozen = false; 4596 ctx->original.id = blobid; 4597 ctx->xattrs = snapshot_xattrs; 4598 4599 spdk_bs_open_blob(bs, ctx->original.id, _spdk_bs_snapshot_origblob_open_cpl, ctx); 4600 } 4601 /* END spdk_bs_create_snapshot */ 4602 4603 /* START spdk_bs_create_clone */ 4604 4605 static void 4606 _spdk_bs_xattr_clone(void *arg, const char *name, 4607 const void **value, size_t *value_len) 4608 { 4609 assert(strncmp(name, BLOB_SNAPSHOT, sizeof(BLOB_SNAPSHOT)) == 0); 4610 4611 struct spdk_blob *blob = (struct spdk_blob *)arg; 4612 *value = &blob->id; 4613 *value_len = sizeof(blob->id); 4614 } 4615 4616 static void 4617 _spdk_bs_clone_newblob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 4618 { 4619 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4620 struct spdk_blob *clone = _blob; 4621 4622 ctx->new.blob = clone; 4623 _spdk_bs_blob_list_add(clone); 4624 4625 spdk_blob_close(clone, _spdk_bs_clone_snapshot_origblob_cleanup, ctx); 4626 } 4627 4628 static void 4629 _spdk_bs_clone_newblob_create_cpl(void *cb_arg, spdk_blob_id blobid, int bserrno) 4630 { 4631 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4632 4633 ctx->cpl.u.blobid.blobid = blobid; 4634 spdk_bs_open_blob(ctx->original.blob->bs, blobid, _spdk_bs_clone_newblob_open_cpl, ctx); 4635 } 4636 4637 static void 4638 _spdk_bs_clone_origblob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 4639 { 4640 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4641 struct spdk_blob_opts opts; 4642 struct spdk_blob_xattr_opts internal_xattrs; 4643 char *xattr_names[] = { BLOB_SNAPSHOT }; 4644 4645 if (bserrno != 0) { 4646 _spdk_bs_clone_snapshot_cleanup_finish(ctx, bserrno); 4647 return; 4648 } 4649 4650 ctx->original.blob = _blob; 4651 4652 if (!_blob->data_ro || !_blob->md_ro) { 4653 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Clone not from read-only blob\n"); 4654 ctx->bserrno = -EINVAL; 4655 spdk_blob_close(_blob, _spdk_bs_clone_snapshot_cleanup_finish, ctx); 4656 return; 4657 } 4658 4659 if (_blob->locked_operation_in_progress) { 4660 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Cannot create clone - another operation in progress\n"); 4661 ctx->bserrno = -EBUSY; 4662 spdk_blob_close(_blob, _spdk_bs_clone_snapshot_cleanup_finish, ctx); 4663 return; 4664 } 4665 4666 _blob->locked_operation_in_progress = true; 4667 4668 spdk_blob_opts_init(&opts); 4669 _spdk_blob_xattrs_init(&internal_xattrs); 4670 4671 opts.thin_provision = true; 4672 opts.num_clusters = spdk_blob_get_num_clusters(_blob); 4673 if (ctx->xattrs) { 4674 memcpy(&opts.xattrs, ctx->xattrs, sizeof(*ctx->xattrs)); 4675 } 4676 4677 /* Set internal xattr BLOB_SNAPSHOT */ 4678 internal_xattrs.count = 1; 4679 internal_xattrs.ctx = _blob; 4680 internal_xattrs.names = xattr_names; 4681 internal_xattrs.get_value = _spdk_bs_xattr_clone; 4682 4683 _spdk_bs_create_blob(_blob->bs, &opts, &internal_xattrs, 4684 _spdk_bs_clone_newblob_create_cpl, ctx); 4685 } 4686 4687 void spdk_bs_create_clone(struct spdk_blob_store *bs, spdk_blob_id blobid, 4688 const struct spdk_blob_xattr_opts *clone_xattrs, 4689 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 4690 { 4691 struct spdk_clone_snapshot_ctx *ctx = calloc(1, sizeof(*ctx)); 4692 4693 if (!ctx) { 4694 cb_fn(cb_arg, SPDK_BLOBID_INVALID, -ENOMEM); 4695 return; 4696 } 4697 4698 ctx->cpl.type = SPDK_BS_CPL_TYPE_BLOBID; 4699 ctx->cpl.u.blobid.cb_fn = cb_fn; 4700 ctx->cpl.u.blobid.cb_arg = cb_arg; 4701 ctx->cpl.u.blobid.blobid = SPDK_BLOBID_INVALID; 4702 ctx->bserrno = 0; 4703 ctx->xattrs = clone_xattrs; 4704 ctx->original.id = blobid; 4705 4706 spdk_bs_open_blob(bs, ctx->original.id, _spdk_bs_clone_origblob_open_cpl, ctx); 4707 } 4708 4709 /* END spdk_bs_create_clone */ 4710 4711 /* START spdk_bs_inflate_blob */ 4712 4713 static void 4714 _spdk_bs_inflate_blob_set_parent_cpl(void *cb_arg, struct spdk_blob *_parent, int bserrno) 4715 { 4716 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4717 struct spdk_blob *_blob = ctx->original.blob; 4718 4719 if (bserrno != 0) { 4720 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4721 return; 4722 } 4723 4724 assert(_parent != NULL); 4725 4726 _spdk_bs_blob_list_remove(_blob); 4727 _blob->parent_id = _parent->id; 4728 _spdk_blob_set_xattr(_blob, BLOB_SNAPSHOT, &_blob->parent_id, 4729 sizeof(spdk_blob_id), true); 4730 4731 _blob->back_bs_dev->destroy(_blob->back_bs_dev); 4732 _blob->back_bs_dev = spdk_bs_create_blob_bs_dev(_parent); 4733 _spdk_bs_blob_list_add(_blob); 4734 4735 spdk_blob_sync_md(_blob, _spdk_bs_clone_snapshot_origblob_cleanup, ctx); 4736 } 4737 4738 static void 4739 _spdk_bs_inflate_blob_done(void *cb_arg, int bserrno) 4740 { 4741 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4742 struct spdk_blob *_blob = ctx->original.blob; 4743 struct spdk_blob *_parent; 4744 4745 if (bserrno != 0) { 4746 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4747 return; 4748 } 4749 4750 if (ctx->allocate_all) { 4751 /* remove thin provisioning */ 4752 _spdk_bs_blob_list_remove(_blob); 4753 _spdk_blob_remove_xattr(_blob, BLOB_SNAPSHOT, true); 4754 _blob->invalid_flags = _blob->invalid_flags & ~SPDK_BLOB_THIN_PROV; 4755 _blob->back_bs_dev->destroy(_blob->back_bs_dev); 4756 _blob->back_bs_dev = NULL; 4757 _blob->parent_id = SPDK_BLOBID_INVALID; 4758 } else { 4759 _parent = ((struct spdk_blob_bs_dev *)(_blob->back_bs_dev))->blob; 4760 if (_parent->parent_id != SPDK_BLOBID_INVALID) { 4761 /* We must change the parent of the inflated blob */ 4762 spdk_bs_open_blob(_blob->bs, _parent->parent_id, 4763 _spdk_bs_inflate_blob_set_parent_cpl, ctx); 4764 return; 4765 } 4766 4767 _spdk_bs_blob_list_remove(_blob); 4768 _spdk_blob_remove_xattr(_blob, BLOB_SNAPSHOT, true); 4769 _blob->parent_id = SPDK_BLOBID_INVALID; 4770 _blob->back_bs_dev->destroy(_blob->back_bs_dev); 4771 _blob->back_bs_dev = spdk_bs_create_zeroes_dev(); 4772 } 4773 4774 _blob->state = SPDK_BLOB_STATE_DIRTY; 4775 spdk_blob_sync_md(_blob, _spdk_bs_clone_snapshot_origblob_cleanup, ctx); 4776 } 4777 4778 /* Check if cluster needs allocation */ 4779 static inline bool 4780 _spdk_bs_cluster_needs_allocation(struct spdk_blob *blob, uint64_t cluster, bool allocate_all) 4781 { 4782 struct spdk_blob_bs_dev *b; 4783 4784 assert(blob != NULL); 4785 4786 if (blob->active.clusters[cluster] != 0) { 4787 /* Cluster is already allocated */ 4788 return false; 4789 } 4790 4791 if (blob->parent_id == SPDK_BLOBID_INVALID) { 4792 /* Blob have no parent blob */ 4793 return allocate_all; 4794 } 4795 4796 b = (struct spdk_blob_bs_dev *)blob->back_bs_dev; 4797 return (allocate_all || b->blob->active.clusters[cluster] != 0); 4798 } 4799 4800 static void 4801 _spdk_bs_inflate_blob_touch_next(void *cb_arg, int bserrno) 4802 { 4803 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4804 struct spdk_blob *_blob = ctx->original.blob; 4805 uint64_t offset; 4806 4807 if (bserrno != 0) { 4808 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, bserrno); 4809 return; 4810 } 4811 4812 for (; ctx->cluster < _blob->active.num_clusters; ctx->cluster++) { 4813 if (_spdk_bs_cluster_needs_allocation(_blob, ctx->cluster, ctx->allocate_all)) { 4814 break; 4815 } 4816 } 4817 4818 if (ctx->cluster < _blob->active.num_clusters) { 4819 offset = _spdk_bs_cluster_to_lba(_blob->bs, ctx->cluster); 4820 4821 /* We may safely increment a cluster before write */ 4822 ctx->cluster++; 4823 4824 /* Use zero length write to touch a cluster */ 4825 spdk_blob_io_write(_blob, ctx->channel, NULL, offset, 0, 4826 _spdk_bs_inflate_blob_touch_next, ctx); 4827 } else { 4828 _spdk_bs_inflate_blob_done(cb_arg, bserrno); 4829 } 4830 } 4831 4832 static void 4833 _spdk_bs_inflate_blob_open_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 4834 { 4835 struct spdk_clone_snapshot_ctx *ctx = (struct spdk_clone_snapshot_ctx *)cb_arg; 4836 uint64_t lfc; /* lowest free cluster */ 4837 uint64_t i; 4838 4839 if (bserrno != 0) { 4840 _spdk_bs_clone_snapshot_cleanup_finish(ctx, bserrno); 4841 return; 4842 } 4843 4844 ctx->original.blob = _blob; 4845 4846 if (_blob->locked_operation_in_progress) { 4847 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Cannot inflate blob - another operation in progress\n"); 4848 ctx->bserrno = -EBUSY; 4849 spdk_blob_close(_blob, _spdk_bs_clone_snapshot_cleanup_finish, ctx); 4850 return; 4851 } 4852 4853 _blob->locked_operation_in_progress = true; 4854 4855 if (!ctx->allocate_all && _blob->parent_id == SPDK_BLOBID_INVALID) { 4856 /* This blob have no parent, so we cannot decouple it. */ 4857 SPDK_ERRLOG("Cannot decouple parent of blob with no parent.\n"); 4858 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, -EINVAL); 4859 return; 4860 } 4861 4862 if (spdk_blob_is_thin_provisioned(_blob) == false) { 4863 /* This is not thin provisioned blob. No need to inflate. */ 4864 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, 0); 4865 return; 4866 } 4867 4868 /* Do two passes - one to verify that we can obtain enough clusters 4869 * and another to actually claim them. 4870 */ 4871 lfc = 0; 4872 for (i = 0; i < _blob->active.num_clusters; i++) { 4873 if (_spdk_bs_cluster_needs_allocation(_blob, i, ctx->allocate_all)) { 4874 lfc = spdk_bit_array_find_first_clear(_blob->bs->used_clusters, lfc); 4875 if (lfc == UINT32_MAX) { 4876 /* No more free clusters. Cannot satisfy the request */ 4877 _spdk_bs_clone_snapshot_origblob_cleanup(ctx, -ENOSPC); 4878 return; 4879 } 4880 lfc++; 4881 } 4882 } 4883 4884 ctx->cluster = 0; 4885 _spdk_bs_inflate_blob_touch_next(ctx, 0); 4886 } 4887 4888 static void 4889 _spdk_bs_inflate_blob(struct spdk_blob_store *bs, struct spdk_io_channel *channel, 4890 spdk_blob_id blobid, bool allocate_all, spdk_blob_op_complete cb_fn, void *cb_arg) 4891 { 4892 struct spdk_clone_snapshot_ctx *ctx = calloc(1, sizeof(*ctx)); 4893 4894 if (!ctx) { 4895 cb_fn(cb_arg, -ENOMEM); 4896 return; 4897 } 4898 ctx->cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 4899 ctx->cpl.u.bs_basic.cb_fn = cb_fn; 4900 ctx->cpl.u.bs_basic.cb_arg = cb_arg; 4901 ctx->bserrno = 0; 4902 ctx->original.id = blobid; 4903 ctx->channel = channel; 4904 ctx->allocate_all = allocate_all; 4905 4906 spdk_bs_open_blob(bs, ctx->original.id, _spdk_bs_inflate_blob_open_cpl, ctx); 4907 } 4908 4909 void 4910 spdk_bs_inflate_blob(struct spdk_blob_store *bs, struct spdk_io_channel *channel, 4911 spdk_blob_id blobid, spdk_blob_op_complete cb_fn, void *cb_arg) 4912 { 4913 _spdk_bs_inflate_blob(bs, channel, blobid, true, cb_fn, cb_arg); 4914 } 4915 4916 void 4917 spdk_bs_blob_decouple_parent(struct spdk_blob_store *bs, struct spdk_io_channel *channel, 4918 spdk_blob_id blobid, spdk_blob_op_complete cb_fn, void *cb_arg) 4919 { 4920 _spdk_bs_inflate_blob(bs, channel, blobid, false, cb_fn, cb_arg); 4921 } 4922 /* END spdk_bs_inflate_blob */ 4923 4924 /* START spdk_blob_resize */ 4925 struct spdk_bs_resize_ctx { 4926 spdk_blob_op_complete cb_fn; 4927 void *cb_arg; 4928 struct spdk_blob *blob; 4929 uint64_t sz; 4930 int rc; 4931 }; 4932 4933 static void 4934 _spdk_bs_resize_unfreeze_cpl(void *cb_arg, int rc) 4935 { 4936 struct spdk_bs_resize_ctx *ctx = (struct spdk_bs_resize_ctx *)cb_arg; 4937 4938 if (rc != 0) { 4939 SPDK_ERRLOG("Unfreeze failed, rc=%d\n", rc); 4940 } 4941 4942 if (ctx->rc != 0) { 4943 SPDK_ERRLOG("Unfreeze failed, ctx->rc=%d\n", ctx->rc); 4944 rc = ctx->rc; 4945 } 4946 4947 ctx->blob->locked_operation_in_progress = false; 4948 4949 ctx->cb_fn(ctx->cb_arg, rc); 4950 free(ctx); 4951 } 4952 4953 static void 4954 _spdk_bs_resize_freeze_cpl(void *cb_arg, int rc) 4955 { 4956 struct spdk_bs_resize_ctx *ctx = (struct spdk_bs_resize_ctx *)cb_arg; 4957 4958 if (rc != 0) { 4959 ctx->blob->locked_operation_in_progress = false; 4960 ctx->cb_fn(ctx->cb_arg, rc); 4961 free(ctx); 4962 return; 4963 } 4964 4965 ctx->rc = _spdk_blob_resize(ctx->blob, ctx->sz); 4966 4967 _spdk_blob_unfreeze_io(ctx->blob, _spdk_bs_resize_unfreeze_cpl, ctx); 4968 } 4969 4970 void 4971 spdk_blob_resize(struct spdk_blob *blob, uint64_t sz, spdk_blob_op_complete cb_fn, void *cb_arg) 4972 { 4973 struct spdk_bs_resize_ctx *ctx; 4974 4975 _spdk_blob_verify_md_op(blob); 4976 4977 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Resizing blob %lu to %lu clusters\n", blob->id, sz); 4978 4979 if (blob->md_ro) { 4980 cb_fn(cb_arg, -EPERM); 4981 return; 4982 } 4983 4984 if (sz == blob->active.num_clusters) { 4985 cb_fn(cb_arg, 0); 4986 return; 4987 } 4988 4989 if (blob->locked_operation_in_progress) { 4990 cb_fn(cb_arg, -EBUSY); 4991 return; 4992 } 4993 4994 ctx = calloc(1, sizeof(*ctx)); 4995 if (!ctx) { 4996 cb_fn(cb_arg, -ENOMEM); 4997 return; 4998 } 4999 5000 blob->locked_operation_in_progress = true; 5001 ctx->cb_fn = cb_fn; 5002 ctx->cb_arg = cb_arg; 5003 ctx->blob = blob; 5004 ctx->sz = sz; 5005 _spdk_blob_freeze_io(blob, _spdk_bs_resize_freeze_cpl, ctx); 5006 } 5007 5008 /* END spdk_blob_resize */ 5009 5010 5011 /* START spdk_bs_delete_blob */ 5012 5013 static void 5014 _spdk_bs_delete_ebusy_close_cpl(void *cb_arg, int bserrno) 5015 { 5016 spdk_bs_sequence_t *seq = cb_arg; 5017 5018 spdk_bs_sequence_finish(seq, -EBUSY); 5019 } 5020 5021 static void 5022 _spdk_bs_delete_close_cpl(void *cb_arg, int bserrno) 5023 { 5024 spdk_bs_sequence_t *seq = cb_arg; 5025 5026 spdk_bs_sequence_finish(seq, bserrno); 5027 } 5028 5029 static void 5030 _spdk_bs_delete_persist_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 5031 { 5032 struct spdk_blob *blob = cb_arg; 5033 5034 if (bserrno != 0) { 5035 /* 5036 * We already removed this blob from the blobstore tailq, so 5037 * we need to free it here since this is the last reference 5038 * to it. 5039 */ 5040 _spdk_blob_free(blob); 5041 _spdk_bs_delete_close_cpl(seq, bserrno); 5042 return; 5043 } 5044 5045 /* 5046 * This will immediately decrement the ref_count and call 5047 * the completion routine since the metadata state is clean. 5048 * By calling spdk_blob_close, we reduce the number of call 5049 * points into code that touches the blob->open_ref count 5050 * and the blobstore's blob list. 5051 */ 5052 spdk_blob_close(blob, _spdk_bs_delete_close_cpl, seq); 5053 } 5054 5055 static void 5056 _spdk_bs_delete_blob_finish(void *cb_arg, struct spdk_blob *blob, int bserrno) 5057 { 5058 spdk_bs_sequence_t *seq = cb_arg; 5059 struct spdk_blob_list *snapshot_entry = NULL; 5060 uint32_t page_num; 5061 5062 if (bserrno) { 5063 SPDK_ERRLOG("Failed to remove blob\n"); 5064 spdk_bs_sequence_finish(seq, bserrno); 5065 return; 5066 } 5067 5068 /* Remove snapshot from the list */ 5069 snapshot_entry = _spdk_bs_get_snapshot_entry(blob->bs, blob->id); 5070 if (snapshot_entry != NULL) { 5071 TAILQ_REMOVE(&blob->bs->snapshots, snapshot_entry, link); 5072 free(snapshot_entry); 5073 } 5074 5075 page_num = _spdk_bs_blobid_to_page(blob->id); 5076 spdk_bit_array_clear(blob->bs->used_blobids, page_num); 5077 blob->state = SPDK_BLOB_STATE_DIRTY; 5078 blob->active.num_pages = 0; 5079 _spdk_blob_resize(blob, 0); 5080 5081 _spdk_blob_persist(seq, blob, _spdk_bs_delete_persist_cpl, blob); 5082 } 5083 5084 static int 5085 _spdk_bs_is_blob_deletable(struct spdk_blob *blob) 5086 { 5087 struct spdk_blob_list *snapshot_entry = NULL; 5088 5089 if (blob->open_ref > 1) { 5090 /* Someone has this blob open (besides this delete context). */ 5091 return -EBUSY; 5092 } 5093 5094 /* Check if this is a snapshot with clones */ 5095 snapshot_entry = _spdk_bs_get_snapshot_entry(blob->bs, blob->id); 5096 if (snapshot_entry != NULL) { 5097 /* If snapshot have clones, we cannot remove it */ 5098 if (!TAILQ_EMPTY(&snapshot_entry->clones)) { 5099 SPDK_ERRLOG("Cannot remove snapshot with clones\n"); 5100 return -EBUSY; 5101 } 5102 } 5103 5104 return 0; 5105 } 5106 5107 static void 5108 _spdk_bs_delete_open_cpl(void *cb_arg, struct spdk_blob *blob, int bserrno) 5109 { 5110 spdk_bs_sequence_t *seq = cb_arg; 5111 5112 if (bserrno != 0) { 5113 spdk_bs_sequence_finish(seq, bserrno); 5114 return; 5115 } 5116 5117 _spdk_blob_verify_md_op(blob); 5118 5119 bserrno = _spdk_bs_is_blob_deletable(blob); 5120 if (bserrno) { 5121 spdk_blob_close(blob, _spdk_bs_delete_ebusy_close_cpl, seq); 5122 return; 5123 } 5124 5125 _spdk_bs_blob_list_remove(blob); 5126 5127 if (blob->locked_operation_in_progress) { 5128 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Cannot remove blob - another operation in progress\n"); 5129 spdk_blob_close(blob, _spdk_bs_delete_ebusy_close_cpl, seq); 5130 return; 5131 } 5132 5133 blob->locked_operation_in_progress = true; 5134 5135 /* 5136 * Remove the blob from the blob_store list now, to ensure it does not 5137 * get returned after this point by _spdk_blob_lookup(). 5138 */ 5139 TAILQ_REMOVE(&blob->bs->blobs, blob, link); 5140 5141 _spdk_bs_delete_blob_finish(seq, blob, 0); 5142 } 5143 5144 void 5145 spdk_bs_delete_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 5146 spdk_blob_op_complete cb_fn, void *cb_arg) 5147 { 5148 struct spdk_bs_cpl cpl; 5149 spdk_bs_sequence_t *seq; 5150 5151 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Deleting blob %lu\n", blobid); 5152 5153 assert(spdk_get_thread() == bs->md_thread); 5154 5155 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 5156 cpl.u.blob_basic.cb_fn = cb_fn; 5157 cpl.u.blob_basic.cb_arg = cb_arg; 5158 5159 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 5160 if (!seq) { 5161 cb_fn(cb_arg, -ENOMEM); 5162 return; 5163 } 5164 5165 spdk_bs_open_blob(bs, blobid, _spdk_bs_delete_open_cpl, seq); 5166 } 5167 5168 /* END spdk_bs_delete_blob */ 5169 5170 /* START spdk_bs_open_blob */ 5171 5172 static void 5173 _spdk_bs_open_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 5174 { 5175 struct spdk_blob *blob = cb_arg; 5176 5177 /* If the blob have crc error, we just return NULL. */ 5178 if (blob == NULL) { 5179 seq->cpl.u.blob_handle.blob = NULL; 5180 spdk_bs_sequence_finish(seq, bserrno); 5181 return; 5182 } 5183 5184 blob->open_ref++; 5185 5186 TAILQ_INSERT_HEAD(&blob->bs->blobs, blob, link); 5187 5188 spdk_bs_sequence_finish(seq, bserrno); 5189 } 5190 5191 static void _spdk_bs_open_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 5192 struct spdk_blob_open_opts *opts, spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 5193 { 5194 struct spdk_blob *blob; 5195 struct spdk_bs_cpl cpl; 5196 struct spdk_blob_open_opts opts_default; 5197 spdk_bs_sequence_t *seq; 5198 uint32_t page_num; 5199 5200 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Opening blob %lu\n", blobid); 5201 assert(spdk_get_thread() == bs->md_thread); 5202 5203 page_num = _spdk_bs_blobid_to_page(blobid); 5204 if (spdk_bit_array_get(bs->used_blobids, page_num) == false) { 5205 /* Invalid blobid */ 5206 cb_fn(cb_arg, NULL, -ENOENT); 5207 return; 5208 } 5209 5210 blob = _spdk_blob_lookup(bs, blobid); 5211 if (blob) { 5212 blob->open_ref++; 5213 cb_fn(cb_arg, blob, 0); 5214 return; 5215 } 5216 5217 blob = _spdk_blob_alloc(bs, blobid); 5218 if (!blob) { 5219 cb_fn(cb_arg, NULL, -ENOMEM); 5220 return; 5221 } 5222 5223 if (!opts) { 5224 spdk_blob_open_opts_init(&opts_default); 5225 opts = &opts_default; 5226 } 5227 5228 blob->clear_method = opts->clear_method; 5229 5230 cpl.type = SPDK_BS_CPL_TYPE_BLOB_HANDLE; 5231 cpl.u.blob_handle.cb_fn = cb_fn; 5232 cpl.u.blob_handle.cb_arg = cb_arg; 5233 cpl.u.blob_handle.blob = blob; 5234 5235 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 5236 if (!seq) { 5237 _spdk_blob_free(blob); 5238 cb_fn(cb_arg, NULL, -ENOMEM); 5239 return; 5240 } 5241 5242 _spdk_blob_load(seq, blob, _spdk_bs_open_blob_cpl, blob); 5243 } 5244 5245 void spdk_bs_open_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 5246 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 5247 { 5248 _spdk_bs_open_blob(bs, blobid, NULL, cb_fn, cb_arg); 5249 } 5250 5251 void spdk_bs_open_blob_ext(struct spdk_blob_store *bs, spdk_blob_id blobid, 5252 struct spdk_blob_open_opts *opts, spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 5253 { 5254 _spdk_bs_open_blob(bs, blobid, opts, cb_fn, cb_arg); 5255 } 5256 5257 /* END spdk_bs_open_blob */ 5258 5259 /* START spdk_blob_set_read_only */ 5260 int spdk_blob_set_read_only(struct spdk_blob *blob) 5261 { 5262 _spdk_blob_verify_md_op(blob); 5263 5264 blob->data_ro_flags |= SPDK_BLOB_READ_ONLY; 5265 5266 blob->state = SPDK_BLOB_STATE_DIRTY; 5267 return 0; 5268 } 5269 /* END spdk_blob_set_read_only */ 5270 5271 /* START spdk_blob_sync_md */ 5272 5273 static void 5274 _spdk_blob_sync_md_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 5275 { 5276 struct spdk_blob *blob = cb_arg; 5277 5278 if (bserrno == 0 && (blob->data_ro_flags & SPDK_BLOB_READ_ONLY)) { 5279 blob->data_ro = true; 5280 blob->md_ro = true; 5281 } 5282 5283 spdk_bs_sequence_finish(seq, bserrno); 5284 } 5285 5286 static void 5287 _spdk_blob_sync_md(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 5288 { 5289 struct spdk_bs_cpl cpl; 5290 spdk_bs_sequence_t *seq; 5291 5292 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 5293 cpl.u.blob_basic.cb_fn = cb_fn; 5294 cpl.u.blob_basic.cb_arg = cb_arg; 5295 5296 seq = spdk_bs_sequence_start(blob->bs->md_channel, &cpl); 5297 if (!seq) { 5298 cb_fn(cb_arg, -ENOMEM); 5299 return; 5300 } 5301 5302 _spdk_blob_persist(seq, blob, _spdk_blob_sync_md_cpl, blob); 5303 } 5304 5305 void 5306 spdk_blob_sync_md(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 5307 { 5308 _spdk_blob_verify_md_op(blob); 5309 5310 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Syncing blob %lu\n", blob->id); 5311 5312 if (blob->md_ro) { 5313 assert(blob->state == SPDK_BLOB_STATE_CLEAN); 5314 cb_fn(cb_arg, 0); 5315 return; 5316 } 5317 5318 _spdk_blob_sync_md(blob, cb_fn, cb_arg); 5319 } 5320 5321 /* END spdk_blob_sync_md */ 5322 5323 struct spdk_blob_insert_cluster_ctx { 5324 struct spdk_thread *thread; 5325 struct spdk_blob *blob; 5326 uint32_t cluster_num; /* cluster index in blob */ 5327 uint32_t cluster; /* cluster on disk */ 5328 int rc; 5329 spdk_blob_op_complete cb_fn; 5330 void *cb_arg; 5331 }; 5332 5333 static void 5334 _spdk_blob_insert_cluster_msg_cpl(void *arg) 5335 { 5336 struct spdk_blob_insert_cluster_ctx *ctx = arg; 5337 5338 ctx->cb_fn(ctx->cb_arg, ctx->rc); 5339 free(ctx); 5340 } 5341 5342 static void 5343 _spdk_blob_insert_cluster_msg_cb(void *arg, int bserrno) 5344 { 5345 struct spdk_blob_insert_cluster_ctx *ctx = arg; 5346 5347 ctx->rc = bserrno; 5348 spdk_thread_send_msg(ctx->thread, _spdk_blob_insert_cluster_msg_cpl, ctx); 5349 } 5350 5351 static void 5352 _spdk_blob_insert_cluster_msg(void *arg) 5353 { 5354 struct spdk_blob_insert_cluster_ctx *ctx = arg; 5355 5356 ctx->rc = _spdk_blob_insert_cluster(ctx->blob, ctx->cluster_num, ctx->cluster); 5357 if (ctx->rc != 0) { 5358 spdk_thread_send_msg(ctx->thread, _spdk_blob_insert_cluster_msg_cpl, ctx); 5359 return; 5360 } 5361 5362 ctx->blob->state = SPDK_BLOB_STATE_DIRTY; 5363 _spdk_blob_sync_md(ctx->blob, _spdk_blob_insert_cluster_msg_cb, ctx); 5364 } 5365 5366 static void 5367 _spdk_blob_insert_cluster_on_md_thread(struct spdk_blob *blob, uint32_t cluster_num, 5368 uint64_t cluster, spdk_blob_op_complete cb_fn, void *cb_arg) 5369 { 5370 struct spdk_blob_insert_cluster_ctx *ctx; 5371 5372 ctx = calloc(1, sizeof(*ctx)); 5373 if (ctx == NULL) { 5374 cb_fn(cb_arg, -ENOMEM); 5375 return; 5376 } 5377 5378 ctx->thread = spdk_get_thread(); 5379 ctx->blob = blob; 5380 ctx->cluster_num = cluster_num; 5381 ctx->cluster = cluster; 5382 ctx->cb_fn = cb_fn; 5383 ctx->cb_arg = cb_arg; 5384 5385 spdk_thread_send_msg(blob->bs->md_thread, _spdk_blob_insert_cluster_msg, ctx); 5386 } 5387 5388 /* START spdk_blob_close */ 5389 5390 static void 5391 _spdk_blob_close_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 5392 { 5393 struct spdk_blob *blob = cb_arg; 5394 5395 if (bserrno == 0) { 5396 blob->open_ref--; 5397 if (blob->open_ref == 0) { 5398 /* 5399 * Blobs with active.num_pages == 0 are deleted blobs. 5400 * these blobs are removed from the blob_store list 5401 * when the deletion process starts - so don't try to 5402 * remove them again. 5403 */ 5404 if (blob->active.num_pages > 0) { 5405 TAILQ_REMOVE(&blob->bs->blobs, blob, link); 5406 } 5407 _spdk_blob_free(blob); 5408 } 5409 } 5410 5411 spdk_bs_sequence_finish(seq, bserrno); 5412 } 5413 5414 void spdk_blob_close(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 5415 { 5416 struct spdk_bs_cpl cpl; 5417 spdk_bs_sequence_t *seq; 5418 5419 _spdk_blob_verify_md_op(blob); 5420 5421 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Closing blob %lu\n", blob->id); 5422 5423 if (blob->open_ref == 0) { 5424 cb_fn(cb_arg, -EBADF); 5425 return; 5426 } 5427 5428 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 5429 cpl.u.blob_basic.cb_fn = cb_fn; 5430 cpl.u.blob_basic.cb_arg = cb_arg; 5431 5432 seq = spdk_bs_sequence_start(blob->bs->md_channel, &cpl); 5433 if (!seq) { 5434 cb_fn(cb_arg, -ENOMEM); 5435 return; 5436 } 5437 5438 /* Sync metadata */ 5439 _spdk_blob_persist(seq, blob, _spdk_blob_close_cpl, blob); 5440 } 5441 5442 /* END spdk_blob_close */ 5443 5444 struct spdk_io_channel *spdk_bs_alloc_io_channel(struct spdk_blob_store *bs) 5445 { 5446 return spdk_get_io_channel(bs); 5447 } 5448 5449 void spdk_bs_free_io_channel(struct spdk_io_channel *channel) 5450 { 5451 spdk_put_io_channel(channel); 5452 } 5453 5454 void spdk_blob_io_unmap(struct spdk_blob *blob, struct spdk_io_channel *channel, 5455 uint64_t offset, uint64_t length, spdk_blob_op_complete cb_fn, void *cb_arg) 5456 { 5457 _spdk_blob_request_submit_op(blob, channel, NULL, offset, length, cb_fn, cb_arg, 5458 SPDK_BLOB_UNMAP); 5459 } 5460 5461 void spdk_blob_io_write_zeroes(struct spdk_blob *blob, struct spdk_io_channel *channel, 5462 uint64_t offset, uint64_t length, spdk_blob_op_complete cb_fn, void *cb_arg) 5463 { 5464 _spdk_blob_request_submit_op(blob, channel, NULL, offset, length, cb_fn, cb_arg, 5465 SPDK_BLOB_WRITE_ZEROES); 5466 } 5467 5468 void spdk_blob_io_write(struct spdk_blob *blob, struct spdk_io_channel *channel, 5469 void *payload, uint64_t offset, uint64_t length, 5470 spdk_blob_op_complete cb_fn, void *cb_arg) 5471 { 5472 _spdk_blob_request_submit_op(blob, channel, payload, offset, length, cb_fn, cb_arg, 5473 SPDK_BLOB_WRITE); 5474 } 5475 5476 void spdk_blob_io_read(struct spdk_blob *blob, struct spdk_io_channel *channel, 5477 void *payload, uint64_t offset, uint64_t length, 5478 spdk_blob_op_complete cb_fn, void *cb_arg) 5479 { 5480 _spdk_blob_request_submit_op(blob, channel, payload, offset, length, cb_fn, cb_arg, 5481 SPDK_BLOB_READ); 5482 } 5483 5484 void spdk_blob_io_writev(struct spdk_blob *blob, struct spdk_io_channel *channel, 5485 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 5486 spdk_blob_op_complete cb_fn, void *cb_arg) 5487 { 5488 _spdk_blob_request_submit_rw_iov(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg, false); 5489 } 5490 5491 void spdk_blob_io_readv(struct spdk_blob *blob, struct spdk_io_channel *channel, 5492 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 5493 spdk_blob_op_complete cb_fn, void *cb_arg) 5494 { 5495 _spdk_blob_request_submit_rw_iov(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg, true); 5496 } 5497 5498 struct spdk_bs_iter_ctx { 5499 int64_t page_num; 5500 struct spdk_blob_store *bs; 5501 5502 spdk_blob_op_with_handle_complete cb_fn; 5503 void *cb_arg; 5504 }; 5505 5506 static void 5507 _spdk_bs_iter_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 5508 { 5509 struct spdk_bs_iter_ctx *ctx = cb_arg; 5510 struct spdk_blob_store *bs = ctx->bs; 5511 spdk_blob_id id; 5512 5513 if (bserrno == 0) { 5514 ctx->cb_fn(ctx->cb_arg, _blob, bserrno); 5515 free(ctx); 5516 return; 5517 } 5518 5519 ctx->page_num++; 5520 ctx->page_num = spdk_bit_array_find_first_set(bs->used_blobids, ctx->page_num); 5521 if (ctx->page_num >= spdk_bit_array_capacity(bs->used_blobids)) { 5522 ctx->cb_fn(ctx->cb_arg, NULL, -ENOENT); 5523 free(ctx); 5524 return; 5525 } 5526 5527 id = _spdk_bs_page_to_blobid(ctx->page_num); 5528 5529 spdk_bs_open_blob(bs, id, _spdk_bs_iter_cpl, ctx); 5530 } 5531 5532 void 5533 spdk_bs_iter_first(struct spdk_blob_store *bs, 5534 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 5535 { 5536 struct spdk_bs_iter_ctx *ctx; 5537 5538 ctx = calloc(1, sizeof(*ctx)); 5539 if (!ctx) { 5540 cb_fn(cb_arg, NULL, -ENOMEM); 5541 return; 5542 } 5543 5544 ctx->page_num = -1; 5545 ctx->bs = bs; 5546 ctx->cb_fn = cb_fn; 5547 ctx->cb_arg = cb_arg; 5548 5549 _spdk_bs_iter_cpl(ctx, NULL, -1); 5550 } 5551 5552 static void 5553 _spdk_bs_iter_close_cpl(void *cb_arg, int bserrno) 5554 { 5555 struct spdk_bs_iter_ctx *ctx = cb_arg; 5556 5557 _spdk_bs_iter_cpl(ctx, NULL, -1); 5558 } 5559 5560 void 5561 spdk_bs_iter_next(struct spdk_blob_store *bs, struct spdk_blob *blob, 5562 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 5563 { 5564 struct spdk_bs_iter_ctx *ctx; 5565 5566 assert(blob != NULL); 5567 5568 ctx = calloc(1, sizeof(*ctx)); 5569 if (!ctx) { 5570 cb_fn(cb_arg, NULL, -ENOMEM); 5571 return; 5572 } 5573 5574 ctx->page_num = _spdk_bs_blobid_to_page(blob->id); 5575 ctx->bs = bs; 5576 ctx->cb_fn = cb_fn; 5577 ctx->cb_arg = cb_arg; 5578 5579 /* Close the existing blob */ 5580 spdk_blob_close(blob, _spdk_bs_iter_close_cpl, ctx); 5581 } 5582 5583 static int 5584 _spdk_blob_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 5585 uint16_t value_len, bool internal) 5586 { 5587 struct spdk_xattr_tailq *xattrs; 5588 struct spdk_xattr *xattr; 5589 5590 _spdk_blob_verify_md_op(blob); 5591 5592 if (blob->md_ro) { 5593 return -EPERM; 5594 } 5595 5596 if (internal) { 5597 xattrs = &blob->xattrs_internal; 5598 blob->invalid_flags |= SPDK_BLOB_INTERNAL_XATTR; 5599 } else { 5600 xattrs = &blob->xattrs; 5601 } 5602 5603 TAILQ_FOREACH(xattr, xattrs, link) { 5604 if (!strcmp(name, xattr->name)) { 5605 free(xattr->value); 5606 xattr->value_len = value_len; 5607 xattr->value = malloc(value_len); 5608 memcpy(xattr->value, value, value_len); 5609 5610 blob->state = SPDK_BLOB_STATE_DIRTY; 5611 5612 return 0; 5613 } 5614 } 5615 5616 xattr = calloc(1, sizeof(*xattr)); 5617 if (!xattr) { 5618 return -ENOMEM; 5619 } 5620 xattr->name = strdup(name); 5621 xattr->value_len = value_len; 5622 xattr->value = malloc(value_len); 5623 memcpy(xattr->value, value, value_len); 5624 TAILQ_INSERT_TAIL(xattrs, xattr, link); 5625 5626 blob->state = SPDK_BLOB_STATE_DIRTY; 5627 5628 return 0; 5629 } 5630 5631 int 5632 spdk_blob_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 5633 uint16_t value_len) 5634 { 5635 return _spdk_blob_set_xattr(blob, name, value, value_len, false); 5636 } 5637 5638 static int 5639 _spdk_blob_remove_xattr(struct spdk_blob *blob, const char *name, bool internal) 5640 { 5641 struct spdk_xattr_tailq *xattrs; 5642 struct spdk_xattr *xattr; 5643 5644 _spdk_blob_verify_md_op(blob); 5645 5646 if (blob->md_ro) { 5647 return -EPERM; 5648 } 5649 xattrs = internal ? &blob->xattrs_internal : &blob->xattrs; 5650 5651 TAILQ_FOREACH(xattr, xattrs, link) { 5652 if (!strcmp(name, xattr->name)) { 5653 TAILQ_REMOVE(xattrs, xattr, link); 5654 free(xattr->value); 5655 free(xattr->name); 5656 free(xattr); 5657 5658 if (internal && TAILQ_EMPTY(&blob->xattrs_internal)) { 5659 blob->invalid_flags &= ~SPDK_BLOB_INTERNAL_XATTR; 5660 } 5661 blob->state = SPDK_BLOB_STATE_DIRTY; 5662 5663 return 0; 5664 } 5665 } 5666 5667 return -ENOENT; 5668 } 5669 5670 int 5671 spdk_blob_remove_xattr(struct spdk_blob *blob, const char *name) 5672 { 5673 return _spdk_blob_remove_xattr(blob, name, false); 5674 } 5675 5676 static int 5677 _spdk_blob_get_xattr_value(struct spdk_blob *blob, const char *name, 5678 const void **value, size_t *value_len, bool internal) 5679 { 5680 struct spdk_xattr *xattr; 5681 struct spdk_xattr_tailq *xattrs; 5682 5683 xattrs = internal ? &blob->xattrs_internal : &blob->xattrs; 5684 5685 TAILQ_FOREACH(xattr, xattrs, link) { 5686 if (!strcmp(name, xattr->name)) { 5687 *value = xattr->value; 5688 *value_len = xattr->value_len; 5689 return 0; 5690 } 5691 } 5692 return -ENOENT; 5693 } 5694 5695 int 5696 spdk_blob_get_xattr_value(struct spdk_blob *blob, const char *name, 5697 const void **value, size_t *value_len) 5698 { 5699 _spdk_blob_verify_md_op(blob); 5700 5701 return _spdk_blob_get_xattr_value(blob, name, value, value_len, false); 5702 } 5703 5704 struct spdk_xattr_names { 5705 uint32_t count; 5706 const char *names[0]; 5707 }; 5708 5709 static int 5710 _spdk_blob_get_xattr_names(struct spdk_xattr_tailq *xattrs, struct spdk_xattr_names **names) 5711 { 5712 struct spdk_xattr *xattr; 5713 int count = 0; 5714 5715 TAILQ_FOREACH(xattr, xattrs, link) { 5716 count++; 5717 } 5718 5719 *names = calloc(1, sizeof(struct spdk_xattr_names) + count * sizeof(char *)); 5720 if (*names == NULL) { 5721 return -ENOMEM; 5722 } 5723 5724 TAILQ_FOREACH(xattr, xattrs, link) { 5725 (*names)->names[(*names)->count++] = xattr->name; 5726 } 5727 5728 return 0; 5729 } 5730 5731 int 5732 spdk_blob_get_xattr_names(struct spdk_blob *blob, struct spdk_xattr_names **names) 5733 { 5734 _spdk_blob_verify_md_op(blob); 5735 5736 return _spdk_blob_get_xattr_names(&blob->xattrs, names); 5737 } 5738 5739 uint32_t 5740 spdk_xattr_names_get_count(struct spdk_xattr_names *names) 5741 { 5742 assert(names != NULL); 5743 5744 return names->count; 5745 } 5746 5747 const char * 5748 spdk_xattr_names_get_name(struct spdk_xattr_names *names, uint32_t index) 5749 { 5750 if (index >= names->count) { 5751 return NULL; 5752 } 5753 5754 return names->names[index]; 5755 } 5756 5757 void 5758 spdk_xattr_names_free(struct spdk_xattr_names *names) 5759 { 5760 free(names); 5761 } 5762 5763 struct spdk_bs_type 5764 spdk_bs_get_bstype(struct spdk_blob_store *bs) 5765 { 5766 return bs->bstype; 5767 } 5768 5769 void 5770 spdk_bs_set_bstype(struct spdk_blob_store *bs, struct spdk_bs_type bstype) 5771 { 5772 memcpy(&bs->bstype, &bstype, sizeof(bstype)); 5773 } 5774 5775 bool 5776 spdk_blob_is_read_only(struct spdk_blob *blob) 5777 { 5778 assert(blob != NULL); 5779 return (blob->data_ro || blob->md_ro); 5780 } 5781 5782 bool 5783 spdk_blob_is_snapshot(struct spdk_blob *blob) 5784 { 5785 struct spdk_blob_list *snapshot_entry; 5786 5787 assert(blob != NULL); 5788 5789 snapshot_entry = _spdk_bs_get_snapshot_entry(blob->bs, blob->id); 5790 if (snapshot_entry == NULL) { 5791 return false; 5792 } 5793 5794 return true; 5795 } 5796 5797 bool 5798 spdk_blob_is_clone(struct spdk_blob *blob) 5799 { 5800 assert(blob != NULL); 5801 5802 if (blob->parent_id != SPDK_BLOBID_INVALID) { 5803 assert(spdk_blob_is_thin_provisioned(blob)); 5804 return true; 5805 } 5806 5807 return false; 5808 } 5809 5810 bool 5811 spdk_blob_is_thin_provisioned(struct spdk_blob *blob) 5812 { 5813 assert(blob != NULL); 5814 return !!(blob->invalid_flags & SPDK_BLOB_THIN_PROV); 5815 } 5816 5817 spdk_blob_id 5818 spdk_blob_get_parent_snapshot(struct spdk_blob_store *bs, spdk_blob_id blob_id) 5819 { 5820 struct spdk_blob_list *snapshot_entry = NULL; 5821 struct spdk_blob_list *clone_entry = NULL; 5822 5823 TAILQ_FOREACH(snapshot_entry, &bs->snapshots, link) { 5824 TAILQ_FOREACH(clone_entry, &snapshot_entry->clones, link) { 5825 if (clone_entry->id == blob_id) { 5826 return snapshot_entry->id; 5827 } 5828 } 5829 } 5830 5831 return SPDK_BLOBID_INVALID; 5832 } 5833 5834 int 5835 spdk_blob_get_clones(struct spdk_blob_store *bs, spdk_blob_id blobid, spdk_blob_id *ids, 5836 size_t *count) 5837 { 5838 struct spdk_blob_list *snapshot_entry, *clone_entry; 5839 size_t n; 5840 5841 snapshot_entry = _spdk_bs_get_snapshot_entry(bs, blobid); 5842 if (snapshot_entry == NULL) { 5843 *count = 0; 5844 return 0; 5845 } 5846 5847 if (ids == NULL || *count < snapshot_entry->clone_count) { 5848 *count = snapshot_entry->clone_count; 5849 return -ENOMEM; 5850 } 5851 *count = snapshot_entry->clone_count; 5852 5853 n = 0; 5854 TAILQ_FOREACH(clone_entry, &snapshot_entry->clones, link) { 5855 ids[n++] = clone_entry->id; 5856 } 5857 5858 return 0; 5859 } 5860 5861 SPDK_LOG_REGISTER_COMPONENT("blob", SPDK_LOG_BLOB) 5862