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/io_channel.h" 41 #include "spdk/bit_array.h" 42 #include "spdk/likely.h" 43 44 #include "spdk_internal/log.h" 45 46 #include "blobstore.h" 47 48 #define BLOB_CRC32C_INITIAL 0xffffffffUL 49 50 static int spdk_bs_register_md_thread(struct spdk_blob_store *bs); 51 static int spdk_bs_unregister_md_thread(struct spdk_blob_store *bs); 52 static void _spdk_blob_close_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno); 53 void _spdk_blob_insert_cluster_on_md_thread(struct spdk_blob *blob, uint32_t cluster_num, 54 uint64_t cluster, spdk_blob_op_complete cb_fn, void *cb_arg); 55 56 static int _spdk_blob_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 57 uint16_t value_len, bool internal); 58 static int _spdk_blob_get_xattr_value(struct spdk_blob *blob, const char *name, 59 const void **value, size_t *value_len, bool internal); 60 static int _spdk_blob_remove_xattr(struct spdk_blob *blob, const char *name, bool internal); 61 62 static void 63 _spdk_blob_verify_md_op(struct spdk_blob *blob) 64 { 65 assert(blob != NULL); 66 assert(spdk_get_thread() == blob->bs->md_thread); 67 assert(blob->state != SPDK_BLOB_STATE_LOADING); 68 } 69 70 static inline size_t 71 divide_round_up(size_t num, size_t divisor) 72 { 73 return (num + divisor - 1) / divisor; 74 } 75 76 static void 77 _spdk_bs_claim_cluster(struct spdk_blob_store *bs, uint32_t cluster_num) 78 { 79 assert(cluster_num < spdk_bit_array_capacity(bs->used_clusters)); 80 assert(spdk_bit_array_get(bs->used_clusters, cluster_num) == false); 81 assert(bs->num_free_clusters > 0); 82 83 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Claiming cluster %u\n", cluster_num); 84 85 spdk_bit_array_set(bs->used_clusters, cluster_num); 86 bs->num_free_clusters--; 87 } 88 89 static int 90 _spdk_blob_insert_cluster(struct spdk_blob *blob, uint32_t cluster_num, uint64_t cluster) 91 { 92 uint64_t *cluster_lba = &blob->active.clusters[cluster_num]; 93 94 _spdk_blob_verify_md_op(blob); 95 96 if (*cluster_lba != 0) { 97 return -EEXIST; 98 } 99 100 *cluster_lba = _spdk_bs_cluster_to_lba(blob->bs, cluster); 101 return 0; 102 } 103 104 static int 105 _spdk_bs_allocate_cluster(struct spdk_blob *blob, uint32_t cluster_num, 106 uint64_t *lowest_free_cluster, bool update_map) 107 { 108 pthread_mutex_lock(&blob->bs->used_clusters_mutex); 109 *lowest_free_cluster = spdk_bit_array_find_first_clear(blob->bs->used_clusters, 110 *lowest_free_cluster); 111 if (*lowest_free_cluster >= blob->bs->total_clusters) { 112 /* No more free clusters. Cannot satisfy the request */ 113 pthread_mutex_unlock(&blob->bs->used_clusters_mutex); 114 return -ENOSPC; 115 } 116 117 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Claiming cluster %lu for blob %lu\n", *lowest_free_cluster, blob->id); 118 _spdk_bs_claim_cluster(blob->bs, *lowest_free_cluster); 119 pthread_mutex_unlock(&blob->bs->used_clusters_mutex); 120 121 if (update_map) { 122 _spdk_blob_insert_cluster(blob, cluster_num, *lowest_free_cluster); 123 } 124 125 return 0; 126 } 127 128 static void 129 _spdk_bs_release_cluster(struct spdk_blob_store *bs, uint32_t cluster_num) 130 { 131 assert(cluster_num < spdk_bit_array_capacity(bs->used_clusters)); 132 assert(spdk_bit_array_get(bs->used_clusters, cluster_num) == true); 133 assert(bs->num_free_clusters < bs->total_clusters); 134 135 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Releasing cluster %u\n", cluster_num); 136 137 pthread_mutex_lock(&bs->used_clusters_mutex); 138 spdk_bit_array_clear(bs->used_clusters, cluster_num); 139 bs->num_free_clusters++; 140 pthread_mutex_unlock(&bs->used_clusters_mutex); 141 } 142 143 void 144 spdk_blob_opts_init(struct spdk_blob_opts *opts) 145 { 146 opts->num_clusters = 0; 147 opts->thin_provision = false; 148 opts->xattrs.count = 0; 149 opts->xattrs.names = NULL; 150 opts->xattrs.ctx = NULL; 151 opts->xattrs.get_value = NULL; 152 } 153 154 static struct spdk_blob * 155 _spdk_blob_alloc(struct spdk_blob_store *bs, spdk_blob_id id) 156 { 157 struct spdk_blob *blob; 158 159 blob = calloc(1, sizeof(*blob)); 160 if (!blob) { 161 return NULL; 162 } 163 164 blob->id = id; 165 blob->bs = bs; 166 167 blob->state = SPDK_BLOB_STATE_DIRTY; 168 blob->active.num_pages = 1; 169 blob->active.pages = calloc(1, sizeof(*blob->active.pages)); 170 if (!blob->active.pages) { 171 free(blob); 172 return NULL; 173 } 174 175 blob->active.pages[0] = _spdk_bs_blobid_to_page(id); 176 177 TAILQ_INIT(&blob->xattrs); 178 TAILQ_INIT(&blob->xattrs_internal); 179 180 return blob; 181 } 182 183 static void 184 _spdk_xattrs_free(struct spdk_xattr_tailq *xattrs) 185 { 186 struct spdk_xattr *xattr, *xattr_tmp; 187 188 TAILQ_FOREACH_SAFE(xattr, xattrs, link, xattr_tmp) { 189 TAILQ_REMOVE(xattrs, xattr, link); 190 free(xattr->name); 191 free(xattr->value); 192 free(xattr); 193 } 194 } 195 196 static void 197 _spdk_blob_free(struct spdk_blob *blob) 198 { 199 assert(blob != NULL); 200 201 free(blob->active.clusters); 202 free(blob->clean.clusters); 203 free(blob->active.pages); 204 free(blob->clean.pages); 205 206 _spdk_xattrs_free(&blob->xattrs); 207 _spdk_xattrs_free(&blob->xattrs_internal); 208 209 free(blob); 210 } 211 212 static int 213 _spdk_blob_mark_clean(struct spdk_blob *blob) 214 { 215 uint64_t *clusters = NULL; 216 uint32_t *pages = NULL; 217 218 assert(blob != NULL); 219 220 if (blob->active.num_clusters) { 221 assert(blob->active.clusters); 222 clusters = calloc(blob->active.num_clusters, sizeof(*blob->active.clusters)); 223 if (!clusters) { 224 return -1; 225 } 226 memcpy(clusters, blob->active.clusters, blob->active.num_clusters * sizeof(*clusters)); 227 } 228 229 if (blob->active.num_pages) { 230 assert(blob->active.pages); 231 pages = calloc(blob->active.num_pages, sizeof(*blob->active.pages)); 232 if (!pages) { 233 free(clusters); 234 return -1; 235 } 236 memcpy(pages, blob->active.pages, blob->active.num_pages * sizeof(*pages)); 237 } 238 239 free(blob->clean.clusters); 240 free(blob->clean.pages); 241 242 blob->clean.num_clusters = blob->active.num_clusters; 243 blob->clean.clusters = blob->active.clusters; 244 blob->clean.num_pages = blob->active.num_pages; 245 blob->clean.pages = blob->active.pages; 246 247 blob->active.clusters = clusters; 248 blob->active.pages = pages; 249 250 /* If the metadata was dirtied again while the metadata was being written to disk, 251 * we do not want to revert the DIRTY state back to CLEAN here. 252 */ 253 if (blob->state == SPDK_BLOB_STATE_LOADING) { 254 blob->state = SPDK_BLOB_STATE_CLEAN; 255 } 256 257 return 0; 258 } 259 260 static int 261 _spdk_blob_deserialize_xattr(struct spdk_blob *blob, 262 struct spdk_blob_md_descriptor_xattr *desc_xattr, bool internal) 263 { 264 struct spdk_xattr *xattr; 265 266 if (desc_xattr->length != sizeof(desc_xattr->name_length) + 267 sizeof(desc_xattr->value_length) + 268 desc_xattr->name_length + desc_xattr->value_length) { 269 return -EINVAL; 270 } 271 272 xattr = calloc(1, sizeof(*xattr)); 273 if (xattr == NULL) { 274 return -ENOMEM; 275 } 276 277 xattr->name = malloc(desc_xattr->name_length + 1); 278 if (xattr->name == NULL) { 279 free(xattr); 280 return -ENOMEM; 281 } 282 strncpy(xattr->name, desc_xattr->name, desc_xattr->name_length); 283 xattr->name[desc_xattr->name_length] = '\0'; 284 285 xattr->value = malloc(desc_xattr->value_length); 286 if (xattr->value == NULL) { 287 free(xattr->name); 288 free(xattr); 289 return -ENOMEM; 290 } 291 xattr->value_len = desc_xattr->value_length; 292 memcpy(xattr->value, 293 (void *)((uintptr_t)desc_xattr->name + desc_xattr->name_length), 294 desc_xattr->value_length); 295 296 TAILQ_INSERT_TAIL(internal ? &blob->xattrs_internal : &blob->xattrs, xattr, link); 297 298 return 0; 299 } 300 301 302 static int 303 _spdk_blob_parse_page(const struct spdk_blob_md_page *page, struct spdk_blob *blob) 304 { 305 struct spdk_blob_md_descriptor *desc; 306 size_t cur_desc = 0; 307 void *tmp; 308 309 desc = (struct spdk_blob_md_descriptor *)page->descriptors; 310 while (cur_desc < sizeof(page->descriptors)) { 311 if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_PADDING) { 312 if (desc->length == 0) { 313 /* If padding and length are 0, this terminates the page */ 314 break; 315 } 316 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_FLAGS) { 317 struct spdk_blob_md_descriptor_flags *desc_flags; 318 319 desc_flags = (struct spdk_blob_md_descriptor_flags *)desc; 320 321 if (desc_flags->length != sizeof(*desc_flags) - sizeof(*desc)) { 322 return -EINVAL; 323 } 324 325 if ((desc_flags->invalid_flags | SPDK_BLOB_INVALID_FLAGS_MASK) != 326 SPDK_BLOB_INVALID_FLAGS_MASK) { 327 return -EINVAL; 328 } 329 330 if ((desc_flags->data_ro_flags | SPDK_BLOB_DATA_RO_FLAGS_MASK) != 331 SPDK_BLOB_DATA_RO_FLAGS_MASK) { 332 blob->data_ro = true; 333 blob->md_ro = true; 334 } 335 336 if ((desc_flags->md_ro_flags | SPDK_BLOB_MD_RO_FLAGS_MASK) != 337 SPDK_BLOB_MD_RO_FLAGS_MASK) { 338 blob->md_ro = true; 339 } 340 341 if ((desc_flags->data_ro_flags & SPDK_BLOB_READ_ONLY)) { 342 blob->data_ro = true; 343 blob->md_ro = true; 344 } 345 346 blob->invalid_flags = desc_flags->invalid_flags; 347 blob->data_ro_flags = desc_flags->data_ro_flags; 348 blob->md_ro_flags = desc_flags->md_ro_flags; 349 350 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_EXTENT) { 351 struct spdk_blob_md_descriptor_extent *desc_extent; 352 unsigned int i, j; 353 unsigned int cluster_count = blob->active.num_clusters; 354 355 desc_extent = (struct spdk_blob_md_descriptor_extent *)desc; 356 357 if (desc_extent->length == 0 || 358 (desc_extent->length % sizeof(desc_extent->extents[0]) != 0)) { 359 return -EINVAL; 360 } 361 362 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 363 for (j = 0; j < desc_extent->extents[i].length; j++) { 364 if (!spdk_bit_array_get(blob->bs->used_clusters, 365 desc_extent->extents[i].cluster_idx + j)) { 366 return -EINVAL; 367 } 368 cluster_count++; 369 } 370 } 371 372 if (cluster_count == 0) { 373 return -EINVAL; 374 } 375 tmp = realloc(blob->active.clusters, cluster_count * sizeof(uint64_t)); 376 if (tmp == NULL) { 377 return -ENOMEM; 378 } 379 blob->active.clusters = tmp; 380 blob->active.cluster_array_size = cluster_count; 381 382 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 383 for (j = 0; j < desc_extent->extents[i].length; j++) { 384 if (desc_extent->extents[i].cluster_idx != 0) { 385 blob->active.clusters[blob->active.num_clusters++] = _spdk_bs_cluster_to_lba(blob->bs, 386 desc_extent->extents[i].cluster_idx + j); 387 } else if (spdk_blob_is_thin_provisioned(blob)) { 388 blob->active.clusters[blob->active.num_clusters++] = 0; 389 } else { 390 return -EINVAL; 391 } 392 } 393 } 394 395 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR) { 396 int rc; 397 398 rc = _spdk_blob_deserialize_xattr(blob, 399 (struct spdk_blob_md_descriptor_xattr *) desc, false); 400 if (rc != 0) { 401 return rc; 402 } 403 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR_INTERNAL) { 404 int rc; 405 406 rc = _spdk_blob_deserialize_xattr(blob, 407 (struct spdk_blob_md_descriptor_xattr *) desc, true); 408 if (rc != 0) { 409 return rc; 410 } 411 } else { 412 /* Unrecognized descriptor type. Do not fail - just continue to the 413 * next descriptor. If this descriptor is associated with some feature 414 * defined in a newer version of blobstore, that version of blobstore 415 * should create and set an associated feature flag to specify if this 416 * blob can be loaded or not. 417 */ 418 } 419 420 /* Advance to the next descriptor */ 421 cur_desc += sizeof(*desc) + desc->length; 422 if (cur_desc + sizeof(*desc) > sizeof(page->descriptors)) { 423 break; 424 } 425 desc = (struct spdk_blob_md_descriptor *)((uintptr_t)page->descriptors + cur_desc); 426 } 427 428 return 0; 429 } 430 431 static int 432 _spdk_blob_parse(const struct spdk_blob_md_page *pages, uint32_t page_count, 433 struct spdk_blob *blob) 434 { 435 const struct spdk_blob_md_page *page; 436 uint32_t i; 437 int rc; 438 439 assert(page_count > 0); 440 assert(pages[0].sequence_num == 0); 441 assert(blob != NULL); 442 assert(blob->state == SPDK_BLOB_STATE_LOADING); 443 assert(blob->active.clusters == NULL); 444 445 /* The blobid provided doesn't match what's in the MD, this can 446 * happen for example if a bogus blobid is passed in through open. 447 */ 448 if (blob->id != pages[0].id) { 449 SPDK_ERRLOG("Blobid (%lu) doesn't match what's in metadata (%lu)\n", 450 blob->id, pages[0].id); 451 return -ENOENT; 452 } 453 454 for (i = 0; i < page_count; i++) { 455 page = &pages[i]; 456 457 assert(page->id == blob->id); 458 assert(page->sequence_num == i); 459 460 rc = _spdk_blob_parse_page(page, blob); 461 if (rc != 0) { 462 return rc; 463 } 464 } 465 466 return 0; 467 } 468 469 static int 470 _spdk_blob_serialize_add_page(const struct spdk_blob *blob, 471 struct spdk_blob_md_page **pages, 472 uint32_t *page_count, 473 struct spdk_blob_md_page **last_page) 474 { 475 struct spdk_blob_md_page *page; 476 477 assert(pages != NULL); 478 assert(page_count != NULL); 479 480 if (*page_count == 0) { 481 assert(*pages == NULL); 482 *page_count = 1; 483 *pages = spdk_dma_malloc(SPDK_BS_PAGE_SIZE, 484 SPDK_BS_PAGE_SIZE, 485 NULL); 486 } else { 487 assert(*pages != NULL); 488 (*page_count)++; 489 *pages = spdk_dma_realloc(*pages, 490 SPDK_BS_PAGE_SIZE * (*page_count), 491 SPDK_BS_PAGE_SIZE, 492 NULL); 493 } 494 495 if (*pages == NULL) { 496 *page_count = 0; 497 *last_page = NULL; 498 return -ENOMEM; 499 } 500 501 page = &(*pages)[*page_count - 1]; 502 memset(page, 0, sizeof(*page)); 503 page->id = blob->id; 504 page->sequence_num = *page_count - 1; 505 page->next = SPDK_INVALID_MD_PAGE; 506 *last_page = page; 507 508 return 0; 509 } 510 511 /* Transform the in-memory representation 'xattr' into an on-disk xattr descriptor. 512 * Update required_sz on both success and failure. 513 * 514 */ 515 static int 516 _spdk_blob_serialize_xattr(const struct spdk_xattr *xattr, 517 uint8_t *buf, size_t buf_sz, 518 size_t *required_sz, bool internal) 519 { 520 struct spdk_blob_md_descriptor_xattr *desc; 521 522 *required_sz = sizeof(struct spdk_blob_md_descriptor_xattr) + 523 strlen(xattr->name) + 524 xattr->value_len; 525 526 if (buf_sz < *required_sz) { 527 return -1; 528 } 529 530 desc = (struct spdk_blob_md_descriptor_xattr *)buf; 531 532 desc->type = internal ? SPDK_MD_DESCRIPTOR_TYPE_XATTR_INTERNAL : SPDK_MD_DESCRIPTOR_TYPE_XATTR; 533 desc->length = sizeof(desc->name_length) + 534 sizeof(desc->value_length) + 535 strlen(xattr->name) + 536 xattr->value_len; 537 desc->name_length = strlen(xattr->name); 538 desc->value_length = xattr->value_len; 539 540 memcpy(desc->name, xattr->name, desc->name_length); 541 memcpy((void *)((uintptr_t)desc->name + desc->name_length), 542 xattr->value, 543 desc->value_length); 544 545 return 0; 546 } 547 548 static void 549 _spdk_blob_serialize_extent(const struct spdk_blob *blob, 550 uint64_t start_cluster, uint64_t *next_cluster, 551 uint8_t *buf, size_t buf_sz) 552 { 553 struct spdk_blob_md_descriptor_extent *desc; 554 size_t cur_sz; 555 uint64_t i, extent_idx; 556 uint32_t lba, lba_per_cluster, lba_count; 557 558 /* The buffer must have room for at least one extent */ 559 cur_sz = sizeof(struct spdk_blob_md_descriptor) + sizeof(desc->extents[0]); 560 if (buf_sz < cur_sz) { 561 *next_cluster = start_cluster; 562 return; 563 } 564 565 desc = (struct spdk_blob_md_descriptor_extent *)buf; 566 desc->type = SPDK_MD_DESCRIPTOR_TYPE_EXTENT; 567 568 lba_per_cluster = _spdk_bs_cluster_to_lba(blob->bs, 1); 569 570 lba = blob->active.clusters[start_cluster]; 571 lba_count = lba_per_cluster; 572 extent_idx = 0; 573 for (i = start_cluster + 1; i < blob->active.num_clusters; i++) { 574 if ((lba + lba_count) == blob->active.clusters[i]) { 575 lba_count += lba_per_cluster; 576 continue; 577 } 578 desc->extents[extent_idx].cluster_idx = lba / lba_per_cluster; 579 desc->extents[extent_idx].length = lba_count / lba_per_cluster; 580 extent_idx++; 581 582 cur_sz += sizeof(desc->extents[extent_idx]); 583 584 if (buf_sz < cur_sz) { 585 /* If we ran out of buffer space, return */ 586 desc->length = sizeof(desc->extents[0]) * extent_idx; 587 *next_cluster = i; 588 return; 589 } 590 591 lba = blob->active.clusters[i]; 592 lba_count = lba_per_cluster; 593 } 594 595 desc->extents[extent_idx].cluster_idx = lba / lba_per_cluster; 596 desc->extents[extent_idx].length = lba_count / lba_per_cluster; 597 extent_idx++; 598 599 desc->length = sizeof(desc->extents[0]) * extent_idx; 600 *next_cluster = blob->active.num_clusters; 601 602 return; 603 } 604 605 static void 606 _spdk_blob_serialize_flags(const struct spdk_blob *blob, 607 uint8_t *buf, size_t *buf_sz) 608 { 609 struct spdk_blob_md_descriptor_flags *desc; 610 611 /* 612 * Flags get serialized first, so we should always have room for the flags 613 * descriptor. 614 */ 615 assert(*buf_sz >= sizeof(*desc)); 616 617 desc = (struct spdk_blob_md_descriptor_flags *)buf; 618 desc->type = SPDK_MD_DESCRIPTOR_TYPE_FLAGS; 619 desc->length = sizeof(*desc) - sizeof(struct spdk_blob_md_descriptor); 620 desc->invalid_flags = blob->invalid_flags; 621 desc->data_ro_flags = blob->data_ro_flags; 622 desc->md_ro_flags = blob->md_ro_flags; 623 624 *buf_sz -= sizeof(*desc); 625 } 626 627 static int 628 _spdk_blob_serialize_xattrs(const struct spdk_blob *blob, 629 const struct spdk_xattr_tailq *xattrs, bool internal, 630 struct spdk_blob_md_page **pages, 631 struct spdk_blob_md_page *cur_page, 632 uint32_t *page_count, uint8_t **buf, 633 size_t *remaining_sz) 634 { 635 const struct spdk_xattr *xattr; 636 int rc; 637 638 TAILQ_FOREACH(xattr, xattrs, link) { 639 size_t required_sz = 0; 640 641 rc = _spdk_blob_serialize_xattr(xattr, 642 *buf, *remaining_sz, 643 &required_sz, internal); 644 if (rc < 0) { 645 /* Need to add a new page to the chain */ 646 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, 647 &cur_page); 648 if (rc < 0) { 649 spdk_dma_free(*pages); 650 *pages = NULL; 651 *page_count = 0; 652 return rc; 653 } 654 655 *buf = (uint8_t *)cur_page->descriptors; 656 *remaining_sz = sizeof(cur_page->descriptors); 657 658 /* Try again */ 659 required_sz = 0; 660 rc = _spdk_blob_serialize_xattr(xattr, 661 *buf, *remaining_sz, 662 &required_sz, internal); 663 664 if (rc < 0) { 665 spdk_dma_free(*pages); 666 *pages = NULL; 667 *page_count = 0; 668 return -1; 669 } 670 } 671 672 *remaining_sz -= required_sz; 673 *buf += required_sz; 674 } 675 676 return 0; 677 } 678 679 static int 680 _spdk_blob_serialize(const struct spdk_blob *blob, struct spdk_blob_md_page **pages, 681 uint32_t *page_count) 682 { 683 struct spdk_blob_md_page *cur_page; 684 int rc; 685 uint8_t *buf; 686 size_t remaining_sz; 687 uint64_t last_cluster; 688 689 assert(pages != NULL); 690 assert(page_count != NULL); 691 assert(blob != NULL); 692 assert(blob->state == SPDK_BLOB_STATE_DIRTY); 693 694 *pages = NULL; 695 *page_count = 0; 696 697 /* A blob always has at least 1 page, even if it has no descriptors */ 698 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, &cur_page); 699 if (rc < 0) { 700 return rc; 701 } 702 703 buf = (uint8_t *)cur_page->descriptors; 704 remaining_sz = sizeof(cur_page->descriptors); 705 706 /* Serialize flags */ 707 _spdk_blob_serialize_flags(blob, buf, &remaining_sz); 708 buf += sizeof(struct spdk_blob_md_descriptor_flags); 709 710 /* Serialize xattrs */ 711 rc = _spdk_blob_serialize_xattrs(blob, &blob->xattrs, false, 712 pages, cur_page, page_count, &buf, &remaining_sz); 713 if (rc < 0) { 714 return rc; 715 } 716 717 /* Serialize internal xattrs */ 718 rc = _spdk_blob_serialize_xattrs(blob, &blob->xattrs_internal, true, 719 pages, cur_page, page_count, &buf, &remaining_sz); 720 if (rc < 0) { 721 return rc; 722 } 723 724 /* Serialize extents */ 725 last_cluster = 0; 726 while (last_cluster < blob->active.num_clusters) { 727 _spdk_blob_serialize_extent(blob, last_cluster, &last_cluster, 728 buf, remaining_sz); 729 730 if (last_cluster == blob->active.num_clusters) { 731 break; 732 } 733 734 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, 735 &cur_page); 736 if (rc < 0) { 737 return rc; 738 } 739 740 buf = (uint8_t *)cur_page->descriptors; 741 remaining_sz = sizeof(cur_page->descriptors); 742 } 743 744 return 0; 745 } 746 747 struct spdk_blob_load_ctx { 748 struct spdk_blob *blob; 749 750 struct spdk_blob_md_page *pages; 751 uint32_t num_pages; 752 753 spdk_bs_sequence_cpl cb_fn; 754 void *cb_arg; 755 }; 756 757 static uint32_t 758 _spdk_blob_md_page_calc_crc(void *page) 759 { 760 uint32_t crc; 761 762 crc = BLOB_CRC32C_INITIAL; 763 crc = spdk_crc32c_update(page, SPDK_BS_PAGE_SIZE - 4, crc); 764 crc ^= BLOB_CRC32C_INITIAL; 765 766 return crc; 767 768 } 769 770 static void 771 _spdk_blob_load_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 772 { 773 struct spdk_blob_load_ctx *ctx = cb_arg; 774 struct spdk_blob *blob = ctx->blob; 775 struct spdk_blob_md_page *page; 776 int rc; 777 uint32_t crc; 778 779 page = &ctx->pages[ctx->num_pages - 1]; 780 crc = _spdk_blob_md_page_calc_crc(page); 781 if (crc != page->crc) { 782 SPDK_ERRLOG("Metadata page %d crc mismatch\n", ctx->num_pages); 783 _spdk_blob_free(blob); 784 ctx->cb_fn(seq, NULL, -EINVAL); 785 spdk_dma_free(ctx->pages); 786 free(ctx); 787 return; 788 } 789 790 if (page->next != SPDK_INVALID_MD_PAGE) { 791 uint32_t next_page = page->next; 792 uint64_t next_lba = _spdk_bs_page_to_lba(blob->bs, blob->bs->md_start + next_page); 793 794 795 assert(next_lba < (blob->bs->md_start + blob->bs->md_len)); 796 797 /* Read the next page */ 798 ctx->num_pages++; 799 ctx->pages = spdk_dma_realloc(ctx->pages, (sizeof(*page) * ctx->num_pages), 800 sizeof(*page), NULL); 801 if (ctx->pages == NULL) { 802 ctx->cb_fn(seq, ctx->cb_arg, -ENOMEM); 803 free(ctx); 804 return; 805 } 806 807 spdk_bs_sequence_read_dev(seq, &ctx->pages[ctx->num_pages - 1], 808 next_lba, 809 _spdk_bs_byte_to_lba(blob->bs, sizeof(*page)), 810 _spdk_blob_load_cpl, ctx); 811 return; 812 } 813 814 /* Parse the pages */ 815 rc = _spdk_blob_parse(ctx->pages, ctx->num_pages, blob); 816 if (rc) { 817 _spdk_blob_free(blob); 818 ctx->cb_fn(seq, NULL, rc); 819 spdk_dma_free(ctx->pages); 820 free(ctx); 821 return; 822 } 823 824 if (spdk_blob_is_thin_provisioned(blob) == true) { 825 blob->back_bs_dev = spdk_bs_create_zeroes_dev(); 826 } 827 828 _spdk_blob_mark_clean(blob); 829 830 ctx->cb_fn(seq, ctx->cb_arg, rc); 831 832 /* Free the memory */ 833 spdk_dma_free(ctx->pages); 834 free(ctx); 835 } 836 837 /* Load a blob from disk given a blobid */ 838 static void 839 _spdk_blob_load(spdk_bs_sequence_t *seq, struct spdk_blob *blob, 840 spdk_bs_sequence_cpl cb_fn, void *cb_arg) 841 { 842 struct spdk_blob_load_ctx *ctx; 843 struct spdk_blob_store *bs; 844 uint32_t page_num; 845 uint64_t lba; 846 847 _spdk_blob_verify_md_op(blob); 848 849 bs = blob->bs; 850 851 ctx = calloc(1, sizeof(*ctx)); 852 if (!ctx) { 853 cb_fn(seq, cb_arg, -ENOMEM); 854 return; 855 } 856 857 ctx->blob = blob; 858 ctx->pages = spdk_dma_realloc(ctx->pages, SPDK_BS_PAGE_SIZE, 859 SPDK_BS_PAGE_SIZE, NULL); 860 if (!ctx->pages) { 861 free(ctx); 862 cb_fn(seq, cb_arg, -ENOMEM); 863 return; 864 } 865 ctx->num_pages = 1; 866 ctx->cb_fn = cb_fn; 867 ctx->cb_arg = cb_arg; 868 869 page_num = _spdk_bs_blobid_to_page(blob->id); 870 lba = _spdk_bs_page_to_lba(blob->bs, bs->md_start + page_num); 871 872 blob->state = SPDK_BLOB_STATE_LOADING; 873 874 spdk_bs_sequence_read_dev(seq, &ctx->pages[0], lba, 875 _spdk_bs_byte_to_lba(bs, SPDK_BS_PAGE_SIZE), 876 _spdk_blob_load_cpl, ctx); 877 } 878 879 struct spdk_blob_persist_ctx { 880 struct spdk_blob *blob; 881 882 struct spdk_blob_md_page *pages; 883 884 uint64_t idx; 885 886 spdk_bs_sequence_t *seq; 887 spdk_bs_sequence_cpl cb_fn; 888 void *cb_arg; 889 }; 890 891 static void 892 _spdk_blob_persist_complete(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 893 { 894 struct spdk_blob_persist_ctx *ctx = cb_arg; 895 struct spdk_blob *blob = ctx->blob; 896 897 if (bserrno == 0) { 898 _spdk_blob_mark_clean(blob); 899 } 900 901 /* Call user callback */ 902 ctx->cb_fn(seq, ctx->cb_arg, bserrno); 903 904 /* Free the memory */ 905 spdk_dma_free(ctx->pages); 906 free(ctx); 907 } 908 909 static void 910 _spdk_blob_persist_unmap_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 911 { 912 struct spdk_blob_persist_ctx *ctx = cb_arg; 913 struct spdk_blob *blob = ctx->blob; 914 struct spdk_blob_store *bs = blob->bs; 915 void *tmp; 916 size_t i; 917 918 /* Release all clusters that were truncated */ 919 for (i = blob->active.num_clusters; i < blob->active.cluster_array_size; i++) { 920 uint32_t cluster_num = _spdk_bs_lba_to_cluster(bs, blob->active.clusters[i]); 921 922 /* Nothing to release if it was not allocated */ 923 if (blob->active.clusters[i] != 0) { 924 _spdk_bs_release_cluster(bs, cluster_num); 925 } 926 } 927 928 if (blob->active.num_clusters == 0) { 929 free(blob->active.clusters); 930 blob->active.clusters = NULL; 931 blob->active.cluster_array_size = 0; 932 } else { 933 tmp = realloc(blob->active.clusters, sizeof(uint64_t) * blob->active.num_clusters); 934 assert(tmp != NULL); 935 blob->active.clusters = tmp; 936 blob->active.cluster_array_size = blob->active.num_clusters; 937 } 938 939 _spdk_blob_persist_complete(seq, ctx, bserrno); 940 } 941 942 static void 943 _spdk_blob_persist_unmap_clusters(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 944 { 945 struct spdk_blob_persist_ctx *ctx = cb_arg; 946 struct spdk_blob *blob = ctx->blob; 947 struct spdk_blob_store *bs = blob->bs; 948 spdk_bs_batch_t *batch; 949 size_t i; 950 uint64_t lba; 951 uint32_t lba_count; 952 953 /* Clusters don't move around in blobs. The list shrinks or grows 954 * at the end, but no changes ever occur in the middle of the list. 955 */ 956 957 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_unmap_clusters_cpl, ctx); 958 959 /* Unmap all clusters that were truncated */ 960 lba = 0; 961 lba_count = 0; 962 for (i = blob->active.num_clusters; i < blob->active.cluster_array_size; i++) { 963 uint64_t next_lba = blob->active.clusters[i]; 964 uint32_t next_lba_count = _spdk_bs_cluster_to_lba(bs, 1); 965 966 if (next_lba > 0 && (lba + lba_count) == next_lba) { 967 /* This cluster is contiguous with the previous one. */ 968 lba_count += next_lba_count; 969 continue; 970 } 971 972 /* This cluster is not contiguous with the previous one. */ 973 974 /* If a run of LBAs previously existing, send them 975 * as an unmap. 976 */ 977 if (lba_count > 0) { 978 spdk_bs_batch_unmap_dev(batch, lba, lba_count); 979 } 980 981 /* Start building the next batch */ 982 lba = next_lba; 983 if (next_lba > 0) { 984 lba_count = next_lba_count; 985 } else { 986 lba_count = 0; 987 } 988 } 989 990 /* If we ended with a contiguous set of LBAs, send the unmap now */ 991 if (lba_count > 0) { 992 spdk_bs_batch_unmap_dev(batch, lba, lba_count); 993 } 994 995 spdk_bs_batch_close(batch); 996 } 997 998 static void 999 _spdk_blob_persist_zero_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1000 { 1001 struct spdk_blob_persist_ctx *ctx = cb_arg; 1002 struct spdk_blob *blob = ctx->blob; 1003 struct spdk_blob_store *bs = blob->bs; 1004 size_t i; 1005 1006 /* This loop starts at 1 because the first page is special and handled 1007 * below. The pages (except the first) are never written in place, 1008 * so any pages in the clean list must be zeroed. 1009 */ 1010 for (i = 1; i < blob->clean.num_pages; i++) { 1011 spdk_bit_array_clear(bs->used_md_pages, blob->clean.pages[i]); 1012 } 1013 1014 if (blob->active.num_pages == 0) { 1015 uint32_t page_num; 1016 1017 page_num = _spdk_bs_blobid_to_page(blob->id); 1018 spdk_bit_array_clear(bs->used_md_pages, page_num); 1019 } 1020 1021 /* Move on to unmapping clusters */ 1022 _spdk_blob_persist_unmap_clusters(seq, ctx, 0); 1023 } 1024 1025 static void 1026 _spdk_blob_persist_zero_pages(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1027 { 1028 struct spdk_blob_persist_ctx *ctx = cb_arg; 1029 struct spdk_blob *blob = ctx->blob; 1030 struct spdk_blob_store *bs = blob->bs; 1031 uint64_t lba; 1032 uint32_t lba_count; 1033 spdk_bs_batch_t *batch; 1034 size_t i; 1035 1036 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_zero_pages_cpl, ctx); 1037 1038 lba_count = _spdk_bs_byte_to_lba(bs, SPDK_BS_PAGE_SIZE); 1039 1040 /* This loop starts at 1 because the first page is special and handled 1041 * below. The pages (except the first) are never written in place, 1042 * so any pages in the clean list must be zeroed. 1043 */ 1044 for (i = 1; i < blob->clean.num_pages; i++) { 1045 lba = _spdk_bs_page_to_lba(bs, bs->md_start + blob->clean.pages[i]); 1046 1047 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1048 } 1049 1050 /* The first page will only be zeroed if this is a delete. */ 1051 if (blob->active.num_pages == 0) { 1052 uint32_t page_num; 1053 1054 /* The first page in the metadata goes where the blobid indicates */ 1055 page_num = _spdk_bs_blobid_to_page(blob->id); 1056 lba = _spdk_bs_page_to_lba(bs, bs->md_start + page_num); 1057 1058 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1059 } 1060 1061 spdk_bs_batch_close(batch); 1062 } 1063 1064 static void 1065 _spdk_blob_persist_write_page_root(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1066 { 1067 struct spdk_blob_persist_ctx *ctx = cb_arg; 1068 struct spdk_blob *blob = ctx->blob; 1069 struct spdk_blob_store *bs = blob->bs; 1070 uint64_t lba; 1071 uint32_t lba_count; 1072 struct spdk_blob_md_page *page; 1073 1074 if (blob->active.num_pages == 0) { 1075 /* Move on to the next step */ 1076 _spdk_blob_persist_zero_pages(seq, ctx, 0); 1077 return; 1078 } 1079 1080 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(*page)); 1081 1082 page = &ctx->pages[0]; 1083 /* The first page in the metadata goes where the blobid indicates */ 1084 lba = _spdk_bs_page_to_lba(bs, bs->md_start + _spdk_bs_blobid_to_page(blob->id)); 1085 1086 spdk_bs_sequence_write_dev(seq, page, lba, lba_count, 1087 _spdk_blob_persist_zero_pages, ctx); 1088 } 1089 1090 static void 1091 _spdk_blob_persist_write_page_chain(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1092 { 1093 struct spdk_blob_persist_ctx *ctx = cb_arg; 1094 struct spdk_blob *blob = ctx->blob; 1095 struct spdk_blob_store *bs = blob->bs; 1096 uint64_t lba; 1097 uint32_t lba_count; 1098 struct spdk_blob_md_page *page; 1099 spdk_bs_batch_t *batch; 1100 size_t i; 1101 1102 /* Clusters don't move around in blobs. The list shrinks or grows 1103 * at the end, but no changes ever occur in the middle of the list. 1104 */ 1105 1106 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(*page)); 1107 1108 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_write_page_root, ctx); 1109 1110 /* This starts at 1. The root page is not written until 1111 * all of the others are finished 1112 */ 1113 for (i = 1; i < blob->active.num_pages; i++) { 1114 page = &ctx->pages[i]; 1115 assert(page->sequence_num == i); 1116 1117 lba = _spdk_bs_page_to_lba(bs, bs->md_start + blob->active.pages[i]); 1118 1119 spdk_bs_batch_write_dev(batch, page, lba, lba_count); 1120 } 1121 1122 spdk_bs_batch_close(batch); 1123 } 1124 1125 static int 1126 _spdk_resize_blob(struct spdk_blob *blob, uint64_t sz) 1127 { 1128 uint64_t i; 1129 uint64_t *tmp; 1130 uint64_t lfc; /* lowest free cluster */ 1131 uint64_t num_clusters; 1132 struct spdk_blob_store *bs; 1133 1134 bs = blob->bs; 1135 1136 _spdk_blob_verify_md_op(blob); 1137 1138 if (blob->active.num_clusters == sz) { 1139 return 0; 1140 } 1141 1142 if (blob->active.num_clusters < blob->active.cluster_array_size) { 1143 /* If this blob was resized to be larger, then smaller, then 1144 * larger without syncing, then the cluster array already 1145 * contains spare assigned clusters we can use. 1146 */ 1147 num_clusters = spdk_min(blob->active.cluster_array_size, 1148 sz); 1149 } else { 1150 num_clusters = blob->active.num_clusters; 1151 } 1152 1153 /* Do two passes - one to verify that we can obtain enough clusters 1154 * and another to actually claim them. 1155 */ 1156 1157 if (spdk_blob_is_thin_provisioned(blob) == false) { 1158 lfc = 0; 1159 for (i = num_clusters; i < sz; i++) { 1160 lfc = spdk_bit_array_find_first_clear(bs->used_clusters, lfc); 1161 if (lfc >= bs->total_clusters) { 1162 /* No more free clusters. Cannot satisfy the request */ 1163 return -ENOSPC; 1164 } 1165 lfc++; 1166 } 1167 } 1168 1169 if (sz > num_clusters) { 1170 /* Expand the cluster array if necessary. 1171 * We only shrink the array when persisting. 1172 */ 1173 tmp = realloc(blob->active.clusters, sizeof(uint64_t) * sz); 1174 if (sz > 0 && tmp == NULL) { 1175 return -ENOMEM; 1176 } 1177 memset(tmp + blob->active.cluster_array_size, 0, 1178 sizeof(uint64_t) * (sz - blob->active.cluster_array_size)); 1179 blob->active.clusters = tmp; 1180 blob->active.cluster_array_size = sz; 1181 } 1182 1183 blob->state = SPDK_BLOB_STATE_DIRTY; 1184 1185 if (spdk_blob_is_thin_provisioned(blob) == false) { 1186 lfc = 0; 1187 for (i = num_clusters; i < sz; i++) { 1188 _spdk_bs_allocate_cluster(blob, i, &lfc, true); 1189 lfc++; 1190 } 1191 } 1192 1193 blob->active.num_clusters = sz; 1194 1195 return 0; 1196 } 1197 1198 static void 1199 _spdk_blob_persist_start(struct spdk_blob_persist_ctx *ctx) 1200 { 1201 spdk_bs_sequence_t *seq = ctx->seq; 1202 struct spdk_blob *blob = ctx->blob; 1203 struct spdk_blob_store *bs = blob->bs; 1204 uint64_t i; 1205 uint32_t page_num; 1206 int rc; 1207 1208 if (blob->active.num_pages == 0) { 1209 /* This is the signal that the blob should be deleted. 1210 * Immediately jump to the clean up routine. */ 1211 assert(blob->clean.num_pages > 0); 1212 ctx->idx = blob->clean.num_pages - 1; 1213 blob->state = SPDK_BLOB_STATE_CLEAN; 1214 _spdk_blob_persist_zero_pages(seq, ctx, 0); 1215 return; 1216 1217 } 1218 1219 /* Generate the new metadata */ 1220 rc = _spdk_blob_serialize(blob, &ctx->pages, &blob->active.num_pages); 1221 if (rc < 0) { 1222 _spdk_blob_persist_complete(seq, ctx, rc); 1223 return; 1224 } 1225 1226 assert(blob->active.num_pages >= 1); 1227 1228 /* Resize the cache of page indices */ 1229 blob->active.pages = realloc(blob->active.pages, 1230 blob->active.num_pages * sizeof(*blob->active.pages)); 1231 if (!blob->active.pages) { 1232 _spdk_blob_persist_complete(seq, ctx, -ENOMEM); 1233 return; 1234 } 1235 1236 /* Assign this metadata to pages. This requires two passes - 1237 * one to verify that there are enough pages and a second 1238 * to actually claim them. */ 1239 page_num = 0; 1240 /* Note that this loop starts at one. The first page location is fixed by the blobid. */ 1241 for (i = 1; i < blob->active.num_pages; i++) { 1242 page_num = spdk_bit_array_find_first_clear(bs->used_md_pages, page_num); 1243 if (page_num >= spdk_bit_array_capacity(bs->used_md_pages)) { 1244 _spdk_blob_persist_complete(seq, ctx, -ENOMEM); 1245 return; 1246 } 1247 page_num++; 1248 } 1249 1250 page_num = 0; 1251 blob->active.pages[0] = _spdk_bs_blobid_to_page(blob->id); 1252 for (i = 1; i < blob->active.num_pages; i++) { 1253 page_num = spdk_bit_array_find_first_clear(bs->used_md_pages, page_num); 1254 ctx->pages[i - 1].next = page_num; 1255 /* Now that previous metadata page is complete, calculate the crc for it. */ 1256 ctx->pages[i - 1].crc = _spdk_blob_md_page_calc_crc(&ctx->pages[i - 1]); 1257 blob->active.pages[i] = page_num; 1258 spdk_bit_array_set(bs->used_md_pages, page_num); 1259 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Claiming page %u for blob %lu\n", page_num, blob->id); 1260 page_num++; 1261 } 1262 ctx->pages[i - 1].crc = _spdk_blob_md_page_calc_crc(&ctx->pages[i - 1]); 1263 /* Start writing the metadata from last page to first */ 1264 ctx->idx = blob->active.num_pages - 1; 1265 blob->state = SPDK_BLOB_STATE_CLEAN; 1266 _spdk_blob_persist_write_page_chain(seq, ctx, 0); 1267 } 1268 1269 /* Write a blob to disk */ 1270 static void 1271 _spdk_blob_persist(spdk_bs_sequence_t *seq, struct spdk_blob *blob, 1272 spdk_bs_sequence_cpl cb_fn, void *cb_arg) 1273 { 1274 struct spdk_blob_persist_ctx *ctx; 1275 1276 _spdk_blob_verify_md_op(blob); 1277 1278 if (blob->state == SPDK_BLOB_STATE_CLEAN) { 1279 cb_fn(seq, cb_arg, 0); 1280 return; 1281 } 1282 1283 ctx = calloc(1, sizeof(*ctx)); 1284 if (!ctx) { 1285 cb_fn(seq, cb_arg, -ENOMEM); 1286 return; 1287 } 1288 ctx->blob = blob; 1289 ctx->seq = seq; 1290 ctx->cb_fn = cb_fn; 1291 ctx->cb_arg = cb_arg; 1292 1293 _spdk_blob_persist_start(ctx); 1294 } 1295 1296 struct spdk_blob_copy_cluster_ctx { 1297 struct spdk_blob *blob; 1298 uint8_t *buf; 1299 uint64_t page; 1300 uint64_t new_cluster; 1301 spdk_bs_sequence_t *seq; 1302 }; 1303 1304 static void 1305 _spdk_blob_allocate_and_copy_cluster_cpl(void *cb_arg, int bserrno) 1306 { 1307 struct spdk_blob_copy_cluster_ctx *ctx = cb_arg; 1308 struct spdk_bs_request_set *set = (struct spdk_bs_request_set *)ctx->seq; 1309 TAILQ_HEAD(, spdk_bs_request_set) requests; 1310 spdk_bs_user_op_t *op; 1311 1312 TAILQ_INIT(&requests); 1313 TAILQ_SWAP(&set->channel->need_cluster_alloc, &requests, spdk_bs_request_set, link); 1314 1315 while (!TAILQ_EMPTY(&requests)) { 1316 op = TAILQ_FIRST(&requests); 1317 TAILQ_REMOVE(&requests, op, link); 1318 if (bserrno == 0) { 1319 spdk_bs_user_op_execute(op); 1320 } else { 1321 spdk_bs_user_op_abort(op); 1322 } 1323 } 1324 1325 spdk_dma_free(ctx->buf); 1326 free(ctx); 1327 } 1328 1329 static void 1330 _spdk_blob_insert_cluster_cpl(void *cb_arg, int bserrno) 1331 { 1332 struct spdk_blob_copy_cluster_ctx *ctx = cb_arg; 1333 1334 if (bserrno) { 1335 uint32_t cluster_number; 1336 1337 if (bserrno == -EEXIST) { 1338 /* The metadata insert failed because another thread 1339 * allocated the cluster first. Free our cluster 1340 * but continue without error. */ 1341 bserrno = 0; 1342 } 1343 1344 cluster_number = _spdk_bs_page_to_cluster(ctx->blob->bs, ctx->page); 1345 _spdk_bs_release_cluster(ctx->blob->bs, cluster_number); 1346 } 1347 1348 spdk_bs_sequence_finish(ctx->seq, bserrno); 1349 } 1350 1351 static void 1352 _spdk_blob_write_copy_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1353 { 1354 struct spdk_blob_copy_cluster_ctx *ctx = cb_arg; 1355 uint32_t cluster_number; 1356 1357 if (bserrno) { 1358 /* The write failed, so jump to the final completion handler */ 1359 spdk_bs_sequence_finish(seq, bserrno); 1360 return; 1361 } 1362 1363 cluster_number = _spdk_bs_page_to_cluster(ctx->blob->bs, ctx->page); 1364 1365 _spdk_blob_insert_cluster_on_md_thread(ctx->blob, cluster_number, ctx->new_cluster, 1366 _spdk_blob_insert_cluster_cpl, ctx); 1367 } 1368 1369 static void 1370 _spdk_blob_write_copy(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1371 { 1372 struct spdk_blob_copy_cluster_ctx *ctx = cb_arg; 1373 1374 if (bserrno != 0) { 1375 /* The read failed, so jump to the final completion handler */ 1376 spdk_bs_sequence_finish(seq, bserrno); 1377 return; 1378 } 1379 1380 /* Write whole cluster */ 1381 spdk_bs_sequence_write_dev(seq, ctx->buf, 1382 _spdk_bs_cluster_to_lba(ctx->blob->bs, ctx->new_cluster), 1383 _spdk_bs_cluster_to_lba(ctx->blob->bs, 1), 1384 _spdk_blob_write_copy_cpl, ctx); 1385 } 1386 1387 static void 1388 _spdk_bs_allocate_and_copy_cluster(struct spdk_blob *blob, 1389 struct spdk_io_channel *_ch, 1390 uint64_t offset, spdk_bs_user_op_t *op) 1391 { 1392 struct spdk_bs_cpl cpl; 1393 struct spdk_bs_channel *ch; 1394 struct spdk_blob_copy_cluster_ctx *ctx; 1395 uint32_t cluster_start_page; 1396 uint32_t cluster_number; 1397 int rc; 1398 1399 ch = spdk_io_channel_get_ctx(_ch); 1400 1401 if (!TAILQ_EMPTY(&ch->need_cluster_alloc)) { 1402 /* There are already operations pending. Queue this user op 1403 * and return because it will be re-executed when the outstanding 1404 * cluster allocation completes. */ 1405 TAILQ_INSERT_TAIL(&ch->need_cluster_alloc, op, link); 1406 return; 1407 } 1408 1409 /* Round the page offset down to the first page in the cluster */ 1410 cluster_start_page = _spdk_bs_page_to_cluster_start(blob, offset); 1411 1412 /* Calculate which index in the metadata cluster array the corresponding 1413 * cluster is supposed to be at. */ 1414 cluster_number = _spdk_bs_page_to_cluster(blob->bs, cluster_start_page); 1415 1416 ctx = calloc(1, sizeof(*ctx)); 1417 if (!ctx) { 1418 spdk_bs_user_op_abort(op); 1419 return; 1420 } 1421 1422 assert(blob->bs->cluster_sz % blob->back_bs_dev->blocklen == 0); 1423 1424 ctx->blob = blob; 1425 ctx->page = cluster_start_page; 1426 1427 ctx->buf = spdk_dma_malloc(blob->bs->cluster_sz, blob->back_bs_dev->blocklen, NULL); 1428 if (!ctx->buf) { 1429 SPDK_ERRLOG("DMA allocation for cluster of size = %" PRIu32 " failed.\n", 1430 blob->bs->cluster_sz); 1431 free(ctx); 1432 spdk_bs_user_op_abort(op); 1433 return; 1434 } 1435 1436 rc = _spdk_bs_allocate_cluster(blob, cluster_number, &ctx->new_cluster, false); 1437 if (rc != 0) { 1438 spdk_dma_free(ctx->buf); 1439 free(ctx); 1440 spdk_bs_user_op_abort(op); 1441 return; 1442 } 1443 1444 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1445 cpl.u.blob_basic.cb_fn = _spdk_blob_allocate_and_copy_cluster_cpl; 1446 cpl.u.blob_basic.cb_arg = ctx; 1447 1448 ctx->seq = spdk_bs_sequence_start(_ch, &cpl); 1449 if (!ctx->seq) { 1450 _spdk_bs_release_cluster(blob->bs, ctx->new_cluster); 1451 spdk_dma_free(ctx->buf); 1452 free(ctx); 1453 spdk_bs_user_op_abort(op); 1454 return; 1455 } 1456 1457 /* Queue the user op to block other incoming operations */ 1458 TAILQ_INSERT_TAIL(&ch->need_cluster_alloc, op, link); 1459 1460 /* Read cluster from backing device */ 1461 spdk_bs_sequence_read_bs_dev(ctx->seq, blob->back_bs_dev, ctx->buf, 1462 _spdk_bs_dev_page_to_lba(blob->back_bs_dev, cluster_start_page), 1463 _spdk_bs_dev_byte_to_lba(blob->back_bs_dev, blob->bs->cluster_sz), 1464 _spdk_blob_write_copy, ctx); 1465 } 1466 1467 static void 1468 _spdk_blob_calculate_lba_and_lba_count(struct spdk_blob *blob, uint64_t page, uint64_t length, 1469 uint64_t *lba, uint32_t *lba_count) 1470 { 1471 *lba_count = _spdk_bs_page_to_lba(blob->bs, length); 1472 1473 if (!_spdk_bs_page_is_allocated(blob, page)) { 1474 assert(blob->back_bs_dev != NULL); 1475 *lba = _spdk_bs_dev_page_to_lba(blob->back_bs_dev, page); 1476 *lba_count = _spdk_bs_blob_lba_to_back_dev_lba(blob, *lba_count); 1477 } else { 1478 *lba = _spdk_bs_blob_page_to_lba(blob, page); 1479 } 1480 } 1481 1482 static void 1483 _spdk_blob_request_submit_op_split(struct spdk_io_channel *ch, struct spdk_blob *blob, 1484 void *payload, uint64_t offset, uint64_t length, 1485 spdk_blob_op_complete cb_fn, void *cb_arg, enum spdk_blob_op_type op_type) 1486 { 1487 spdk_bs_batch_t *batch; 1488 struct spdk_bs_cpl cpl; 1489 uint64_t op_length; 1490 uint8_t *buf; 1491 1492 assert(blob != NULL); 1493 1494 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1495 cpl.u.blob_basic.cb_fn = cb_fn; 1496 cpl.u.blob_basic.cb_arg = cb_arg; 1497 1498 batch = spdk_bs_batch_open(ch, &cpl); 1499 if (!batch) { 1500 cb_fn(cb_arg, -ENOMEM); 1501 return; 1502 } 1503 1504 buf = payload; 1505 while (length > 0) { 1506 op_length = spdk_min(length, _spdk_bs_num_pages_to_cluster_boundary(blob, offset)); 1507 1508 switch (op_type) { 1509 case SPDK_BLOB_READ: 1510 spdk_bs_batch_read_blob(batch, blob, buf, offset, op_length); 1511 break; 1512 case SPDK_BLOB_WRITE: 1513 spdk_bs_batch_write_blob(batch, blob, buf, offset, op_length); 1514 break; 1515 case SPDK_BLOB_UNMAP: 1516 spdk_bs_batch_unmap_blob(batch, blob, offset, op_length); 1517 break; 1518 case SPDK_BLOB_WRITE_ZEROES: 1519 spdk_bs_batch_write_zeroes_blob(batch, blob, offset, op_length); 1520 break; 1521 case SPDK_BLOB_READV: 1522 case SPDK_BLOB_WRITEV: 1523 SPDK_ERRLOG("readv/write not valid for %s\n", __func__); 1524 break; 1525 } 1526 1527 length -= op_length; 1528 offset += op_length; 1529 if (op_type == SPDK_BLOB_WRITE || op_type == SPDK_BLOB_READ) { 1530 buf += op_length * SPDK_BS_PAGE_SIZE; 1531 } 1532 } 1533 1534 spdk_bs_batch_close(batch); 1535 } 1536 1537 static void 1538 _spdk_blob_request_submit_op_single(struct spdk_io_channel *_ch, struct spdk_blob *blob, 1539 void *payload, uint64_t offset, uint64_t length, 1540 spdk_blob_op_complete cb_fn, void *cb_arg, enum spdk_blob_op_type op_type) 1541 { 1542 struct spdk_bs_cpl cpl; 1543 uint64_t lba; 1544 uint32_t lba_count; 1545 1546 assert(blob != NULL); 1547 1548 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1549 cpl.u.blob_basic.cb_fn = cb_fn; 1550 cpl.u.blob_basic.cb_arg = cb_arg; 1551 1552 _spdk_blob_calculate_lba_and_lba_count(blob, offset, length, &lba, &lba_count); 1553 1554 switch (op_type) { 1555 case SPDK_BLOB_READ: { 1556 spdk_bs_batch_t *batch; 1557 1558 batch = spdk_bs_batch_open(_ch, &cpl); 1559 if (!batch) { 1560 cb_fn(cb_arg, -ENOMEM); 1561 return; 1562 } 1563 1564 if (_spdk_bs_page_is_allocated(blob, offset)) { 1565 /* Read from the blob */ 1566 spdk_bs_batch_read_dev(batch, payload, lba, lba_count); 1567 } else { 1568 /* Read from the backing block device */ 1569 spdk_bs_batch_read_bs_dev(batch, blob->back_bs_dev, payload, lba, lba_count); 1570 } 1571 1572 spdk_bs_batch_close(batch); 1573 break; 1574 } 1575 case SPDK_BLOB_WRITE: 1576 case SPDK_BLOB_WRITE_ZEROES: { 1577 if (_spdk_bs_page_is_allocated(blob, offset)) { 1578 /* Write to the blob */ 1579 spdk_bs_batch_t *batch; 1580 1581 batch = spdk_bs_batch_open(_ch, &cpl); 1582 if (!batch) { 1583 cb_fn(cb_arg, -ENOMEM); 1584 return; 1585 } 1586 1587 if (op_type == SPDK_BLOB_WRITE) { 1588 spdk_bs_batch_write_dev(batch, payload, lba, lba_count); 1589 } else { 1590 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1591 } 1592 1593 spdk_bs_batch_close(batch); 1594 } else { 1595 /* Queue this operation and allocate the cluster */ 1596 spdk_bs_user_op_t *op; 1597 1598 op = spdk_bs_user_op_alloc(_ch, &cpl, op_type, blob, payload, 0, offset, length); 1599 if (!op) { 1600 cb_fn(cb_arg, -ENOMEM); 1601 return; 1602 } 1603 1604 _spdk_bs_allocate_and_copy_cluster(blob, _ch, offset, op); 1605 } 1606 break; 1607 } 1608 case SPDK_BLOB_UNMAP: { 1609 spdk_bs_batch_t *batch; 1610 1611 batch = spdk_bs_batch_open(_ch, &cpl); 1612 if (!batch) { 1613 cb_fn(cb_arg, -ENOMEM); 1614 return; 1615 } 1616 1617 if (_spdk_bs_page_is_allocated(blob, offset)) { 1618 spdk_bs_batch_unmap_dev(batch, lba, lba_count); 1619 } 1620 1621 spdk_bs_batch_close(batch); 1622 break; 1623 } 1624 case SPDK_BLOB_READV: 1625 case SPDK_BLOB_WRITEV: 1626 SPDK_ERRLOG("readv/write not valid\n"); 1627 cb_fn(cb_arg, -EINVAL); 1628 break; 1629 } 1630 } 1631 1632 static void 1633 _spdk_blob_request_submit_op(struct spdk_blob *blob, struct spdk_io_channel *_channel, 1634 void *payload, uint64_t offset, uint64_t length, 1635 spdk_blob_op_complete cb_fn, void *cb_arg, enum spdk_blob_op_type op_type) 1636 { 1637 assert(blob != NULL); 1638 1639 if (blob->data_ro && op_type != SPDK_BLOB_READ) { 1640 cb_fn(cb_arg, -EPERM); 1641 return; 1642 } 1643 1644 if (offset + length > blob->active.num_clusters * blob->bs->pages_per_cluster) { 1645 cb_fn(cb_arg, -EINVAL); 1646 return; 1647 } 1648 1649 if (length <= _spdk_bs_num_pages_to_cluster_boundary(blob, offset)) { 1650 _spdk_blob_request_submit_op_single(_channel, blob, payload, offset, length, 1651 cb_fn, cb_arg, op_type); 1652 } else { 1653 _spdk_blob_request_submit_op_split(_channel, blob, payload, offset, length, 1654 cb_fn, cb_arg, op_type); 1655 } 1656 } 1657 1658 struct rw_iov_ctx { 1659 struct spdk_blob *blob; 1660 struct spdk_io_channel *channel; 1661 spdk_blob_op_complete cb_fn; 1662 void *cb_arg; 1663 bool read; 1664 int iovcnt; 1665 struct iovec *orig_iov; 1666 uint64_t page_offset; 1667 uint64_t pages_remaining; 1668 uint64_t pages_done; 1669 struct iovec iov[0]; 1670 }; 1671 1672 static void 1673 _spdk_rw_iov_done(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1674 { 1675 assert(cb_arg == NULL); 1676 spdk_bs_sequence_finish(seq, bserrno); 1677 } 1678 1679 static void 1680 _spdk_rw_iov_split_next(void *cb_arg, int bserrno) 1681 { 1682 struct rw_iov_ctx *ctx = cb_arg; 1683 struct spdk_blob *blob = ctx->blob; 1684 struct iovec *iov, *orig_iov; 1685 int iovcnt; 1686 size_t orig_iovoff; 1687 uint64_t page_count, pages_to_boundary, page_offset; 1688 uint64_t byte_count; 1689 1690 if (bserrno != 0 || ctx->pages_remaining == 0) { 1691 ctx->cb_fn(ctx->cb_arg, bserrno); 1692 free(ctx); 1693 return; 1694 } 1695 1696 page_offset = ctx->page_offset; 1697 pages_to_boundary = _spdk_bs_num_pages_to_cluster_boundary(blob, page_offset); 1698 page_count = spdk_min(ctx->pages_remaining, pages_to_boundary); 1699 1700 /* 1701 * Get index and offset into the original iov array for our current position in the I/O sequence. 1702 * byte_count will keep track of how many bytes remaining until orig_iov and orig_iovoff will 1703 * point to the current position in the I/O sequence. 1704 */ 1705 byte_count = ctx->pages_done * sizeof(struct spdk_blob_md_page); 1706 orig_iov = &ctx->orig_iov[0]; 1707 orig_iovoff = 0; 1708 while (byte_count > 0) { 1709 if (byte_count >= orig_iov->iov_len) { 1710 byte_count -= orig_iov->iov_len; 1711 orig_iov++; 1712 } else { 1713 orig_iovoff = byte_count; 1714 byte_count = 0; 1715 } 1716 } 1717 1718 /* 1719 * Build an iov array for the next I/O in the sequence. byte_count will keep track of how many 1720 * bytes of this next I/O remain to be accounted for in the new iov array. 1721 */ 1722 byte_count = page_count * sizeof(struct spdk_blob_md_page); 1723 iov = &ctx->iov[0]; 1724 iovcnt = 0; 1725 while (byte_count > 0) { 1726 iov->iov_len = spdk_min(byte_count, orig_iov->iov_len - orig_iovoff); 1727 iov->iov_base = orig_iov->iov_base + orig_iovoff; 1728 byte_count -= iov->iov_len; 1729 orig_iovoff = 0; 1730 orig_iov++; 1731 iov++; 1732 iovcnt++; 1733 } 1734 1735 ctx->page_offset += page_count; 1736 ctx->pages_done += page_count; 1737 ctx->pages_remaining -= page_count; 1738 iov = &ctx->iov[0]; 1739 1740 if (ctx->read) { 1741 spdk_blob_io_readv(ctx->blob, ctx->channel, iov, iovcnt, page_offset, 1742 page_count, _spdk_rw_iov_split_next, ctx); 1743 } else { 1744 spdk_blob_io_writev(ctx->blob, ctx->channel, iov, iovcnt, page_offset, 1745 page_count, _spdk_rw_iov_split_next, ctx); 1746 } 1747 } 1748 1749 static void 1750 _spdk_blob_request_submit_rw_iov(struct spdk_blob *blob, struct spdk_io_channel *_channel, 1751 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 1752 spdk_blob_op_complete cb_fn, void *cb_arg, bool read) 1753 { 1754 struct spdk_bs_cpl cpl; 1755 1756 assert(blob != NULL); 1757 1758 if (!read && blob->data_ro) { 1759 cb_fn(cb_arg, -EPERM); 1760 return; 1761 } 1762 1763 if (length == 0) { 1764 cb_fn(cb_arg, 0); 1765 return; 1766 } 1767 1768 if (offset + length > blob->active.num_clusters * blob->bs->pages_per_cluster) { 1769 cb_fn(cb_arg, -EINVAL); 1770 return; 1771 } 1772 1773 /* 1774 * For now, we implement readv/writev using a sequence (instead of a batch) to account for having 1775 * to split a request that spans a cluster boundary. For I/O that do not span a cluster boundary, 1776 * there will be no noticeable difference compared to using a batch. For I/O that do span a cluster 1777 * boundary, the target LBAs (after blob offset to LBA translation) may not be contiguous, so we need 1778 * to allocate a separate iov array and split the I/O such that none of the resulting 1779 * smaller I/O cross a cluster boundary. These smaller I/O will be issued in sequence (not in parallel) 1780 * but since this case happens very infrequently, any performance impact will be negligible. 1781 * 1782 * This could be optimized in the future to allocate a big enough iov array to account for all of the iovs 1783 * for all of the smaller I/Os, pre-build all of the iov arrays for the smaller I/Os, then issue them 1784 * in a batch. That would also require creating an intermediate spdk_bs_cpl that would get called 1785 * when the batch was completed, to allow for freeing the memory for the iov arrays. 1786 */ 1787 if (spdk_likely(length <= _spdk_bs_num_pages_to_cluster_boundary(blob, offset))) { 1788 uint32_t lba_count; 1789 uint64_t lba; 1790 1791 _spdk_blob_calculate_lba_and_lba_count(blob, offset, length, &lba, &lba_count); 1792 1793 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1794 cpl.u.blob_basic.cb_fn = cb_fn; 1795 cpl.u.blob_basic.cb_arg = cb_arg; 1796 1797 if (read) { 1798 spdk_bs_sequence_t *seq; 1799 1800 seq = spdk_bs_sequence_start(_channel, &cpl); 1801 if (!seq) { 1802 cb_fn(cb_arg, -ENOMEM); 1803 return; 1804 } 1805 1806 if (_spdk_bs_page_is_allocated(blob, offset)) { 1807 spdk_bs_sequence_readv_dev(seq, iov, iovcnt, lba, lba_count, _spdk_rw_iov_done, NULL); 1808 } else { 1809 spdk_bs_sequence_readv_bs_dev(seq, blob->back_bs_dev, iov, iovcnt, lba, lba_count, 1810 _spdk_rw_iov_done, NULL); 1811 } 1812 } else { 1813 if (_spdk_bs_page_is_allocated(blob, offset)) { 1814 spdk_bs_sequence_t *seq; 1815 1816 seq = spdk_bs_sequence_start(_channel, &cpl); 1817 if (!seq) { 1818 cb_fn(cb_arg, -ENOMEM); 1819 return; 1820 } 1821 1822 spdk_bs_sequence_writev_dev(seq, iov, iovcnt, lba, lba_count, _spdk_rw_iov_done, NULL); 1823 } else { 1824 /* Queue this operation and allocate the cluster */ 1825 spdk_bs_user_op_t *op; 1826 1827 op = spdk_bs_user_op_alloc(_channel, &cpl, SPDK_BLOB_WRITEV, blob, iov, iovcnt, offset, length); 1828 if (!op) { 1829 cb_fn(cb_arg, -ENOMEM); 1830 return; 1831 } 1832 1833 _spdk_bs_allocate_and_copy_cluster(blob, _channel, offset, op); 1834 } 1835 } 1836 } else { 1837 struct rw_iov_ctx *ctx; 1838 1839 ctx = calloc(1, sizeof(struct rw_iov_ctx) + iovcnt * sizeof(struct iovec)); 1840 if (ctx == NULL) { 1841 cb_fn(cb_arg, -ENOMEM); 1842 return; 1843 } 1844 1845 ctx->blob = blob; 1846 ctx->channel = _channel; 1847 ctx->cb_fn = cb_fn; 1848 ctx->cb_arg = cb_arg; 1849 ctx->read = read; 1850 ctx->orig_iov = iov; 1851 ctx->iovcnt = iovcnt; 1852 ctx->page_offset = offset; 1853 ctx->pages_remaining = length; 1854 ctx->pages_done = 0; 1855 1856 _spdk_rw_iov_split_next(ctx, 0); 1857 } 1858 } 1859 1860 static struct spdk_blob * 1861 _spdk_blob_lookup(struct spdk_blob_store *bs, spdk_blob_id blobid) 1862 { 1863 struct spdk_blob *blob; 1864 1865 TAILQ_FOREACH(blob, &bs->blobs, link) { 1866 if (blob->id == blobid) { 1867 return blob; 1868 } 1869 } 1870 1871 return NULL; 1872 } 1873 1874 static int 1875 _spdk_bs_channel_create(void *io_device, void *ctx_buf) 1876 { 1877 struct spdk_blob_store *bs = io_device; 1878 struct spdk_bs_channel *channel = ctx_buf; 1879 struct spdk_bs_dev *dev; 1880 uint32_t max_ops = bs->max_channel_ops; 1881 uint32_t i; 1882 1883 dev = bs->dev; 1884 1885 channel->req_mem = calloc(max_ops, sizeof(struct spdk_bs_request_set)); 1886 if (!channel->req_mem) { 1887 return -1; 1888 } 1889 1890 TAILQ_INIT(&channel->reqs); 1891 1892 for (i = 0; i < max_ops; i++) { 1893 TAILQ_INSERT_TAIL(&channel->reqs, &channel->req_mem[i], link); 1894 } 1895 1896 channel->bs = bs; 1897 channel->dev = dev; 1898 channel->dev_channel = dev->create_channel(dev); 1899 1900 if (!channel->dev_channel) { 1901 SPDK_ERRLOG("Failed to create device channel.\n"); 1902 free(channel->req_mem); 1903 return -1; 1904 } 1905 1906 TAILQ_INIT(&channel->need_cluster_alloc); 1907 1908 return 0; 1909 } 1910 1911 static void 1912 _spdk_bs_channel_destroy(void *io_device, void *ctx_buf) 1913 { 1914 struct spdk_bs_channel *channel = ctx_buf; 1915 spdk_bs_user_op_t *op; 1916 1917 while (!TAILQ_EMPTY(&channel->need_cluster_alloc)) { 1918 op = TAILQ_FIRST(&channel->need_cluster_alloc); 1919 TAILQ_REMOVE(&channel->need_cluster_alloc, op, link); 1920 spdk_bs_user_op_abort(op); 1921 } 1922 1923 free(channel->req_mem); 1924 channel->dev->destroy_channel(channel->dev, channel->dev_channel); 1925 } 1926 1927 static void 1928 _spdk_bs_dev_destroy(void *io_device) 1929 { 1930 struct spdk_blob_store *bs = io_device; 1931 struct spdk_blob *blob, *blob_tmp; 1932 1933 bs->dev->destroy(bs->dev); 1934 1935 TAILQ_FOREACH_SAFE(blob, &bs->blobs, link, blob_tmp) { 1936 TAILQ_REMOVE(&bs->blobs, blob, link); 1937 _spdk_blob_free(blob); 1938 } 1939 1940 pthread_mutex_destroy(&bs->used_clusters_mutex); 1941 1942 spdk_bit_array_free(&bs->used_blobids); 1943 spdk_bit_array_free(&bs->used_md_pages); 1944 spdk_bit_array_free(&bs->used_clusters); 1945 /* 1946 * If this function is called for any reason except a successful unload, 1947 * the unload_cpl type will be NONE and this will be a nop. 1948 */ 1949 spdk_bs_call_cpl(&bs->unload_cpl, bs->unload_err); 1950 1951 free(bs); 1952 } 1953 1954 static void 1955 _spdk_bs_free(struct spdk_blob_store *bs) 1956 { 1957 spdk_bs_unregister_md_thread(bs); 1958 spdk_io_device_unregister(bs, _spdk_bs_dev_destroy); 1959 } 1960 1961 void 1962 spdk_bs_opts_init(struct spdk_bs_opts *opts) 1963 { 1964 opts->cluster_sz = SPDK_BLOB_OPTS_CLUSTER_SZ; 1965 opts->num_md_pages = SPDK_BLOB_OPTS_NUM_MD_PAGES; 1966 opts->max_md_ops = SPDK_BLOB_OPTS_MAX_MD_OPS; 1967 opts->max_channel_ops = SPDK_BLOB_OPTS_DEFAULT_CHANNEL_OPS; 1968 memset(&opts->bstype, 0, sizeof(opts->bstype)); 1969 opts->iter_cb_fn = NULL; 1970 opts->iter_cb_arg = NULL; 1971 } 1972 1973 static int 1974 _spdk_bs_opts_verify(struct spdk_bs_opts *opts) 1975 { 1976 if (opts->cluster_sz == 0 || opts->num_md_pages == 0 || opts->max_md_ops == 0 || 1977 opts->max_channel_ops == 0) { 1978 SPDK_ERRLOG("Blobstore options cannot be set to 0\n"); 1979 return -1; 1980 } 1981 1982 return 0; 1983 } 1984 1985 static struct spdk_blob_store * 1986 _spdk_bs_alloc(struct spdk_bs_dev *dev, struct spdk_bs_opts *opts) 1987 { 1988 struct spdk_blob_store *bs; 1989 uint64_t dev_size; 1990 int rc; 1991 1992 dev_size = dev->blocklen * dev->blockcnt; 1993 if (dev_size < opts->cluster_sz) { 1994 /* Device size cannot be smaller than cluster size of blobstore */ 1995 SPDK_ERRLOG("Device size %" PRIu64 " is smaller than cluster size %" PRIu32 "\n", 1996 dev_size, opts->cluster_sz); 1997 return NULL; 1998 } 1999 if (opts->cluster_sz < SPDK_BS_PAGE_SIZE) { 2000 /* Cluster size cannot be smaller than page size */ 2001 SPDK_ERRLOG("Cluster size %" PRIu32 " is smaller than page size %d\n", 2002 opts->cluster_sz, SPDK_BS_PAGE_SIZE); 2003 return NULL; 2004 } 2005 bs = calloc(1, sizeof(struct spdk_blob_store)); 2006 if (!bs) { 2007 return NULL; 2008 } 2009 2010 TAILQ_INIT(&bs->blobs); 2011 bs->dev = dev; 2012 bs->md_thread = spdk_get_thread(); 2013 assert(bs->md_thread != NULL); 2014 2015 /* 2016 * Do not use _spdk_bs_lba_to_cluster() here since blockcnt may not be an 2017 * even multiple of the cluster size. 2018 */ 2019 bs->cluster_sz = opts->cluster_sz; 2020 bs->total_clusters = dev->blockcnt / (bs->cluster_sz / dev->blocklen); 2021 bs->pages_per_cluster = bs->cluster_sz / SPDK_BS_PAGE_SIZE; 2022 bs->num_free_clusters = bs->total_clusters; 2023 bs->used_clusters = spdk_bit_array_create(bs->total_clusters); 2024 if (bs->used_clusters == NULL) { 2025 free(bs); 2026 return NULL; 2027 } 2028 2029 bs->max_channel_ops = opts->max_channel_ops; 2030 bs->super_blob = SPDK_BLOBID_INVALID; 2031 memcpy(&bs->bstype, &opts->bstype, sizeof(opts->bstype)); 2032 2033 /* The metadata is assumed to be at least 1 page */ 2034 bs->used_md_pages = spdk_bit_array_create(1); 2035 bs->used_blobids = spdk_bit_array_create(0); 2036 2037 pthread_mutex_init(&bs->used_clusters_mutex, NULL); 2038 2039 spdk_io_device_register(bs, _spdk_bs_channel_create, _spdk_bs_channel_destroy, 2040 sizeof(struct spdk_bs_channel)); 2041 rc = spdk_bs_register_md_thread(bs); 2042 if (rc == -1) { 2043 spdk_io_device_unregister(bs, NULL); 2044 pthread_mutex_destroy(&bs->used_clusters_mutex); 2045 spdk_bit_array_free(&bs->used_blobids); 2046 spdk_bit_array_free(&bs->used_md_pages); 2047 spdk_bit_array_free(&bs->used_clusters); 2048 free(bs); 2049 return NULL; 2050 } 2051 2052 return bs; 2053 } 2054 2055 /* START spdk_bs_load, spdk_bs_load_ctx will used for both load and unload. */ 2056 2057 struct spdk_bs_load_ctx { 2058 struct spdk_blob_store *bs; 2059 struct spdk_bs_super_block *super; 2060 2061 struct spdk_bs_md_mask *mask; 2062 bool in_page_chain; 2063 uint32_t page_index; 2064 uint32_t cur_page; 2065 struct spdk_blob_md_page *page; 2066 bool is_load; 2067 2068 spdk_bs_sequence_t *seq; 2069 spdk_blob_op_with_handle_complete iter_cb_fn; 2070 void *iter_cb_arg; 2071 }; 2072 2073 static void 2074 _spdk_bs_load_ctx_fail(spdk_bs_sequence_t *seq, struct spdk_bs_load_ctx *ctx, int bserrno) 2075 { 2076 assert(bserrno != 0); 2077 2078 spdk_dma_free(ctx->super); 2079 spdk_bs_sequence_finish(seq, bserrno); 2080 /* 2081 * Only free the blobstore when a load fails. If an unload fails (for some reason) 2082 * we want to keep the blobstore in case the caller wants to try again. 2083 */ 2084 if (ctx->is_load) { 2085 _spdk_bs_free(ctx->bs); 2086 } 2087 free(ctx); 2088 } 2089 2090 static void 2091 _spdk_bs_set_mask(struct spdk_bit_array *array, struct spdk_bs_md_mask *mask) 2092 { 2093 uint32_t i = 0; 2094 2095 while (true) { 2096 i = spdk_bit_array_find_first_set(array, i); 2097 if (i >= mask->length) { 2098 break; 2099 } 2100 mask->mask[i / 8] |= 1U << (i % 8); 2101 i++; 2102 } 2103 } 2104 2105 static void 2106 _spdk_bs_write_super(spdk_bs_sequence_t *seq, struct spdk_blob_store *bs, 2107 struct spdk_bs_super_block *super, spdk_bs_sequence_cpl cb_fn, void *cb_arg) 2108 { 2109 /* Update the values in the super block */ 2110 super->super_blob = bs->super_blob; 2111 memcpy(&super->bstype, &bs->bstype, sizeof(bs->bstype)); 2112 super->crc = _spdk_blob_md_page_calc_crc(super); 2113 spdk_bs_sequence_write_dev(seq, super, _spdk_bs_page_to_lba(bs, 0), 2114 _spdk_bs_byte_to_lba(bs, sizeof(*super)), 2115 cb_fn, cb_arg); 2116 } 2117 2118 static void 2119 _spdk_bs_write_used_clusters(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 2120 { 2121 struct spdk_bs_load_ctx *ctx = arg; 2122 uint64_t mask_size, lba, lba_count; 2123 2124 /* Write out the used clusters mask */ 2125 mask_size = ctx->super->used_cluster_mask_len * SPDK_BS_PAGE_SIZE; 2126 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 2127 if (!ctx->mask) { 2128 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2129 return; 2130 } 2131 2132 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_CLUSTERS; 2133 ctx->mask->length = ctx->bs->total_clusters; 2134 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_clusters)); 2135 2136 _spdk_bs_set_mask(ctx->bs->used_clusters, ctx->mask); 2137 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_start); 2138 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_len); 2139 spdk_bs_sequence_write_dev(seq, ctx->mask, lba, lba_count, cb_fn, arg); 2140 } 2141 2142 static void 2143 _spdk_bs_write_used_md(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 2144 { 2145 struct spdk_bs_load_ctx *ctx = arg; 2146 uint64_t mask_size, lba, lba_count; 2147 2148 mask_size = ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE; 2149 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 2150 if (!ctx->mask) { 2151 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2152 return; 2153 } 2154 2155 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_PAGES; 2156 ctx->mask->length = ctx->super->md_len; 2157 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_md_pages)); 2158 2159 _spdk_bs_set_mask(ctx->bs->used_md_pages, ctx->mask); 2160 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_start); 2161 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_len); 2162 spdk_bs_sequence_write_dev(seq, ctx->mask, lba, lba_count, cb_fn, arg); 2163 } 2164 2165 static void 2166 _spdk_bs_write_used_blobids(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 2167 { 2168 struct spdk_bs_load_ctx *ctx = arg; 2169 uint64_t mask_size, lba, lba_count; 2170 2171 if (ctx->super->used_blobid_mask_len == 0) { 2172 /* 2173 * This is a pre-v3 on-disk format where the blobid mask does not get 2174 * written to disk. 2175 */ 2176 cb_fn(seq, arg, 0); 2177 return; 2178 } 2179 2180 mask_size = ctx->super->used_blobid_mask_len * SPDK_BS_PAGE_SIZE; 2181 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 2182 if (!ctx->mask) { 2183 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2184 return; 2185 } 2186 2187 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_BLOBIDS; 2188 ctx->mask->length = ctx->super->md_len; 2189 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_blobids)); 2190 2191 _spdk_bs_set_mask(ctx->bs->used_blobids, ctx->mask); 2192 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_start); 2193 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_len); 2194 spdk_bs_sequence_write_dev(seq, ctx->mask, lba, lba_count, cb_fn, arg); 2195 } 2196 2197 static void _spdk_bs_load_complete(spdk_bs_sequence_t *seq, struct spdk_bs_load_ctx *ctx, 2198 int bserrno); 2199 2200 static void 2201 _spdk_bs_load_iter(void *arg, struct spdk_blob *blob, int bserrno) 2202 { 2203 struct spdk_bs_load_ctx *ctx = arg; 2204 2205 if (bserrno == 0) { 2206 ctx->iter_cb_fn(ctx->iter_cb_arg, blob, 0); 2207 spdk_bs_iter_next(ctx->bs, blob, _spdk_bs_load_iter, ctx); 2208 return; 2209 } 2210 2211 if (bserrno == -ENOENT) { 2212 bserrno = 0; 2213 } else { 2214 /* 2215 * This case needs to be looked at further. Same problem 2216 * exists with applications that rely on explicit blob 2217 * iteration. We should just skip the blob that failed 2218 * to load and coontinue on to the next one. 2219 */ 2220 SPDK_ERRLOG("Error in iterating blobs\n"); 2221 } 2222 2223 ctx->iter_cb_fn = NULL; 2224 _spdk_bs_load_complete(ctx->seq, ctx, bserrno); 2225 } 2226 2227 static void 2228 _spdk_bs_load_complete(spdk_bs_sequence_t *seq, struct spdk_bs_load_ctx *ctx, int bserrno) 2229 { 2230 if (ctx->iter_cb_fn) { 2231 ctx->seq = seq; 2232 spdk_bs_iter_first(ctx->bs, _spdk_bs_load_iter, ctx); 2233 return; 2234 } 2235 2236 spdk_dma_free(ctx->super); 2237 spdk_dma_free(ctx->mask); 2238 free(ctx); 2239 spdk_bs_sequence_finish(seq, bserrno); 2240 } 2241 2242 static void 2243 _spdk_bs_load_used_blobids_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2244 { 2245 struct spdk_bs_load_ctx *ctx = cb_arg; 2246 uint32_t i, j; 2247 int rc; 2248 2249 /* The type must be correct */ 2250 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_BLOBIDS); 2251 2252 /* The length of the mask (in bits) must not be greater than 2253 * the length of the buffer (converted to bits) */ 2254 assert(ctx->mask->length <= (ctx->super->used_blobid_mask_len * SPDK_BS_PAGE_SIZE * 8)); 2255 2256 /* The length of the mask must be exactly equal to the size 2257 * (in pages) of the metadata region */ 2258 assert(ctx->mask->length == ctx->super->md_len); 2259 2260 rc = spdk_bit_array_resize(&ctx->bs->used_blobids, ctx->mask->length); 2261 if (rc < 0) { 2262 spdk_dma_free(ctx->mask); 2263 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2264 return; 2265 } 2266 2267 for (i = 0; i < ctx->mask->length / 8; i++) { 2268 uint8_t segment = ctx->mask->mask[i]; 2269 for (j = 0; segment; j++) { 2270 if (segment & 1U) { 2271 spdk_bit_array_set(ctx->bs->used_blobids, (i * 8) + j); 2272 } 2273 segment >>= 1U; 2274 } 2275 } 2276 2277 _spdk_bs_load_complete(seq, ctx, bserrno); 2278 } 2279 2280 static void 2281 _spdk_bs_load_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2282 { 2283 struct spdk_bs_load_ctx *ctx = cb_arg; 2284 uint64_t lba, lba_count, mask_size; 2285 uint32_t i, j; 2286 int rc; 2287 2288 /* The type must be correct */ 2289 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_CLUSTERS); 2290 /* The length of the mask (in bits) must not be greater than the length of the buffer (converted to bits) */ 2291 assert(ctx->mask->length <= (ctx->super->used_cluster_mask_len * sizeof( 2292 struct spdk_blob_md_page) * 8)); 2293 /* The length of the mask must be exactly equal to the total number of clusters */ 2294 assert(ctx->mask->length == ctx->bs->total_clusters); 2295 2296 rc = spdk_bit_array_resize(&ctx->bs->used_clusters, ctx->bs->total_clusters); 2297 if (rc < 0) { 2298 spdk_dma_free(ctx->mask); 2299 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2300 return; 2301 } 2302 2303 ctx->bs->num_free_clusters = ctx->bs->total_clusters; 2304 for (i = 0; i < ctx->mask->length / 8; i++) { 2305 uint8_t segment = ctx->mask->mask[i]; 2306 for (j = 0; segment && (j < 8); j++) { 2307 if (segment & 1U) { 2308 spdk_bit_array_set(ctx->bs->used_clusters, (i * 8) + j); 2309 assert(ctx->bs->num_free_clusters > 0); 2310 ctx->bs->num_free_clusters--; 2311 } 2312 segment >>= 1U; 2313 } 2314 } 2315 2316 spdk_dma_free(ctx->mask); 2317 2318 /* Read the used blobids mask */ 2319 mask_size = ctx->super->used_blobid_mask_len * SPDK_BS_PAGE_SIZE; 2320 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 2321 if (!ctx->mask) { 2322 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2323 return; 2324 } 2325 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_start); 2326 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_len); 2327 spdk_bs_sequence_read_dev(seq, ctx->mask, lba, lba_count, 2328 _spdk_bs_load_used_blobids_cpl, ctx); 2329 } 2330 2331 static void 2332 _spdk_bs_load_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2333 { 2334 struct spdk_bs_load_ctx *ctx = cb_arg; 2335 uint64_t lba, lba_count, mask_size; 2336 uint32_t i, j; 2337 int rc; 2338 2339 /* The type must be correct */ 2340 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_PAGES); 2341 /* The length of the mask (in bits) must not be greater than the length of the buffer (converted to bits) */ 2342 assert(ctx->mask->length <= (ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE * 2343 8)); 2344 /* The length of the mask must be exactly equal to the size (in pages) of the metadata region */ 2345 assert(ctx->mask->length == ctx->super->md_len); 2346 2347 rc = spdk_bit_array_resize(&ctx->bs->used_md_pages, ctx->mask->length); 2348 if (rc < 0) { 2349 spdk_dma_free(ctx->mask); 2350 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2351 return; 2352 } 2353 2354 for (i = 0; i < ctx->mask->length / 8; i++) { 2355 uint8_t segment = ctx->mask->mask[i]; 2356 for (j = 0; segment && (j < 8); j++) { 2357 if (segment & 1U) { 2358 spdk_bit_array_set(ctx->bs->used_md_pages, (i * 8) + j); 2359 } 2360 segment >>= 1U; 2361 } 2362 } 2363 spdk_dma_free(ctx->mask); 2364 2365 /* Read the used clusters mask */ 2366 mask_size = ctx->super->used_cluster_mask_len * SPDK_BS_PAGE_SIZE; 2367 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 2368 if (!ctx->mask) { 2369 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2370 return; 2371 } 2372 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_start); 2373 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_len); 2374 spdk_bs_sequence_read_dev(seq, ctx->mask, lba, lba_count, 2375 _spdk_bs_load_used_clusters_cpl, ctx); 2376 } 2377 2378 static void 2379 _spdk_bs_load_write_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2380 { 2381 struct spdk_bs_load_ctx *ctx = cb_arg; 2382 uint64_t lba, lba_count, mask_size; 2383 2384 /* Read the used pages mask */ 2385 mask_size = ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE; 2386 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 2387 if (!ctx->mask) { 2388 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2389 return; 2390 } 2391 2392 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_start); 2393 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_len); 2394 spdk_bs_sequence_read_dev(seq, ctx->mask, lba, lba_count, 2395 _spdk_bs_load_used_pages_cpl, ctx); 2396 } 2397 2398 static int 2399 _spdk_bs_load_replay_md_parse_page(const struct spdk_blob_md_page *page, struct spdk_blob_store *bs) 2400 { 2401 struct spdk_blob_md_descriptor *desc; 2402 size_t cur_desc = 0; 2403 2404 desc = (struct spdk_blob_md_descriptor *)page->descriptors; 2405 while (cur_desc < sizeof(page->descriptors)) { 2406 if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_PADDING) { 2407 if (desc->length == 0) { 2408 /* If padding and length are 0, this terminates the page */ 2409 break; 2410 } 2411 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_EXTENT) { 2412 struct spdk_blob_md_descriptor_extent *desc_extent; 2413 unsigned int i, j; 2414 unsigned int cluster_count = 0; 2415 2416 desc_extent = (struct spdk_blob_md_descriptor_extent *)desc; 2417 2418 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 2419 for (j = 0; j < desc_extent->extents[i].length; j++) { 2420 spdk_bit_array_set(bs->used_clusters, desc_extent->extents[i].cluster_idx + j); 2421 if (bs->num_free_clusters == 0) { 2422 return -1; 2423 } 2424 bs->num_free_clusters--; 2425 cluster_count++; 2426 } 2427 } 2428 if (cluster_count == 0) { 2429 return -1; 2430 } 2431 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR) { 2432 /* Skip this item */ 2433 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR_INTERNAL) { 2434 /* Skip this item */ 2435 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_FLAGS) { 2436 /* Skip this item */ 2437 } else { 2438 /* Error */ 2439 return -1; 2440 } 2441 /* Advance to the next descriptor */ 2442 cur_desc += sizeof(*desc) + desc->length; 2443 if (cur_desc + sizeof(*desc) > sizeof(page->descriptors)) { 2444 break; 2445 } 2446 desc = (struct spdk_blob_md_descriptor *)((uintptr_t)page->descriptors + cur_desc); 2447 } 2448 return 0; 2449 } 2450 2451 static bool _spdk_bs_load_cur_md_page_valid(struct spdk_bs_load_ctx *ctx) 2452 { 2453 uint32_t crc; 2454 2455 crc = _spdk_blob_md_page_calc_crc(ctx->page); 2456 if (crc != ctx->page->crc) { 2457 return false; 2458 } 2459 2460 if (_spdk_bs_page_to_blobid(ctx->cur_page) != ctx->page->id) { 2461 return false; 2462 } 2463 return true; 2464 } 2465 2466 static void 2467 _spdk_bs_load_replay_cur_md_page(spdk_bs_sequence_t *seq, void *cb_arg); 2468 2469 static void 2470 _spdk_bs_load_write_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2471 { 2472 struct spdk_bs_load_ctx *ctx = cb_arg; 2473 2474 _spdk_bs_load_complete(seq, ctx, bserrno); 2475 } 2476 2477 static void 2478 _spdk_bs_load_write_used_blobids_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2479 { 2480 struct spdk_bs_load_ctx *ctx = cb_arg; 2481 2482 spdk_dma_free(ctx->mask); 2483 ctx->mask = NULL; 2484 2485 _spdk_bs_write_used_clusters(seq, cb_arg, _spdk_bs_load_write_used_clusters_cpl); 2486 } 2487 2488 static void 2489 _spdk_bs_load_write_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2490 { 2491 struct spdk_bs_load_ctx *ctx = cb_arg; 2492 2493 spdk_dma_free(ctx->mask); 2494 ctx->mask = NULL; 2495 2496 _spdk_bs_write_used_blobids(seq, cb_arg, _spdk_bs_load_write_used_blobids_cpl); 2497 } 2498 2499 static void 2500 _spdk_bs_load_write_used_md(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2501 { 2502 _spdk_bs_write_used_md(seq, cb_arg, _spdk_bs_load_write_used_pages_cpl); 2503 } 2504 2505 static void 2506 _spdk_bs_load_replay_md_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2507 { 2508 struct spdk_bs_load_ctx *ctx = cb_arg; 2509 uint64_t num_md_clusters; 2510 uint64_t i; 2511 uint32_t page_num; 2512 2513 if (bserrno != 0) { 2514 _spdk_bs_load_ctx_fail(seq, ctx, bserrno); 2515 return; 2516 } 2517 2518 page_num = ctx->cur_page; 2519 if (_spdk_bs_load_cur_md_page_valid(ctx) == true) { 2520 if (ctx->page->sequence_num == 0 || ctx->in_page_chain == true) { 2521 spdk_bit_array_set(ctx->bs->used_md_pages, page_num); 2522 if (ctx->page->sequence_num == 0) { 2523 spdk_bit_array_set(ctx->bs->used_blobids, page_num); 2524 } 2525 if (_spdk_bs_load_replay_md_parse_page(ctx->page, ctx->bs)) { 2526 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 2527 return; 2528 } 2529 if (ctx->page->next != SPDK_INVALID_MD_PAGE) { 2530 ctx->in_page_chain = true; 2531 ctx->cur_page = ctx->page->next; 2532 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 2533 return; 2534 } 2535 } 2536 } 2537 2538 ctx->in_page_chain = false; 2539 2540 do { 2541 ctx->page_index++; 2542 } while (spdk_bit_array_get(ctx->bs->used_md_pages, ctx->page_index) == true); 2543 2544 if (ctx->page_index < ctx->super->md_len) { 2545 ctx->cur_page = ctx->page_index; 2546 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 2547 } else { 2548 /* Claim all of the clusters used by the metadata */ 2549 num_md_clusters = divide_round_up(ctx->super->md_len, ctx->bs->pages_per_cluster); 2550 for (i = 0; i < num_md_clusters; i++) { 2551 _spdk_bs_claim_cluster(ctx->bs, i); 2552 } 2553 spdk_dma_free(ctx->page); 2554 _spdk_bs_load_write_used_md(seq, ctx, bserrno); 2555 } 2556 } 2557 2558 static void 2559 _spdk_bs_load_replay_cur_md_page(spdk_bs_sequence_t *seq, void *cb_arg) 2560 { 2561 struct spdk_bs_load_ctx *ctx = cb_arg; 2562 uint64_t lba; 2563 2564 assert(ctx->cur_page < ctx->super->md_len); 2565 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->md_start + ctx->cur_page); 2566 spdk_bs_sequence_read_dev(seq, ctx->page, lba, 2567 _spdk_bs_byte_to_lba(ctx->bs, SPDK_BS_PAGE_SIZE), 2568 _spdk_bs_load_replay_md_cpl, ctx); 2569 } 2570 2571 static void 2572 _spdk_bs_load_replay_md(spdk_bs_sequence_t *seq, void *cb_arg) 2573 { 2574 struct spdk_bs_load_ctx *ctx = cb_arg; 2575 2576 ctx->page_index = 0; 2577 ctx->cur_page = 0; 2578 ctx->page = spdk_dma_zmalloc(SPDK_BS_PAGE_SIZE, 2579 SPDK_BS_PAGE_SIZE, 2580 NULL); 2581 if (!ctx->page) { 2582 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2583 return; 2584 } 2585 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 2586 } 2587 2588 static void 2589 _spdk_bs_recover(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2590 { 2591 struct spdk_bs_load_ctx *ctx = cb_arg; 2592 int rc; 2593 2594 if (bserrno != 0) { 2595 _spdk_bs_load_ctx_fail(seq, ctx, -EIO); 2596 return; 2597 } 2598 2599 rc = spdk_bit_array_resize(&ctx->bs->used_md_pages, ctx->super->md_len); 2600 if (rc < 0) { 2601 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2602 return; 2603 } 2604 2605 rc = spdk_bit_array_resize(&ctx->bs->used_blobids, ctx->super->md_len); 2606 if (rc < 0) { 2607 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2608 return; 2609 } 2610 2611 rc = spdk_bit_array_resize(&ctx->bs->used_clusters, ctx->bs->total_clusters); 2612 if (rc < 0) { 2613 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2614 return; 2615 } 2616 2617 ctx->bs->num_free_clusters = ctx->bs->total_clusters; 2618 _spdk_bs_load_replay_md(seq, cb_arg); 2619 } 2620 2621 static void 2622 _spdk_bs_load_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2623 { 2624 struct spdk_bs_load_ctx *ctx = cb_arg; 2625 uint32_t crc; 2626 static const char zeros[SPDK_BLOBSTORE_TYPE_LENGTH]; 2627 2628 if (ctx->super->version > SPDK_BS_VERSION || 2629 ctx->super->version < SPDK_BS_INITIAL_VERSION) { 2630 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 2631 return; 2632 } 2633 2634 if (memcmp(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 2635 sizeof(ctx->super->signature)) != 0) { 2636 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 2637 return; 2638 } 2639 2640 crc = _spdk_blob_md_page_calc_crc(ctx->super); 2641 if (crc != ctx->super->crc) { 2642 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 2643 return; 2644 } 2645 2646 if (memcmp(&ctx->bs->bstype, &ctx->super->bstype, SPDK_BLOBSTORE_TYPE_LENGTH) == 0) { 2647 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Bstype matched - loading blobstore\n"); 2648 } else if (memcmp(&ctx->bs->bstype, zeros, SPDK_BLOBSTORE_TYPE_LENGTH) == 0) { 2649 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Bstype wildcard used - loading blobstore regardless bstype\n"); 2650 } else { 2651 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Unexpected bstype\n"); 2652 SPDK_TRACEDUMP(SPDK_LOG_BLOB, "Expected:", ctx->bs->bstype.bstype, SPDK_BLOBSTORE_TYPE_LENGTH); 2653 SPDK_TRACEDUMP(SPDK_LOG_BLOB, "Found:", ctx->super->bstype.bstype, SPDK_BLOBSTORE_TYPE_LENGTH); 2654 _spdk_bs_load_ctx_fail(seq, ctx, -ENXIO); 2655 return; 2656 } 2657 2658 /* Parse the super block */ 2659 ctx->bs->cluster_sz = ctx->super->cluster_size; 2660 ctx->bs->total_clusters = ctx->bs->dev->blockcnt / (ctx->bs->cluster_sz / ctx->bs->dev->blocklen); 2661 ctx->bs->pages_per_cluster = ctx->bs->cluster_sz / SPDK_BS_PAGE_SIZE; 2662 ctx->bs->md_start = ctx->super->md_start; 2663 ctx->bs->md_len = ctx->super->md_len; 2664 ctx->bs->total_data_clusters = ctx->bs->total_clusters - divide_round_up( 2665 ctx->bs->md_start + ctx->bs->md_len, ctx->bs->pages_per_cluster); 2666 ctx->bs->super_blob = ctx->super->super_blob; 2667 memcpy(&ctx->bs->bstype, &ctx->super->bstype, sizeof(ctx->super->bstype)); 2668 2669 if (ctx->super->clean == 0) { 2670 _spdk_bs_recover(seq, ctx, 0); 2671 } else if (ctx->super->used_blobid_mask_len == 0) { 2672 /* 2673 * Metadata is clean, but this is an old metadata format without 2674 * a blobid mask. Clear the clean bit and then build the masks 2675 * using _spdk_bs_recover. 2676 */ 2677 ctx->super->clean = 0; 2678 _spdk_bs_write_super(seq, ctx->bs, ctx->super, _spdk_bs_recover, ctx); 2679 } else { 2680 ctx->super->clean = 0; 2681 _spdk_bs_write_super(seq, ctx->bs, ctx->super, _spdk_bs_load_write_super_cpl, ctx); 2682 } 2683 } 2684 2685 void 2686 spdk_bs_load(struct spdk_bs_dev *dev, struct spdk_bs_opts *o, 2687 spdk_bs_op_with_handle_complete cb_fn, void *cb_arg) 2688 { 2689 struct spdk_blob_store *bs; 2690 struct spdk_bs_cpl cpl; 2691 spdk_bs_sequence_t *seq; 2692 struct spdk_bs_load_ctx *ctx; 2693 struct spdk_bs_opts opts = {}; 2694 2695 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Loading blobstore from dev %p\n", dev); 2696 2697 if (o) { 2698 opts = *o; 2699 } else { 2700 spdk_bs_opts_init(&opts); 2701 } 2702 2703 if (opts.max_md_ops == 0 || opts.max_channel_ops == 0) { 2704 cb_fn(cb_arg, NULL, -EINVAL); 2705 return; 2706 } 2707 2708 bs = _spdk_bs_alloc(dev, &opts); 2709 if (!bs) { 2710 cb_fn(cb_arg, NULL, -ENOMEM); 2711 return; 2712 } 2713 2714 ctx = calloc(1, sizeof(*ctx)); 2715 if (!ctx) { 2716 _spdk_bs_free(bs); 2717 cb_fn(cb_arg, NULL, -ENOMEM); 2718 return; 2719 } 2720 2721 ctx->bs = bs; 2722 ctx->is_load = true; 2723 ctx->iter_cb_fn = opts.iter_cb_fn; 2724 ctx->iter_cb_arg = opts.iter_cb_arg; 2725 2726 /* Allocate memory for the super block */ 2727 ctx->super = spdk_dma_zmalloc(sizeof(*ctx->super), 0x1000, NULL); 2728 if (!ctx->super) { 2729 free(ctx); 2730 _spdk_bs_free(bs); 2731 return; 2732 } 2733 2734 cpl.type = SPDK_BS_CPL_TYPE_BS_HANDLE; 2735 cpl.u.bs_handle.cb_fn = cb_fn; 2736 cpl.u.bs_handle.cb_arg = cb_arg; 2737 cpl.u.bs_handle.bs = bs; 2738 2739 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 2740 if (!seq) { 2741 spdk_dma_free(ctx->super); 2742 free(ctx); 2743 _spdk_bs_free(bs); 2744 cb_fn(cb_arg, NULL, -ENOMEM); 2745 return; 2746 } 2747 2748 /* Read the super block */ 2749 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 2750 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 2751 _spdk_bs_load_super_cpl, ctx); 2752 } 2753 2754 /* END spdk_bs_load */ 2755 2756 /* START spdk_bs_init */ 2757 2758 struct spdk_bs_init_ctx { 2759 struct spdk_blob_store *bs; 2760 struct spdk_bs_super_block *super; 2761 }; 2762 2763 static void 2764 _spdk_bs_init_persist_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2765 { 2766 struct spdk_bs_init_ctx *ctx = cb_arg; 2767 2768 spdk_dma_free(ctx->super); 2769 free(ctx); 2770 2771 spdk_bs_sequence_finish(seq, bserrno); 2772 } 2773 2774 static void 2775 _spdk_bs_init_trim_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2776 { 2777 struct spdk_bs_init_ctx *ctx = cb_arg; 2778 2779 /* Write super block */ 2780 spdk_bs_sequence_write_dev(seq, ctx->super, _spdk_bs_page_to_lba(ctx->bs, 0), 2781 _spdk_bs_byte_to_lba(ctx->bs, sizeof(*ctx->super)), 2782 _spdk_bs_init_persist_super_cpl, ctx); 2783 } 2784 2785 void 2786 spdk_bs_init(struct spdk_bs_dev *dev, struct spdk_bs_opts *o, 2787 spdk_bs_op_with_handle_complete cb_fn, void *cb_arg) 2788 { 2789 struct spdk_bs_init_ctx *ctx; 2790 struct spdk_blob_store *bs; 2791 struct spdk_bs_cpl cpl; 2792 spdk_bs_sequence_t *seq; 2793 spdk_bs_batch_t *batch; 2794 uint64_t num_md_lba; 2795 uint64_t num_md_pages; 2796 uint64_t num_md_clusters; 2797 uint32_t i; 2798 struct spdk_bs_opts opts = {}; 2799 int rc; 2800 2801 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Initializing blobstore on dev %p\n", dev); 2802 2803 if ((SPDK_BS_PAGE_SIZE % dev->blocklen) != 0) { 2804 SPDK_ERRLOG("unsupported dev block length of %d\n", 2805 dev->blocklen); 2806 dev->destroy(dev); 2807 cb_fn(cb_arg, NULL, -EINVAL); 2808 return; 2809 } 2810 2811 if (o) { 2812 opts = *o; 2813 } else { 2814 spdk_bs_opts_init(&opts); 2815 } 2816 2817 if (_spdk_bs_opts_verify(&opts) != 0) { 2818 dev->destroy(dev); 2819 cb_fn(cb_arg, NULL, -EINVAL); 2820 return; 2821 } 2822 2823 bs = _spdk_bs_alloc(dev, &opts); 2824 if (!bs) { 2825 dev->destroy(dev); 2826 cb_fn(cb_arg, NULL, -ENOMEM); 2827 return; 2828 } 2829 2830 if (opts.num_md_pages == SPDK_BLOB_OPTS_NUM_MD_PAGES) { 2831 /* By default, allocate 1 page per cluster. 2832 * Technically, this over-allocates metadata 2833 * because more metadata will reduce the number 2834 * of usable clusters. This can be addressed with 2835 * more complex math in the future. 2836 */ 2837 bs->md_len = bs->total_clusters; 2838 } else { 2839 bs->md_len = opts.num_md_pages; 2840 } 2841 2842 rc = spdk_bit_array_resize(&bs->used_md_pages, bs->md_len); 2843 if (rc < 0) { 2844 _spdk_bs_free(bs); 2845 cb_fn(cb_arg, NULL, -ENOMEM); 2846 return; 2847 } 2848 2849 rc = spdk_bit_array_resize(&bs->used_blobids, bs->md_len); 2850 if (rc < 0) { 2851 _spdk_bs_free(bs); 2852 cb_fn(cb_arg, NULL, -ENOMEM); 2853 return; 2854 } 2855 2856 ctx = calloc(1, sizeof(*ctx)); 2857 if (!ctx) { 2858 _spdk_bs_free(bs); 2859 cb_fn(cb_arg, NULL, -ENOMEM); 2860 return; 2861 } 2862 2863 ctx->bs = bs; 2864 2865 /* Allocate memory for the super block */ 2866 ctx->super = spdk_dma_zmalloc(sizeof(*ctx->super), 0x1000, NULL); 2867 if (!ctx->super) { 2868 free(ctx); 2869 _spdk_bs_free(bs); 2870 return; 2871 } 2872 memcpy(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 2873 sizeof(ctx->super->signature)); 2874 ctx->super->version = SPDK_BS_VERSION; 2875 ctx->super->length = sizeof(*ctx->super); 2876 ctx->super->super_blob = bs->super_blob; 2877 ctx->super->clean = 0; 2878 ctx->super->cluster_size = bs->cluster_sz; 2879 memcpy(&ctx->super->bstype, &bs->bstype, sizeof(bs->bstype)); 2880 2881 /* Calculate how many pages the metadata consumes at the front 2882 * of the disk. 2883 */ 2884 2885 /* The super block uses 1 page */ 2886 num_md_pages = 1; 2887 2888 /* The used_md_pages mask requires 1 bit per metadata page, rounded 2889 * up to the nearest page, plus a header. 2890 */ 2891 ctx->super->used_page_mask_start = num_md_pages; 2892 ctx->super->used_page_mask_len = divide_round_up(sizeof(struct spdk_bs_md_mask) + 2893 divide_round_up(bs->md_len, 8), 2894 SPDK_BS_PAGE_SIZE); 2895 num_md_pages += ctx->super->used_page_mask_len; 2896 2897 /* The used_clusters mask requires 1 bit per cluster, rounded 2898 * up to the nearest page, plus a header. 2899 */ 2900 ctx->super->used_cluster_mask_start = num_md_pages; 2901 ctx->super->used_cluster_mask_len = divide_round_up(sizeof(struct spdk_bs_md_mask) + 2902 divide_round_up(bs->total_clusters, 8), 2903 SPDK_BS_PAGE_SIZE); 2904 num_md_pages += ctx->super->used_cluster_mask_len; 2905 2906 /* The used_blobids mask requires 1 bit per metadata page, rounded 2907 * up to the nearest page, plus a header. 2908 */ 2909 ctx->super->used_blobid_mask_start = num_md_pages; 2910 ctx->super->used_blobid_mask_len = divide_round_up(sizeof(struct spdk_bs_md_mask) + 2911 divide_round_up(bs->md_len, 8), 2912 SPDK_BS_PAGE_SIZE); 2913 num_md_pages += ctx->super->used_blobid_mask_len; 2914 2915 /* The metadata region size was chosen above */ 2916 ctx->super->md_start = bs->md_start = num_md_pages; 2917 ctx->super->md_len = bs->md_len; 2918 num_md_pages += bs->md_len; 2919 2920 num_md_lba = _spdk_bs_page_to_lba(bs, num_md_pages); 2921 2922 ctx->super->crc = _spdk_blob_md_page_calc_crc(ctx->super); 2923 2924 num_md_clusters = divide_round_up(num_md_pages, bs->pages_per_cluster); 2925 if (num_md_clusters > bs->total_clusters) { 2926 SPDK_ERRLOG("Blobstore metadata cannot use more clusters than is available, " 2927 "please decrease number of pages reserved for metadata " 2928 "or increase cluster size.\n"); 2929 spdk_dma_free(ctx->super); 2930 free(ctx); 2931 _spdk_bs_free(bs); 2932 cb_fn(cb_arg, NULL, -ENOMEM); 2933 return; 2934 } 2935 /* Claim all of the clusters used by the metadata */ 2936 for (i = 0; i < num_md_clusters; i++) { 2937 _spdk_bs_claim_cluster(bs, i); 2938 } 2939 2940 bs->total_data_clusters = bs->num_free_clusters; 2941 2942 cpl.type = SPDK_BS_CPL_TYPE_BS_HANDLE; 2943 cpl.u.bs_handle.cb_fn = cb_fn; 2944 cpl.u.bs_handle.cb_arg = cb_arg; 2945 cpl.u.bs_handle.bs = bs; 2946 2947 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 2948 if (!seq) { 2949 spdk_dma_free(ctx->super); 2950 free(ctx); 2951 _spdk_bs_free(bs); 2952 cb_fn(cb_arg, NULL, -ENOMEM); 2953 return; 2954 } 2955 2956 batch = spdk_bs_sequence_to_batch(seq, _spdk_bs_init_trim_cpl, ctx); 2957 2958 /* Clear metadata space */ 2959 spdk_bs_batch_write_zeroes_dev(batch, 0, num_md_lba); 2960 /* Trim data clusters */ 2961 spdk_bs_batch_unmap_dev(batch, num_md_lba, ctx->bs->dev->blockcnt - num_md_lba); 2962 2963 spdk_bs_batch_close(batch); 2964 } 2965 2966 /* END spdk_bs_init */ 2967 2968 /* START spdk_bs_destroy */ 2969 2970 static void 2971 _spdk_bs_destroy_trim_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2972 { 2973 struct spdk_bs_init_ctx *ctx = cb_arg; 2974 struct spdk_blob_store *bs = ctx->bs; 2975 2976 /* 2977 * We need to defer calling spdk_bs_call_cpl() until after 2978 * dev destruction, so tuck these away for later use. 2979 */ 2980 bs->unload_err = bserrno; 2981 memcpy(&bs->unload_cpl, &seq->cpl, sizeof(struct spdk_bs_cpl)); 2982 seq->cpl.type = SPDK_BS_CPL_TYPE_NONE; 2983 2984 spdk_bs_sequence_finish(seq, bserrno); 2985 2986 _spdk_bs_free(bs); 2987 free(ctx); 2988 } 2989 2990 void 2991 spdk_bs_destroy(struct spdk_blob_store *bs, spdk_bs_op_complete cb_fn, 2992 void *cb_arg) 2993 { 2994 struct spdk_bs_cpl cpl; 2995 spdk_bs_sequence_t *seq; 2996 struct spdk_bs_init_ctx *ctx; 2997 2998 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Destroying blobstore\n"); 2999 3000 if (!TAILQ_EMPTY(&bs->blobs)) { 3001 SPDK_ERRLOG("Blobstore still has open blobs\n"); 3002 cb_fn(cb_arg, -EBUSY); 3003 return; 3004 } 3005 3006 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 3007 cpl.u.bs_basic.cb_fn = cb_fn; 3008 cpl.u.bs_basic.cb_arg = cb_arg; 3009 3010 ctx = calloc(1, sizeof(*ctx)); 3011 if (!ctx) { 3012 cb_fn(cb_arg, -ENOMEM); 3013 return; 3014 } 3015 3016 ctx->bs = bs; 3017 3018 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3019 if (!seq) { 3020 free(ctx); 3021 cb_fn(cb_arg, -ENOMEM); 3022 return; 3023 } 3024 3025 /* Write zeroes to the super block */ 3026 spdk_bs_sequence_write_zeroes_dev(seq, 3027 _spdk_bs_page_to_lba(bs, 0), 3028 _spdk_bs_byte_to_lba(bs, sizeof(struct spdk_bs_super_block)), 3029 _spdk_bs_destroy_trim_cpl, ctx); 3030 } 3031 3032 /* END spdk_bs_destroy */ 3033 3034 /* START spdk_bs_unload */ 3035 3036 static void 3037 _spdk_bs_unload_write_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3038 { 3039 struct spdk_bs_load_ctx *ctx = cb_arg; 3040 3041 spdk_dma_free(ctx->super); 3042 3043 /* 3044 * We need to defer calling spdk_bs_call_cpl() until after 3045 * dev destuction, so tuck these away for later use. 3046 */ 3047 ctx->bs->unload_err = bserrno; 3048 memcpy(&ctx->bs->unload_cpl, &seq->cpl, sizeof(struct spdk_bs_cpl)); 3049 seq->cpl.type = SPDK_BS_CPL_TYPE_NONE; 3050 3051 spdk_bs_sequence_finish(seq, bserrno); 3052 3053 _spdk_bs_free(ctx->bs); 3054 free(ctx); 3055 } 3056 3057 static void 3058 _spdk_bs_unload_write_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3059 { 3060 struct spdk_bs_load_ctx *ctx = cb_arg; 3061 3062 spdk_dma_free(ctx->mask); 3063 ctx->super->clean = 1; 3064 3065 _spdk_bs_write_super(seq, ctx->bs, ctx->super, _spdk_bs_unload_write_super_cpl, ctx); 3066 } 3067 3068 static void 3069 _spdk_bs_unload_write_used_blobids_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3070 { 3071 struct spdk_bs_load_ctx *ctx = cb_arg; 3072 3073 spdk_dma_free(ctx->mask); 3074 ctx->mask = NULL; 3075 3076 _spdk_bs_write_used_clusters(seq, cb_arg, _spdk_bs_unload_write_used_clusters_cpl); 3077 } 3078 3079 static void 3080 _spdk_bs_unload_write_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3081 { 3082 struct spdk_bs_load_ctx *ctx = cb_arg; 3083 3084 spdk_dma_free(ctx->mask); 3085 ctx->mask = NULL; 3086 3087 _spdk_bs_write_used_blobids(seq, cb_arg, _spdk_bs_unload_write_used_blobids_cpl); 3088 } 3089 3090 static void 3091 _spdk_bs_unload_read_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3092 { 3093 _spdk_bs_write_used_md(seq, cb_arg, _spdk_bs_unload_write_used_pages_cpl); 3094 } 3095 3096 void 3097 spdk_bs_unload(struct spdk_blob_store *bs, spdk_bs_op_complete cb_fn, void *cb_arg) 3098 { 3099 struct spdk_bs_cpl cpl; 3100 spdk_bs_sequence_t *seq; 3101 struct spdk_bs_load_ctx *ctx; 3102 3103 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Syncing blobstore\n"); 3104 3105 if (!TAILQ_EMPTY(&bs->blobs)) { 3106 SPDK_ERRLOG("Blobstore still has open blobs\n"); 3107 cb_fn(cb_arg, -EBUSY); 3108 return; 3109 } 3110 3111 ctx = calloc(1, sizeof(*ctx)); 3112 if (!ctx) { 3113 cb_fn(cb_arg, -ENOMEM); 3114 return; 3115 } 3116 3117 ctx->bs = bs; 3118 ctx->is_load = false; 3119 3120 ctx->super = spdk_dma_zmalloc(sizeof(*ctx->super), 0x1000, NULL); 3121 if (!ctx->super) { 3122 free(ctx); 3123 cb_fn(cb_arg, -ENOMEM); 3124 return; 3125 } 3126 3127 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 3128 cpl.u.bs_basic.cb_fn = cb_fn; 3129 cpl.u.bs_basic.cb_arg = cb_arg; 3130 3131 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3132 if (!seq) { 3133 spdk_dma_free(ctx->super); 3134 free(ctx); 3135 cb_fn(cb_arg, -ENOMEM); 3136 return; 3137 } 3138 3139 /* Read super block */ 3140 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 3141 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 3142 _spdk_bs_unload_read_super_cpl, ctx); 3143 } 3144 3145 /* END spdk_bs_unload */ 3146 3147 void 3148 spdk_bs_set_super(struct spdk_blob_store *bs, spdk_blob_id blobid, 3149 spdk_bs_op_complete cb_fn, void *cb_arg) 3150 { 3151 bs->super_blob = blobid; 3152 cb_fn(cb_arg, 0); 3153 } 3154 3155 void 3156 spdk_bs_get_super(struct spdk_blob_store *bs, 3157 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 3158 { 3159 if (bs->super_blob == SPDK_BLOBID_INVALID) { 3160 cb_fn(cb_arg, SPDK_BLOBID_INVALID, -ENOENT); 3161 } else { 3162 cb_fn(cb_arg, bs->super_blob, 0); 3163 } 3164 } 3165 3166 uint64_t 3167 spdk_bs_get_cluster_size(struct spdk_blob_store *bs) 3168 { 3169 return bs->cluster_sz; 3170 } 3171 3172 uint64_t 3173 spdk_bs_get_page_size(struct spdk_blob_store *bs) 3174 { 3175 return SPDK_BS_PAGE_SIZE; 3176 } 3177 3178 uint64_t 3179 spdk_bs_free_cluster_count(struct spdk_blob_store *bs) 3180 { 3181 return bs->num_free_clusters; 3182 } 3183 3184 uint64_t 3185 spdk_bs_total_data_cluster_count(struct spdk_blob_store *bs) 3186 { 3187 return bs->total_data_clusters; 3188 } 3189 3190 static int 3191 spdk_bs_register_md_thread(struct spdk_blob_store *bs) 3192 { 3193 bs->md_channel = spdk_get_io_channel(bs); 3194 if (!bs->md_channel) { 3195 SPDK_ERRLOG("Failed to get IO channel.\n"); 3196 return -1; 3197 } 3198 3199 return 0; 3200 } 3201 3202 static int 3203 spdk_bs_unregister_md_thread(struct spdk_blob_store *bs) 3204 { 3205 spdk_put_io_channel(bs->md_channel); 3206 3207 return 0; 3208 } 3209 3210 spdk_blob_id spdk_blob_get_id(struct spdk_blob *blob) 3211 { 3212 assert(blob != NULL); 3213 3214 return blob->id; 3215 } 3216 3217 uint64_t spdk_blob_get_num_pages(struct spdk_blob *blob) 3218 { 3219 assert(blob != NULL); 3220 3221 return _spdk_bs_cluster_to_page(blob->bs, blob->active.num_clusters); 3222 } 3223 3224 uint64_t spdk_blob_get_num_clusters(struct spdk_blob *blob) 3225 { 3226 assert(blob != NULL); 3227 3228 return blob->active.num_clusters; 3229 } 3230 3231 /* START spdk_bs_create_blob */ 3232 3233 static void 3234 _spdk_bs_create_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3235 { 3236 struct spdk_blob *blob = cb_arg; 3237 3238 _spdk_blob_free(blob); 3239 3240 spdk_bs_sequence_finish(seq, bserrno); 3241 } 3242 3243 static int 3244 _spdk_blob_set_xattrs(struct spdk_blob *blob, const struct spdk_blob_xattr_opts *xattrs, 3245 bool internal) 3246 { 3247 uint64_t i; 3248 size_t value_len = 0; 3249 int rc; 3250 const void *value = NULL; 3251 if (xattrs->count > 0 && xattrs->get_value == NULL) { 3252 return -EINVAL; 3253 } 3254 for (i = 0; i < xattrs->count; i++) { 3255 xattrs->get_value(xattrs->ctx, xattrs->names[i], &value, &value_len); 3256 if (value == NULL || value_len == 0) { 3257 return -EINVAL; 3258 } 3259 rc = _spdk_blob_set_xattr(blob, xattrs->names[i], value, value_len, internal); 3260 if (rc < 0) { 3261 return rc; 3262 } 3263 } 3264 return 0; 3265 } 3266 3267 static void 3268 _spdk_blob_set_thin_provision(struct spdk_blob *blob) 3269 { 3270 _spdk_blob_verify_md_op(blob); 3271 blob->invalid_flags |= SPDK_BLOB_THIN_PROV; 3272 blob->state = SPDK_BLOB_STATE_DIRTY; 3273 } 3274 3275 void spdk_bs_create_blob_ext(struct spdk_blob_store *bs, const struct spdk_blob_opts *opts, 3276 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 3277 { 3278 struct spdk_blob *blob; 3279 uint32_t page_idx; 3280 struct spdk_bs_cpl cpl; 3281 struct spdk_blob_opts opts_default; 3282 spdk_bs_sequence_t *seq; 3283 spdk_blob_id id; 3284 int rc; 3285 3286 assert(spdk_get_thread() == bs->md_thread); 3287 3288 page_idx = spdk_bit_array_find_first_clear(bs->used_md_pages, 0); 3289 if (page_idx >= spdk_bit_array_capacity(bs->used_md_pages)) { 3290 cb_fn(cb_arg, 0, -ENOMEM); 3291 return; 3292 } 3293 spdk_bit_array_set(bs->used_blobids, page_idx); 3294 spdk_bit_array_set(bs->used_md_pages, page_idx); 3295 3296 id = _spdk_bs_page_to_blobid(page_idx); 3297 3298 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Creating blob with id %lu at page %u\n", id, page_idx); 3299 3300 blob = _spdk_blob_alloc(bs, id); 3301 if (!blob) { 3302 cb_fn(cb_arg, 0, -ENOMEM); 3303 return; 3304 } 3305 3306 if (!opts) { 3307 spdk_blob_opts_init(&opts_default); 3308 opts = &opts_default; 3309 } 3310 3311 rc = _spdk_blob_set_xattrs(blob, &opts->xattrs, false); 3312 if (rc < 0) { 3313 _spdk_blob_free(blob); 3314 cb_fn(cb_arg, 0, rc); 3315 return; 3316 } 3317 if (opts->thin_provision) { 3318 _spdk_blob_set_thin_provision(blob); 3319 } 3320 3321 rc = spdk_blob_resize(blob, opts->num_clusters); 3322 if (rc < 0) { 3323 _spdk_blob_free(blob); 3324 cb_fn(cb_arg, 0, rc); 3325 return; 3326 } 3327 cpl.type = SPDK_BS_CPL_TYPE_BLOBID; 3328 cpl.u.blobid.cb_fn = cb_fn; 3329 cpl.u.blobid.cb_arg = cb_arg; 3330 cpl.u.blobid.blobid = blob->id; 3331 3332 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3333 if (!seq) { 3334 _spdk_blob_free(blob); 3335 cb_fn(cb_arg, 0, -ENOMEM); 3336 return; 3337 } 3338 3339 _spdk_blob_persist(seq, blob, _spdk_bs_create_blob_cpl, blob); 3340 } 3341 3342 void spdk_bs_create_blob(struct spdk_blob_store *bs, 3343 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 3344 { 3345 spdk_bs_create_blob_ext(bs, NULL, cb_fn, cb_arg); 3346 } 3347 3348 /* END spdk_bs_create_blob */ 3349 3350 /* START spdk_blob_resize */ 3351 int 3352 spdk_blob_resize(struct spdk_blob *blob, uint64_t sz) 3353 { 3354 int rc; 3355 3356 _spdk_blob_verify_md_op(blob); 3357 3358 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Resizing blob %lu to %lu clusters\n", blob->id, sz); 3359 3360 if (blob->md_ro) { 3361 return -EPERM; 3362 } 3363 3364 if (sz == blob->active.num_clusters) { 3365 return 0; 3366 } 3367 3368 rc = _spdk_resize_blob(blob, sz); 3369 if (rc < 0) { 3370 return rc; 3371 } 3372 3373 return 0; 3374 } 3375 3376 /* END spdk_blob_resize */ 3377 3378 3379 /* START spdk_bs_delete_blob */ 3380 3381 static void 3382 _spdk_bs_delete_close_cpl(void *cb_arg, int bserrno) 3383 { 3384 spdk_bs_sequence_t *seq = cb_arg; 3385 3386 spdk_bs_sequence_finish(seq, bserrno); 3387 } 3388 3389 static void 3390 _spdk_bs_delete_persist_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3391 { 3392 struct spdk_blob *blob = cb_arg; 3393 3394 if (bserrno != 0) { 3395 /* 3396 * We already removed this blob from the blobstore tailq, so 3397 * we need to free it here since this is the last reference 3398 * to it. 3399 */ 3400 _spdk_blob_free(blob); 3401 _spdk_bs_delete_close_cpl(seq, bserrno); 3402 return; 3403 } 3404 3405 /* 3406 * This will immediately decrement the ref_count and call 3407 * the completion routine since the metadata state is clean. 3408 * By calling spdk_blob_close, we reduce the number of call 3409 * points into code that touches the blob->open_ref count 3410 * and the blobstore's blob list. 3411 */ 3412 spdk_blob_close(blob, _spdk_bs_delete_close_cpl, seq); 3413 } 3414 3415 static void 3416 _spdk_bs_delete_open_cpl(void *cb_arg, struct spdk_blob *blob, int bserrno) 3417 { 3418 spdk_bs_sequence_t *seq = cb_arg; 3419 uint32_t page_num; 3420 3421 if (bserrno != 0) { 3422 spdk_bs_sequence_finish(seq, bserrno); 3423 return; 3424 } 3425 3426 _spdk_blob_verify_md_op(blob); 3427 3428 if (blob->open_ref > 1) { 3429 /* 3430 * Someone has this blob open (besides this delete context). 3431 * Decrement the ref count directly and return -EBUSY. 3432 */ 3433 blob->open_ref--; 3434 spdk_bs_sequence_finish(seq, -EBUSY); 3435 return; 3436 } 3437 3438 /* 3439 * Remove the blob from the blob_store list now, to ensure it does not 3440 * get returned after this point by _spdk_blob_lookup(). 3441 */ 3442 TAILQ_REMOVE(&blob->bs->blobs, blob, link); 3443 page_num = _spdk_bs_blobid_to_page(blob->id); 3444 spdk_bit_array_clear(blob->bs->used_blobids, page_num); 3445 blob->state = SPDK_BLOB_STATE_DIRTY; 3446 blob->active.num_pages = 0; 3447 _spdk_resize_blob(blob, 0); 3448 3449 _spdk_blob_persist(seq, blob, _spdk_bs_delete_persist_cpl, blob); 3450 } 3451 3452 void 3453 spdk_bs_delete_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 3454 spdk_blob_op_complete cb_fn, void *cb_arg) 3455 { 3456 struct spdk_bs_cpl cpl; 3457 spdk_bs_sequence_t *seq; 3458 3459 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Deleting blob %lu\n", blobid); 3460 3461 assert(spdk_get_thread() == bs->md_thread); 3462 3463 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 3464 cpl.u.blob_basic.cb_fn = cb_fn; 3465 cpl.u.blob_basic.cb_arg = cb_arg; 3466 3467 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3468 if (!seq) { 3469 cb_fn(cb_arg, -ENOMEM); 3470 return; 3471 } 3472 3473 spdk_bs_open_blob(bs, blobid, _spdk_bs_delete_open_cpl, seq); 3474 } 3475 3476 /* END spdk_bs_delete_blob */ 3477 3478 /* START spdk_bs_open_blob */ 3479 3480 static void 3481 _spdk_bs_open_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3482 { 3483 struct spdk_blob *blob = cb_arg; 3484 3485 /* If the blob have crc error, we just return NULL. */ 3486 if (blob == NULL) { 3487 seq->cpl.u.blob_handle.blob = NULL; 3488 spdk_bs_sequence_finish(seq, bserrno); 3489 return; 3490 } 3491 3492 blob->open_ref++; 3493 3494 TAILQ_INSERT_HEAD(&blob->bs->blobs, blob, link); 3495 3496 spdk_bs_sequence_finish(seq, bserrno); 3497 } 3498 3499 void spdk_bs_open_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 3500 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 3501 { 3502 struct spdk_blob *blob; 3503 struct spdk_bs_cpl cpl; 3504 spdk_bs_sequence_t *seq; 3505 uint32_t page_num; 3506 3507 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Opening blob %lu\n", blobid); 3508 assert(spdk_get_thread() == bs->md_thread); 3509 3510 page_num = _spdk_bs_blobid_to_page(blobid); 3511 if (spdk_bit_array_get(bs->used_blobids, page_num) == false) { 3512 /* Invalid blobid */ 3513 cb_fn(cb_arg, NULL, -ENOENT); 3514 return; 3515 } 3516 3517 blob = _spdk_blob_lookup(bs, blobid); 3518 if (blob) { 3519 blob->open_ref++; 3520 cb_fn(cb_arg, blob, 0); 3521 return; 3522 } 3523 3524 blob = _spdk_blob_alloc(bs, blobid); 3525 if (!blob) { 3526 cb_fn(cb_arg, NULL, -ENOMEM); 3527 return; 3528 } 3529 3530 cpl.type = SPDK_BS_CPL_TYPE_BLOB_HANDLE; 3531 cpl.u.blob_handle.cb_fn = cb_fn; 3532 cpl.u.blob_handle.cb_arg = cb_arg; 3533 cpl.u.blob_handle.blob = blob; 3534 3535 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3536 if (!seq) { 3537 _spdk_blob_free(blob); 3538 cb_fn(cb_arg, NULL, -ENOMEM); 3539 return; 3540 } 3541 3542 _spdk_blob_load(seq, blob, _spdk_bs_open_blob_cpl, blob); 3543 } 3544 /* END spdk_bs_open_blob */ 3545 3546 /* START spdk_blob_set_read_only */ 3547 int spdk_blob_set_read_only(struct spdk_blob *blob) 3548 { 3549 _spdk_blob_verify_md_op(blob); 3550 3551 blob->data_ro_flags |= SPDK_BLOB_READ_ONLY; 3552 3553 blob->state = SPDK_BLOB_STATE_DIRTY; 3554 return 0; 3555 } 3556 /* END spdk_blob_set_read_only */ 3557 3558 /* START spdk_blob_sync_md */ 3559 3560 static void 3561 _spdk_blob_sync_md_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3562 { 3563 struct spdk_blob *blob = cb_arg; 3564 3565 if (bserrno == 0 && (blob->data_ro_flags & SPDK_BLOB_READ_ONLY)) { 3566 blob->data_ro = true; 3567 blob->md_ro = true; 3568 } 3569 3570 spdk_bs_sequence_finish(seq, bserrno); 3571 } 3572 3573 static void 3574 _spdk_blob_sync_md(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 3575 { 3576 struct spdk_bs_cpl cpl; 3577 spdk_bs_sequence_t *seq; 3578 3579 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 3580 cpl.u.blob_basic.cb_fn = cb_fn; 3581 cpl.u.blob_basic.cb_arg = cb_arg; 3582 3583 seq = spdk_bs_sequence_start(blob->bs->md_channel, &cpl); 3584 if (!seq) { 3585 cb_fn(cb_arg, -ENOMEM); 3586 return; 3587 } 3588 3589 _spdk_blob_persist(seq, blob, _spdk_blob_sync_md_cpl, blob); 3590 } 3591 3592 void 3593 spdk_blob_sync_md(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 3594 { 3595 _spdk_blob_verify_md_op(blob); 3596 3597 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Syncing blob %lu\n", blob->id); 3598 3599 if (blob->md_ro) { 3600 assert(blob->state == SPDK_BLOB_STATE_CLEAN); 3601 cb_fn(cb_arg, 0); 3602 return; 3603 } 3604 3605 _spdk_blob_sync_md(blob, cb_fn, cb_arg); 3606 } 3607 3608 /* END spdk_blob_sync_md */ 3609 3610 struct spdk_blob_insert_cluster_ctx { 3611 struct spdk_thread *thread; 3612 struct spdk_blob *blob; 3613 uint32_t cluster_num; /* cluster index in blob */ 3614 uint32_t cluster; /* cluster on disk */ 3615 int rc; 3616 spdk_blob_op_complete cb_fn; 3617 void *cb_arg; 3618 }; 3619 3620 static void 3621 _spdk_blob_insert_cluster_msg_cpl(void *arg) 3622 { 3623 struct spdk_blob_insert_cluster_ctx *ctx = arg; 3624 3625 ctx->cb_fn(ctx->cb_arg, ctx->rc); 3626 free(ctx); 3627 } 3628 3629 static void 3630 _spdk_blob_insert_cluster_msg_cb(void *arg, int bserrno) 3631 { 3632 struct spdk_blob_insert_cluster_ctx *ctx = arg; 3633 3634 ctx->rc = bserrno; 3635 spdk_thread_send_msg(ctx->thread, _spdk_blob_insert_cluster_msg_cpl, ctx); 3636 } 3637 3638 static void 3639 _spdk_blob_insert_cluster_msg(void *arg) 3640 { 3641 struct spdk_blob_insert_cluster_ctx *ctx = arg; 3642 3643 ctx->rc = _spdk_blob_insert_cluster(ctx->blob, ctx->cluster_num, ctx->cluster); 3644 if (ctx->rc != 0) { 3645 spdk_thread_send_msg(ctx->thread, _spdk_blob_insert_cluster_msg_cpl, ctx); 3646 return; 3647 } 3648 3649 ctx->blob->state = SPDK_BLOB_STATE_DIRTY; 3650 _spdk_blob_sync_md(ctx->blob, _spdk_blob_insert_cluster_msg_cb, ctx); 3651 } 3652 3653 void 3654 _spdk_blob_insert_cluster_on_md_thread(struct spdk_blob *blob, uint32_t cluster_num, 3655 uint64_t cluster, spdk_blob_op_complete cb_fn, void *cb_arg) 3656 { 3657 struct spdk_blob_insert_cluster_ctx *ctx; 3658 3659 ctx = calloc(1, sizeof(*ctx)); 3660 if (ctx == NULL) { 3661 cb_fn(cb_arg, -ENOMEM); 3662 return; 3663 } 3664 3665 ctx->thread = spdk_get_thread(); 3666 ctx->blob = blob; 3667 ctx->cluster_num = cluster_num; 3668 ctx->cluster = cluster; 3669 ctx->cb_fn = cb_fn; 3670 ctx->cb_arg = cb_arg; 3671 3672 spdk_thread_send_msg(blob->bs->md_thread, _spdk_blob_insert_cluster_msg, ctx); 3673 } 3674 3675 /* START spdk_blob_close */ 3676 3677 static void 3678 _spdk_blob_close_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3679 { 3680 struct spdk_blob *blob = cb_arg; 3681 3682 if (bserrno == 0) { 3683 blob->open_ref--; 3684 if (blob->open_ref == 0) { 3685 /* 3686 * Blobs with active.num_pages == 0 are deleted blobs. 3687 * these blobs are removed from the blob_store list 3688 * when the deletion process starts - so don't try to 3689 * remove them again. 3690 */ 3691 if (blob->active.num_pages > 0) { 3692 TAILQ_REMOVE(&blob->bs->blobs, blob, link); 3693 } 3694 _spdk_blob_free(blob); 3695 } 3696 } 3697 3698 spdk_bs_sequence_finish(seq, bserrno); 3699 } 3700 3701 void spdk_blob_close(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 3702 { 3703 struct spdk_bs_cpl cpl; 3704 spdk_bs_sequence_t *seq; 3705 3706 _spdk_blob_verify_md_op(blob); 3707 3708 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Closing blob %lu\n", blob->id); 3709 3710 if (blob->open_ref == 0) { 3711 cb_fn(cb_arg, -EBADF); 3712 return; 3713 } 3714 3715 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 3716 cpl.u.blob_basic.cb_fn = cb_fn; 3717 cpl.u.blob_basic.cb_arg = cb_arg; 3718 3719 seq = spdk_bs_sequence_start(blob->bs->md_channel, &cpl); 3720 if (!seq) { 3721 cb_fn(cb_arg, -ENOMEM); 3722 return; 3723 } 3724 3725 /* Sync metadata */ 3726 _spdk_blob_persist(seq, blob, _spdk_blob_close_cpl, blob); 3727 } 3728 3729 /* END spdk_blob_close */ 3730 3731 struct spdk_io_channel *spdk_bs_alloc_io_channel(struct spdk_blob_store *bs) 3732 { 3733 return spdk_get_io_channel(bs); 3734 } 3735 3736 void spdk_bs_free_io_channel(struct spdk_io_channel *channel) 3737 { 3738 spdk_put_io_channel(channel); 3739 } 3740 3741 void spdk_blob_io_unmap(struct spdk_blob *blob, struct spdk_io_channel *channel, 3742 uint64_t offset, uint64_t length, spdk_blob_op_complete cb_fn, void *cb_arg) 3743 { 3744 _spdk_blob_request_submit_op(blob, channel, NULL, offset, length, cb_fn, cb_arg, 3745 SPDK_BLOB_UNMAP); 3746 } 3747 3748 void spdk_blob_io_write_zeroes(struct spdk_blob *blob, struct spdk_io_channel *channel, 3749 uint64_t offset, uint64_t length, spdk_blob_op_complete cb_fn, void *cb_arg) 3750 { 3751 _spdk_blob_request_submit_op(blob, channel, NULL, offset, length, cb_fn, cb_arg, 3752 SPDK_BLOB_WRITE_ZEROES); 3753 } 3754 3755 void spdk_blob_io_write(struct spdk_blob *blob, struct spdk_io_channel *channel, 3756 void *payload, uint64_t offset, uint64_t length, 3757 spdk_blob_op_complete cb_fn, void *cb_arg) 3758 { 3759 _spdk_blob_request_submit_op(blob, channel, payload, offset, length, cb_fn, cb_arg, 3760 SPDK_BLOB_WRITE); 3761 } 3762 3763 void spdk_blob_io_read(struct spdk_blob *blob, struct spdk_io_channel *channel, 3764 void *payload, uint64_t offset, uint64_t length, 3765 spdk_blob_op_complete cb_fn, void *cb_arg) 3766 { 3767 _spdk_blob_request_submit_op(blob, channel, payload, offset, length, cb_fn, cb_arg, 3768 SPDK_BLOB_READ); 3769 } 3770 3771 void spdk_blob_io_writev(struct spdk_blob *blob, struct spdk_io_channel *channel, 3772 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 3773 spdk_blob_op_complete cb_fn, void *cb_arg) 3774 { 3775 _spdk_blob_request_submit_rw_iov(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg, false); 3776 } 3777 3778 void spdk_blob_io_readv(struct spdk_blob *blob, struct spdk_io_channel *channel, 3779 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 3780 spdk_blob_op_complete cb_fn, void *cb_arg) 3781 { 3782 _spdk_blob_request_submit_rw_iov(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg, true); 3783 } 3784 3785 void spdk_bs_io_unmap_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 3786 uint64_t offset, uint64_t length, spdk_blob_op_complete cb_fn, void *cb_arg) 3787 { 3788 spdk_blob_io_unmap(blob, channel, offset, length, cb_fn, cb_arg); 3789 } 3790 3791 void spdk_bs_io_write_zeroes_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 3792 uint64_t offset, uint64_t length, spdk_blob_op_complete cb_fn, void *cb_arg) 3793 { 3794 spdk_blob_io_write_zeroes(blob, channel, offset, length, cb_fn, cb_arg); 3795 } 3796 3797 void spdk_bs_io_write_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 3798 void *payload, uint64_t offset, uint64_t length, 3799 spdk_blob_op_complete cb_fn, void *cb_arg) 3800 { 3801 spdk_blob_io_write(blob, channel, payload, offset, length, cb_fn, cb_arg); 3802 } 3803 3804 void spdk_bs_io_read_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 3805 void *payload, uint64_t offset, uint64_t length, 3806 spdk_blob_op_complete cb_fn, void *cb_arg) 3807 { 3808 spdk_blob_io_read(blob, channel, payload, offset, length, cb_fn, cb_arg); 3809 } 3810 3811 void spdk_bs_io_writev_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 3812 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 3813 spdk_blob_op_complete cb_fn, void *cb_arg) 3814 { 3815 spdk_blob_io_writev(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg); 3816 } 3817 3818 void spdk_bs_io_readv_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 3819 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 3820 spdk_blob_op_complete cb_fn, void *cb_arg) 3821 { 3822 spdk_blob_io_readv(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg); 3823 } 3824 3825 struct spdk_bs_iter_ctx { 3826 int64_t page_num; 3827 struct spdk_blob_store *bs; 3828 3829 spdk_blob_op_with_handle_complete cb_fn; 3830 void *cb_arg; 3831 }; 3832 3833 static void 3834 _spdk_bs_iter_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 3835 { 3836 struct spdk_bs_iter_ctx *ctx = cb_arg; 3837 struct spdk_blob_store *bs = ctx->bs; 3838 spdk_blob_id id; 3839 3840 if (bserrno == 0) { 3841 ctx->cb_fn(ctx->cb_arg, _blob, bserrno); 3842 free(ctx); 3843 return; 3844 } 3845 3846 ctx->page_num++; 3847 ctx->page_num = spdk_bit_array_find_first_set(bs->used_blobids, ctx->page_num); 3848 if (ctx->page_num >= spdk_bit_array_capacity(bs->used_blobids)) { 3849 ctx->cb_fn(ctx->cb_arg, NULL, -ENOENT); 3850 free(ctx); 3851 return; 3852 } 3853 3854 id = _spdk_bs_page_to_blobid(ctx->page_num); 3855 3856 spdk_bs_open_blob(bs, id, _spdk_bs_iter_cpl, ctx); 3857 } 3858 3859 void 3860 spdk_bs_iter_first(struct spdk_blob_store *bs, 3861 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 3862 { 3863 struct spdk_bs_iter_ctx *ctx; 3864 3865 ctx = calloc(1, sizeof(*ctx)); 3866 if (!ctx) { 3867 cb_fn(cb_arg, NULL, -ENOMEM); 3868 return; 3869 } 3870 3871 ctx->page_num = -1; 3872 ctx->bs = bs; 3873 ctx->cb_fn = cb_fn; 3874 ctx->cb_arg = cb_arg; 3875 3876 _spdk_bs_iter_cpl(ctx, NULL, -1); 3877 } 3878 3879 static void 3880 _spdk_bs_iter_close_cpl(void *cb_arg, int bserrno) 3881 { 3882 struct spdk_bs_iter_ctx *ctx = cb_arg; 3883 3884 _spdk_bs_iter_cpl(ctx, NULL, -1); 3885 } 3886 3887 void 3888 spdk_bs_iter_next(struct spdk_blob_store *bs, struct spdk_blob *blob, 3889 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 3890 { 3891 struct spdk_bs_iter_ctx *ctx; 3892 3893 assert(blob != NULL); 3894 3895 ctx = calloc(1, sizeof(*ctx)); 3896 if (!ctx) { 3897 cb_fn(cb_arg, NULL, -ENOMEM); 3898 return; 3899 } 3900 3901 ctx->page_num = _spdk_bs_blobid_to_page(blob->id); 3902 ctx->bs = bs; 3903 ctx->cb_fn = cb_fn; 3904 ctx->cb_arg = cb_arg; 3905 3906 /* Close the existing blob */ 3907 spdk_blob_close(blob, _spdk_bs_iter_close_cpl, ctx); 3908 } 3909 3910 static int 3911 _spdk_blob_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 3912 uint16_t value_len, bool internal) 3913 { 3914 struct spdk_xattr_tailq *xattrs; 3915 struct spdk_xattr *xattr; 3916 3917 _spdk_blob_verify_md_op(blob); 3918 3919 if (blob->md_ro) { 3920 return -EPERM; 3921 } 3922 3923 if (internal) { 3924 xattrs = &blob->xattrs_internal; 3925 blob->invalid_flags |= SPDK_BLOB_INTERNAL_XATTR; 3926 } else { 3927 xattrs = &blob->xattrs; 3928 } 3929 3930 TAILQ_FOREACH(xattr, xattrs, link) { 3931 if (!strcmp(name, xattr->name)) { 3932 free(xattr->value); 3933 xattr->value_len = value_len; 3934 xattr->value = malloc(value_len); 3935 memcpy(xattr->value, value, value_len); 3936 3937 blob->state = SPDK_BLOB_STATE_DIRTY; 3938 3939 return 0; 3940 } 3941 } 3942 3943 xattr = calloc(1, sizeof(*xattr)); 3944 if (!xattr) { 3945 return -1; 3946 } 3947 xattr->name = strdup(name); 3948 xattr->value_len = value_len; 3949 xattr->value = malloc(value_len); 3950 memcpy(xattr->value, value, value_len); 3951 TAILQ_INSERT_TAIL(xattrs, xattr, link); 3952 3953 blob->state = SPDK_BLOB_STATE_DIRTY; 3954 3955 return 0; 3956 } 3957 3958 int 3959 spdk_blob_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 3960 uint16_t value_len) 3961 { 3962 return _spdk_blob_set_xattr(blob, name, value, value_len, false); 3963 } 3964 3965 static int 3966 _spdk_blob_remove_xattr(struct spdk_blob *blob, const char *name, bool internal) 3967 { 3968 struct spdk_xattr_tailq *xattrs; 3969 struct spdk_xattr *xattr; 3970 3971 _spdk_blob_verify_md_op(blob); 3972 3973 if (blob->md_ro) { 3974 return -EPERM; 3975 } 3976 xattrs = internal ? &blob->xattrs_internal : &blob->xattrs; 3977 3978 TAILQ_FOREACH(xattr, xattrs, link) { 3979 if (!strcmp(name, xattr->name)) { 3980 TAILQ_REMOVE(xattrs, xattr, link); 3981 free(xattr->value); 3982 free(xattr->name); 3983 free(xattr); 3984 3985 if (internal && TAILQ_EMPTY(&blob->xattrs_internal)) { 3986 blob->invalid_flags &= ~SPDK_BLOB_INTERNAL_XATTR; 3987 } 3988 blob->state = SPDK_BLOB_STATE_DIRTY; 3989 3990 return 0; 3991 } 3992 } 3993 3994 return -ENOENT; 3995 } 3996 3997 int 3998 spdk_blob_remove_xattr(struct spdk_blob *blob, const char *name) 3999 { 4000 return _spdk_blob_remove_xattr(blob, name, false); 4001 } 4002 4003 static int 4004 _spdk_blob_get_xattr_value(struct spdk_blob *blob, const char *name, 4005 const void **value, size_t *value_len, bool internal) 4006 { 4007 struct spdk_xattr *xattr; 4008 struct spdk_xattr_tailq *xattrs; 4009 4010 _spdk_blob_verify_md_op(blob); 4011 4012 xattrs = internal ? &blob->xattrs_internal : &blob->xattrs; 4013 4014 TAILQ_FOREACH(xattr, xattrs, link) { 4015 if (!strcmp(name, xattr->name)) { 4016 *value = xattr->value; 4017 *value_len = xattr->value_len; 4018 return 0; 4019 } 4020 } 4021 return -ENOENT; 4022 } 4023 4024 int 4025 spdk_blob_get_xattr_value(struct spdk_blob *blob, const char *name, 4026 const void **value, size_t *value_len) 4027 { 4028 return _spdk_blob_get_xattr_value(blob, name, value, value_len, false); 4029 } 4030 4031 struct spdk_xattr_names { 4032 uint32_t count; 4033 const char *names[0]; 4034 }; 4035 4036 static int 4037 _spdk_blob_get_xattr_names(struct spdk_xattr_tailq *xattrs, struct spdk_xattr_names **names) 4038 { 4039 struct spdk_xattr *xattr; 4040 int count = 0; 4041 4042 TAILQ_FOREACH(xattr, xattrs, link) { 4043 count++; 4044 } 4045 4046 *names = calloc(1, sizeof(struct spdk_xattr_names) + count * sizeof(char *)); 4047 if (*names == NULL) { 4048 return -ENOMEM; 4049 } 4050 4051 TAILQ_FOREACH(xattr, xattrs, link) { 4052 (*names)->names[(*names)->count++] = xattr->name; 4053 } 4054 4055 return 0; 4056 } 4057 4058 int 4059 spdk_blob_get_xattr_names(struct spdk_blob *blob, struct spdk_xattr_names **names) 4060 { 4061 _spdk_blob_verify_md_op(blob); 4062 4063 return _spdk_blob_get_xattr_names(&blob->xattrs, names); 4064 } 4065 4066 uint32_t 4067 spdk_xattr_names_get_count(struct spdk_xattr_names *names) 4068 { 4069 assert(names != NULL); 4070 4071 return names->count; 4072 } 4073 4074 const char * 4075 spdk_xattr_names_get_name(struct spdk_xattr_names *names, uint32_t index) 4076 { 4077 if (index >= names->count) { 4078 return NULL; 4079 } 4080 4081 return names->names[index]; 4082 } 4083 4084 void 4085 spdk_xattr_names_free(struct spdk_xattr_names *names) 4086 { 4087 free(names); 4088 } 4089 4090 struct spdk_bs_type 4091 spdk_bs_get_bstype(struct spdk_blob_store *bs) 4092 { 4093 return bs->bstype; 4094 } 4095 4096 void 4097 spdk_bs_set_bstype(struct spdk_blob_store *bs, struct spdk_bs_type bstype) 4098 { 4099 memcpy(&bs->bstype, &bstype, sizeof(bstype)); 4100 } 4101 4102 SPDK_LOG_REGISTER_COMPONENT("blob", SPDK_LOG_BLOB) 4103