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_cpl cb_fn; 887 void *cb_arg; 888 }; 889 890 static void 891 _spdk_blob_persist_complete(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 892 { 893 struct spdk_blob_persist_ctx *ctx = cb_arg; 894 struct spdk_blob *blob = ctx->blob; 895 896 if (bserrno == 0) { 897 _spdk_blob_mark_clean(blob); 898 } 899 900 /* Call user callback */ 901 ctx->cb_fn(seq, ctx->cb_arg, bserrno); 902 903 /* Free the memory */ 904 spdk_dma_free(ctx->pages); 905 free(ctx); 906 } 907 908 static void 909 _spdk_blob_persist_unmap_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 910 { 911 struct spdk_blob_persist_ctx *ctx = cb_arg; 912 struct spdk_blob *blob = ctx->blob; 913 struct spdk_blob_store *bs = blob->bs; 914 void *tmp; 915 size_t i; 916 917 /* Release all clusters that were truncated */ 918 for (i = blob->active.num_clusters; i < blob->active.cluster_array_size; i++) { 919 uint32_t cluster_num = _spdk_bs_lba_to_cluster(bs, blob->active.clusters[i]); 920 921 /* Nothing to release if it was not allocated */ 922 if (blob->active.clusters[i] != 0) { 923 _spdk_bs_release_cluster(bs, cluster_num); 924 } 925 } 926 927 if (blob->active.num_clusters == 0) { 928 free(blob->active.clusters); 929 blob->active.clusters = NULL; 930 blob->active.cluster_array_size = 0; 931 } else { 932 tmp = realloc(blob->active.clusters, sizeof(uint64_t) * blob->active.num_clusters); 933 assert(tmp != NULL); 934 blob->active.clusters = tmp; 935 blob->active.cluster_array_size = blob->active.num_clusters; 936 } 937 938 _spdk_blob_persist_complete(seq, ctx, bserrno); 939 } 940 941 static void 942 _spdk_blob_persist_unmap_clusters(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 943 { 944 struct spdk_blob_persist_ctx *ctx = cb_arg; 945 struct spdk_blob *blob = ctx->blob; 946 struct spdk_blob_store *bs = blob->bs; 947 spdk_bs_batch_t *batch; 948 size_t i; 949 uint64_t lba; 950 uint32_t lba_count; 951 952 /* Clusters don't move around in blobs. The list shrinks or grows 953 * at the end, but no changes ever occur in the middle of the list. 954 */ 955 956 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_unmap_clusters_cpl, ctx); 957 958 /* Unmap all clusters that were truncated */ 959 lba = 0; 960 lba_count = 0; 961 for (i = blob->active.num_clusters; i < blob->active.cluster_array_size; i++) { 962 uint64_t next_lba = blob->active.clusters[i]; 963 uint32_t next_lba_count = _spdk_bs_cluster_to_lba(bs, 1); 964 965 if (next_lba > 0 && (lba + lba_count) == next_lba) { 966 /* This cluster is contiguous with the previous one. */ 967 lba_count += next_lba_count; 968 continue; 969 } 970 971 /* This cluster is not contiguous with the previous one. */ 972 973 /* If a run of LBAs previously existing, send them 974 * as an unmap. 975 */ 976 if (lba_count > 0) { 977 spdk_bs_batch_unmap_dev(batch, lba, lba_count); 978 } 979 980 /* Start building the next batch */ 981 lba = next_lba; 982 if (next_lba > 0) { 983 lba_count = next_lba_count; 984 } else { 985 lba_count = 0; 986 } 987 } 988 989 /* If we ended with a contiguous set of LBAs, send the unmap now */ 990 if (lba_count > 0) { 991 spdk_bs_batch_unmap_dev(batch, lba, lba_count); 992 } 993 994 spdk_bs_batch_close(batch); 995 } 996 997 static void 998 _spdk_blob_persist_zero_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 999 { 1000 struct spdk_blob_persist_ctx *ctx = cb_arg; 1001 struct spdk_blob *blob = ctx->blob; 1002 struct spdk_blob_store *bs = blob->bs; 1003 size_t i; 1004 1005 /* This loop starts at 1 because the first page is special and handled 1006 * below. The pages (except the first) are never written in place, 1007 * so any pages in the clean list must be zeroed. 1008 */ 1009 for (i = 1; i < blob->clean.num_pages; i++) { 1010 spdk_bit_array_clear(bs->used_md_pages, blob->clean.pages[i]); 1011 } 1012 1013 if (blob->active.num_pages == 0) { 1014 uint32_t page_num; 1015 1016 page_num = _spdk_bs_blobid_to_page(blob->id); 1017 spdk_bit_array_clear(bs->used_md_pages, page_num); 1018 } 1019 1020 /* Move on to unmapping clusters */ 1021 _spdk_blob_persist_unmap_clusters(seq, ctx, 0); 1022 } 1023 1024 static void 1025 _spdk_blob_persist_zero_pages(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1026 { 1027 struct spdk_blob_persist_ctx *ctx = cb_arg; 1028 struct spdk_blob *blob = ctx->blob; 1029 struct spdk_blob_store *bs = blob->bs; 1030 uint64_t lba; 1031 uint32_t lba_count; 1032 spdk_bs_batch_t *batch; 1033 size_t i; 1034 1035 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_zero_pages_cpl, ctx); 1036 1037 lba_count = _spdk_bs_byte_to_lba(bs, SPDK_BS_PAGE_SIZE); 1038 1039 /* This loop starts at 1 because the first page is special and handled 1040 * below. The pages (except the first) are never written in place, 1041 * so any pages in the clean list must be zeroed. 1042 */ 1043 for (i = 1; i < blob->clean.num_pages; i++) { 1044 lba = _spdk_bs_page_to_lba(bs, bs->md_start + blob->clean.pages[i]); 1045 1046 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1047 } 1048 1049 /* The first page will only be zeroed if this is a delete. */ 1050 if (blob->active.num_pages == 0) { 1051 uint32_t page_num; 1052 1053 /* The first page in the metadata goes where the blobid indicates */ 1054 page_num = _spdk_bs_blobid_to_page(blob->id); 1055 lba = _spdk_bs_page_to_lba(bs, bs->md_start + page_num); 1056 1057 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1058 } 1059 1060 spdk_bs_batch_close(batch); 1061 } 1062 1063 static void 1064 _spdk_blob_persist_write_page_root(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1065 { 1066 struct spdk_blob_persist_ctx *ctx = cb_arg; 1067 struct spdk_blob *blob = ctx->blob; 1068 struct spdk_blob_store *bs = blob->bs; 1069 uint64_t lba; 1070 uint32_t lba_count; 1071 struct spdk_blob_md_page *page; 1072 1073 if (blob->active.num_pages == 0) { 1074 /* Move on to the next step */ 1075 _spdk_blob_persist_zero_pages(seq, ctx, 0); 1076 return; 1077 } 1078 1079 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(*page)); 1080 1081 page = &ctx->pages[0]; 1082 /* The first page in the metadata goes where the blobid indicates */ 1083 lba = _spdk_bs_page_to_lba(bs, bs->md_start + _spdk_bs_blobid_to_page(blob->id)); 1084 1085 spdk_bs_sequence_write_dev(seq, page, lba, lba_count, 1086 _spdk_blob_persist_zero_pages, ctx); 1087 } 1088 1089 static void 1090 _spdk_blob_persist_write_page_chain(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1091 { 1092 struct spdk_blob_persist_ctx *ctx = cb_arg; 1093 struct spdk_blob *blob = ctx->blob; 1094 struct spdk_blob_store *bs = blob->bs; 1095 uint64_t lba; 1096 uint32_t lba_count; 1097 struct spdk_blob_md_page *page; 1098 spdk_bs_batch_t *batch; 1099 size_t i; 1100 1101 /* Clusters don't move around in blobs. The list shrinks or grows 1102 * at the end, but no changes ever occur in the middle of the list. 1103 */ 1104 1105 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(*page)); 1106 1107 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_write_page_root, ctx); 1108 1109 /* This starts at 1. The root page is not written until 1110 * all of the others are finished 1111 */ 1112 for (i = 1; i < blob->active.num_pages; i++) { 1113 page = &ctx->pages[i]; 1114 assert(page->sequence_num == i); 1115 1116 lba = _spdk_bs_page_to_lba(bs, bs->md_start + blob->active.pages[i]); 1117 1118 spdk_bs_batch_write_dev(batch, page, lba, lba_count); 1119 } 1120 1121 spdk_bs_batch_close(batch); 1122 } 1123 1124 static int 1125 _spdk_resize_blob(struct spdk_blob *blob, uint64_t sz) 1126 { 1127 uint64_t i; 1128 uint64_t *tmp; 1129 uint64_t lfc; /* lowest free cluster */ 1130 uint64_t num_clusters; 1131 struct spdk_blob_store *bs; 1132 1133 bs = blob->bs; 1134 1135 _spdk_blob_verify_md_op(blob); 1136 1137 if (blob->active.num_clusters == sz) { 1138 return 0; 1139 } 1140 1141 if (blob->active.num_clusters < blob->active.cluster_array_size) { 1142 /* If this blob was resized to be larger, then smaller, then 1143 * larger without syncing, then the cluster array already 1144 * contains spare assigned clusters we can use. 1145 */ 1146 num_clusters = spdk_min(blob->active.cluster_array_size, 1147 sz); 1148 } else { 1149 num_clusters = blob->active.num_clusters; 1150 } 1151 1152 /* Do two passes - one to verify that we can obtain enough clusters 1153 * and another to actually claim them. 1154 */ 1155 1156 if (spdk_blob_is_thin_provisioned(blob) == false) { 1157 lfc = 0; 1158 for (i = num_clusters; i < sz; i++) { 1159 lfc = spdk_bit_array_find_first_clear(bs->used_clusters, lfc); 1160 if (lfc >= bs->total_clusters) { 1161 /* No more free clusters. Cannot satisfy the request */ 1162 return -ENOSPC; 1163 } 1164 lfc++; 1165 } 1166 } 1167 1168 if (sz > num_clusters) { 1169 /* Expand the cluster array if necessary. 1170 * We only shrink the array when persisting. 1171 */ 1172 tmp = realloc(blob->active.clusters, sizeof(uint64_t) * sz); 1173 if (sz > 0 && tmp == NULL) { 1174 return -ENOMEM; 1175 } 1176 memset(tmp + blob->active.cluster_array_size, 0, 1177 sizeof(uint64_t) * (sz - blob->active.cluster_array_size)); 1178 blob->active.clusters = tmp; 1179 blob->active.cluster_array_size = sz; 1180 } 1181 1182 blob->state = SPDK_BLOB_STATE_DIRTY; 1183 1184 if (spdk_blob_is_thin_provisioned(blob) == false) { 1185 lfc = 0; 1186 for (i = num_clusters; i < sz; i++) { 1187 _spdk_bs_allocate_cluster(blob, i, &lfc, true); 1188 lfc++; 1189 } 1190 } 1191 1192 blob->active.num_clusters = sz; 1193 1194 return 0; 1195 } 1196 1197 /* Write a blob to disk */ 1198 static void 1199 _spdk_blob_persist(spdk_bs_sequence_t *seq, struct spdk_blob *blob, 1200 spdk_bs_sequence_cpl cb_fn, void *cb_arg) 1201 { 1202 struct spdk_blob_persist_ctx *ctx; 1203 int rc; 1204 uint64_t i; 1205 uint32_t page_num; 1206 struct spdk_blob_store *bs; 1207 1208 _spdk_blob_verify_md_op(blob); 1209 1210 if (blob->state == SPDK_BLOB_STATE_CLEAN) { 1211 cb_fn(seq, cb_arg, 0); 1212 return; 1213 } 1214 1215 bs = blob->bs; 1216 1217 ctx = calloc(1, sizeof(*ctx)); 1218 if (!ctx) { 1219 cb_fn(seq, cb_arg, -ENOMEM); 1220 return; 1221 } 1222 ctx->blob = blob; 1223 ctx->cb_fn = cb_fn; 1224 ctx->cb_arg = cb_arg; 1225 1226 if (blob->active.num_pages == 0) { 1227 /* This is the signal that the blob should be deleted. 1228 * Immediately jump to the clean up routine. */ 1229 assert(blob->clean.num_pages > 0); 1230 ctx->idx = blob->clean.num_pages - 1; 1231 blob->state = SPDK_BLOB_STATE_CLEAN; 1232 _spdk_blob_persist_zero_pages(seq, ctx, 0); 1233 return; 1234 1235 } 1236 1237 /* Generate the new metadata */ 1238 rc = _spdk_blob_serialize(blob, &ctx->pages, &blob->active.num_pages); 1239 if (rc < 0) { 1240 _spdk_blob_persist_complete(seq, ctx, rc); 1241 return; 1242 } 1243 1244 assert(blob->active.num_pages >= 1); 1245 1246 /* Resize the cache of page indices */ 1247 blob->active.pages = realloc(blob->active.pages, 1248 blob->active.num_pages * sizeof(*blob->active.pages)); 1249 if (!blob->active.pages) { 1250 _spdk_blob_persist_complete(seq, ctx, -ENOMEM); 1251 return; 1252 } 1253 1254 /* Assign this metadata to pages. This requires two passes - 1255 * one to verify that there are enough pages and a second 1256 * to actually claim them. */ 1257 page_num = 0; 1258 /* Note that this loop starts at one. The first page location is fixed by the blobid. */ 1259 for (i = 1; i < blob->active.num_pages; i++) { 1260 page_num = spdk_bit_array_find_first_clear(bs->used_md_pages, page_num); 1261 if (page_num >= spdk_bit_array_capacity(bs->used_md_pages)) { 1262 _spdk_blob_persist_complete(seq, ctx, -ENOMEM); 1263 return; 1264 } 1265 page_num++; 1266 } 1267 1268 page_num = 0; 1269 blob->active.pages[0] = _spdk_bs_blobid_to_page(blob->id); 1270 for (i = 1; i < blob->active.num_pages; i++) { 1271 page_num = spdk_bit_array_find_first_clear(bs->used_md_pages, page_num); 1272 ctx->pages[i - 1].next = page_num; 1273 /* Now that previous metadata page is complete, calculate the crc for it. */ 1274 ctx->pages[i - 1].crc = _spdk_blob_md_page_calc_crc(&ctx->pages[i - 1]); 1275 blob->active.pages[i] = page_num; 1276 spdk_bit_array_set(bs->used_md_pages, page_num); 1277 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Claiming page %u for blob %lu\n", page_num, blob->id); 1278 page_num++; 1279 } 1280 ctx->pages[i - 1].crc = _spdk_blob_md_page_calc_crc(&ctx->pages[i - 1]); 1281 /* Start writing the metadata from last page to first */ 1282 ctx->idx = blob->active.num_pages - 1; 1283 blob->state = SPDK_BLOB_STATE_CLEAN; 1284 _spdk_blob_persist_write_page_chain(seq, ctx, 0); 1285 } 1286 1287 struct spdk_blob_copy_cluster_ctx { 1288 struct spdk_blob *blob; 1289 uint8_t *buf; 1290 uint64_t page; 1291 uint64_t new_cluster; 1292 spdk_bs_sequence_t *seq; 1293 }; 1294 1295 static void 1296 _spdk_blob_allocate_and_copy_cluster_cpl(void *cb_arg, int bserrno) 1297 { 1298 struct spdk_blob_copy_cluster_ctx *ctx = cb_arg; 1299 struct spdk_bs_request_set *set = (struct spdk_bs_request_set *)ctx->seq; 1300 TAILQ_HEAD(, spdk_bs_request_set) requests; 1301 spdk_bs_user_op_t *op; 1302 1303 TAILQ_INIT(&requests); 1304 TAILQ_SWAP(&set->channel->need_cluster_alloc, &requests, spdk_bs_request_set, link); 1305 1306 while (!TAILQ_EMPTY(&requests)) { 1307 op = TAILQ_FIRST(&requests); 1308 TAILQ_REMOVE(&requests, op, link); 1309 if (bserrno == 0) { 1310 spdk_bs_user_op_execute(op); 1311 } else { 1312 spdk_bs_user_op_abort(op); 1313 } 1314 } 1315 1316 spdk_dma_free(ctx->buf); 1317 free(ctx); 1318 } 1319 1320 static void 1321 _spdk_blob_insert_cluster_cpl(void *cb_arg, int bserrno) 1322 { 1323 struct spdk_blob_copy_cluster_ctx *ctx = cb_arg; 1324 1325 if (bserrno) { 1326 uint32_t cluster_number; 1327 1328 if (bserrno == -EEXIST) { 1329 /* The metadata insert failed because another thread 1330 * allocated the cluster first. Free our cluster 1331 * but continue without error. */ 1332 bserrno = 0; 1333 } 1334 1335 cluster_number = _spdk_bs_page_to_cluster(ctx->blob->bs, ctx->page); 1336 _spdk_bs_release_cluster(ctx->blob->bs, cluster_number); 1337 } 1338 1339 spdk_bs_sequence_finish(ctx->seq, bserrno); 1340 } 1341 1342 static void 1343 _spdk_blob_write_copy_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1344 { 1345 struct spdk_blob_copy_cluster_ctx *ctx = cb_arg; 1346 uint32_t cluster_number; 1347 1348 if (bserrno) { 1349 /* The write failed, so jump to the final completion handler */ 1350 spdk_bs_sequence_finish(seq, bserrno); 1351 return; 1352 } 1353 1354 cluster_number = _spdk_bs_page_to_cluster(ctx->blob->bs, ctx->page); 1355 1356 _spdk_blob_insert_cluster_on_md_thread(ctx->blob, cluster_number, ctx->new_cluster, 1357 _spdk_blob_insert_cluster_cpl, ctx); 1358 } 1359 1360 static void 1361 _spdk_blob_write_copy(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1362 { 1363 struct spdk_blob_copy_cluster_ctx *ctx = cb_arg; 1364 1365 if (bserrno != 0) { 1366 /* The read failed, so jump to the final completion handler */ 1367 spdk_bs_sequence_finish(seq, bserrno); 1368 return; 1369 } 1370 1371 /* Write whole cluster */ 1372 spdk_bs_sequence_write_dev(seq, ctx->buf, 1373 _spdk_bs_cluster_to_lba(ctx->blob->bs, ctx->new_cluster), 1374 _spdk_bs_cluster_to_lba(ctx->blob->bs, 1), 1375 _spdk_blob_write_copy_cpl, ctx); 1376 } 1377 1378 static void 1379 _spdk_bs_allocate_and_copy_cluster(struct spdk_blob *blob, 1380 struct spdk_io_channel *_ch, 1381 uint64_t offset, spdk_bs_user_op_t *op) 1382 { 1383 struct spdk_bs_cpl cpl; 1384 struct spdk_bs_channel *ch; 1385 struct spdk_blob_copy_cluster_ctx *ctx; 1386 uint32_t cluster_start_page; 1387 uint32_t cluster_number; 1388 int rc; 1389 1390 ch = spdk_io_channel_get_ctx(_ch); 1391 1392 if (!TAILQ_EMPTY(&ch->need_cluster_alloc)) { 1393 /* There are already operations pending. Queue this user op 1394 * and return because it will be re-executed when the outstanding 1395 * cluster allocation completes. */ 1396 TAILQ_INSERT_TAIL(&ch->need_cluster_alloc, op, link); 1397 return; 1398 } 1399 1400 /* Round the page offset down to the first page in the cluster */ 1401 cluster_start_page = _spdk_bs_page_to_cluster_start(blob, offset); 1402 1403 /* Calculate which index in the metadata cluster array the corresponding 1404 * cluster is supposed to be at. */ 1405 cluster_number = _spdk_bs_page_to_cluster(blob->bs, cluster_start_page); 1406 1407 ctx = calloc(1, sizeof(*ctx)); 1408 if (!ctx) { 1409 spdk_bs_user_op_abort(op); 1410 return; 1411 } 1412 1413 assert(blob->bs->cluster_sz % blob->back_bs_dev->blocklen == 0); 1414 1415 ctx->blob = blob; 1416 ctx->page = cluster_start_page; 1417 1418 ctx->buf = spdk_dma_malloc(blob->bs->cluster_sz, blob->back_bs_dev->blocklen, NULL); 1419 if (!ctx->buf) { 1420 SPDK_ERRLOG("DMA allocation for cluster of size = %" PRIu32 " failed.\n", 1421 blob->bs->cluster_sz); 1422 free(ctx); 1423 spdk_bs_user_op_abort(op); 1424 return; 1425 } 1426 1427 rc = _spdk_bs_allocate_cluster(blob, cluster_number, &ctx->new_cluster, false); 1428 if (rc != 0) { 1429 spdk_dma_free(ctx->buf); 1430 free(ctx); 1431 spdk_bs_user_op_abort(op); 1432 return; 1433 } 1434 1435 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1436 cpl.u.blob_basic.cb_fn = _spdk_blob_allocate_and_copy_cluster_cpl; 1437 cpl.u.blob_basic.cb_arg = ctx; 1438 1439 ctx->seq = spdk_bs_sequence_start(_ch, &cpl); 1440 if (!ctx->seq) { 1441 _spdk_bs_release_cluster(blob->bs, ctx->new_cluster); 1442 spdk_dma_free(ctx->buf); 1443 free(ctx); 1444 spdk_bs_user_op_abort(op); 1445 return; 1446 } 1447 1448 /* Queue the user op to block other incoming operations */ 1449 TAILQ_INSERT_TAIL(&ch->need_cluster_alloc, op, link); 1450 1451 /* Read cluster from backing device */ 1452 spdk_bs_sequence_read_bs_dev(ctx->seq, blob->back_bs_dev, ctx->buf, 1453 _spdk_bs_dev_page_to_lba(blob->back_bs_dev, cluster_start_page), 1454 _spdk_bs_dev_byte_to_lba(blob->back_bs_dev, blob->bs->cluster_sz), 1455 _spdk_blob_write_copy, ctx); 1456 } 1457 1458 static void 1459 _spdk_blob_calculate_lba_and_lba_count(struct spdk_blob *blob, uint64_t page, uint64_t length, 1460 uint64_t *lba, uint32_t *lba_count) 1461 { 1462 *lba_count = _spdk_bs_page_to_lba(blob->bs, length); 1463 1464 if (!_spdk_bs_page_is_allocated(blob, page)) { 1465 assert(blob->back_bs_dev != NULL); 1466 *lba = _spdk_bs_dev_page_to_lba(blob->back_bs_dev, page); 1467 *lba_count = _spdk_bs_blob_lba_to_back_dev_lba(blob, *lba_count); 1468 } else { 1469 *lba = _spdk_bs_blob_page_to_lba(blob, page); 1470 } 1471 } 1472 1473 static void 1474 _spdk_blob_request_submit_op_split(struct spdk_io_channel *ch, struct spdk_blob *blob, 1475 void *payload, uint64_t offset, uint64_t length, 1476 spdk_blob_op_complete cb_fn, void *cb_arg, enum spdk_blob_op_type op_type) 1477 { 1478 spdk_bs_batch_t *batch; 1479 struct spdk_bs_cpl cpl; 1480 uint64_t op_length; 1481 uint8_t *buf; 1482 1483 assert(blob != NULL); 1484 1485 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1486 cpl.u.blob_basic.cb_fn = cb_fn; 1487 cpl.u.blob_basic.cb_arg = cb_arg; 1488 1489 batch = spdk_bs_batch_open(ch, &cpl); 1490 if (!batch) { 1491 cb_fn(cb_arg, -ENOMEM); 1492 return; 1493 } 1494 1495 buf = payload; 1496 while (length > 0) { 1497 op_length = spdk_min(length, _spdk_bs_num_pages_to_cluster_boundary(blob, offset)); 1498 1499 switch (op_type) { 1500 case SPDK_BLOB_READ: 1501 spdk_bs_batch_read_blob(batch, blob, buf, offset, op_length); 1502 break; 1503 case SPDK_BLOB_WRITE: 1504 spdk_bs_batch_write_blob(batch, blob, buf, offset, op_length); 1505 break; 1506 case SPDK_BLOB_UNMAP: 1507 spdk_bs_batch_unmap_blob(batch, blob, offset, op_length); 1508 break; 1509 case SPDK_BLOB_WRITE_ZEROES: 1510 spdk_bs_batch_write_zeroes_blob(batch, blob, offset, op_length); 1511 break; 1512 case SPDK_BLOB_READV: 1513 case SPDK_BLOB_WRITEV: 1514 SPDK_ERRLOG("readv/write not valid for %s\n", __func__); 1515 break; 1516 } 1517 1518 length -= op_length; 1519 offset += op_length; 1520 if (op_type == SPDK_BLOB_WRITE || op_type == SPDK_BLOB_READ) { 1521 buf += op_length * SPDK_BS_PAGE_SIZE; 1522 } 1523 } 1524 1525 spdk_bs_batch_close(batch); 1526 } 1527 1528 static void 1529 _spdk_blob_request_submit_op_single(struct spdk_io_channel *_ch, struct spdk_blob *blob, 1530 void *payload, uint64_t offset, uint64_t length, 1531 spdk_blob_op_complete cb_fn, void *cb_arg, enum spdk_blob_op_type op_type) 1532 { 1533 struct spdk_bs_cpl cpl; 1534 uint64_t lba; 1535 uint32_t lba_count; 1536 1537 assert(blob != NULL); 1538 1539 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1540 cpl.u.blob_basic.cb_fn = cb_fn; 1541 cpl.u.blob_basic.cb_arg = cb_arg; 1542 1543 _spdk_blob_calculate_lba_and_lba_count(blob, offset, length, &lba, &lba_count); 1544 1545 switch (op_type) { 1546 case SPDK_BLOB_READ: { 1547 spdk_bs_batch_t *batch; 1548 1549 batch = spdk_bs_batch_open(_ch, &cpl); 1550 if (!batch) { 1551 cb_fn(cb_arg, -ENOMEM); 1552 return; 1553 } 1554 1555 if (_spdk_bs_page_is_allocated(blob, offset)) { 1556 /* Read from the blob */ 1557 spdk_bs_batch_read_dev(batch, payload, lba, lba_count); 1558 } else { 1559 /* Read from the backing block device */ 1560 spdk_bs_batch_read_bs_dev(batch, blob->back_bs_dev, payload, lba, lba_count); 1561 } 1562 1563 spdk_bs_batch_close(batch); 1564 break; 1565 } 1566 case SPDK_BLOB_WRITE: 1567 case SPDK_BLOB_WRITE_ZEROES: { 1568 if (_spdk_bs_page_is_allocated(blob, offset)) { 1569 /* Write to the blob */ 1570 spdk_bs_batch_t *batch; 1571 1572 batch = spdk_bs_batch_open(_ch, &cpl); 1573 if (!batch) { 1574 cb_fn(cb_arg, -ENOMEM); 1575 return; 1576 } 1577 1578 if (op_type == SPDK_BLOB_WRITE) { 1579 spdk_bs_batch_write_dev(batch, payload, lba, lba_count); 1580 } else { 1581 spdk_bs_batch_write_zeroes_dev(batch, lba, lba_count); 1582 } 1583 1584 spdk_bs_batch_close(batch); 1585 } else { 1586 /* Queue this operation and allocate the cluster */ 1587 spdk_bs_user_op_t *op; 1588 1589 op = spdk_bs_user_op_alloc(_ch, &cpl, op_type, blob, payload, 0, offset, length); 1590 if (!op) { 1591 cb_fn(cb_arg, -ENOMEM); 1592 return; 1593 } 1594 1595 _spdk_bs_allocate_and_copy_cluster(blob, _ch, offset, op); 1596 } 1597 break; 1598 } 1599 case SPDK_BLOB_UNMAP: { 1600 spdk_bs_batch_t *batch; 1601 1602 batch = spdk_bs_batch_open(_ch, &cpl); 1603 if (!batch) { 1604 cb_fn(cb_arg, -ENOMEM); 1605 return; 1606 } 1607 1608 if (_spdk_bs_page_is_allocated(blob, offset)) { 1609 spdk_bs_batch_unmap_dev(batch, lba, lba_count); 1610 } 1611 1612 spdk_bs_batch_close(batch); 1613 break; 1614 } 1615 case SPDK_BLOB_READV: 1616 case SPDK_BLOB_WRITEV: 1617 SPDK_ERRLOG("readv/write not valid\n"); 1618 cb_fn(cb_arg, -EINVAL); 1619 break; 1620 } 1621 } 1622 1623 static void 1624 _spdk_blob_request_submit_op(struct spdk_blob *blob, struct spdk_io_channel *_channel, 1625 void *payload, uint64_t offset, uint64_t length, 1626 spdk_blob_op_complete cb_fn, void *cb_arg, enum spdk_blob_op_type op_type) 1627 { 1628 assert(blob != NULL); 1629 1630 if (blob->data_ro && op_type != SPDK_BLOB_READ) { 1631 cb_fn(cb_arg, -EPERM); 1632 return; 1633 } 1634 1635 if (offset + length > blob->active.num_clusters * blob->bs->pages_per_cluster) { 1636 cb_fn(cb_arg, -EINVAL); 1637 return; 1638 } 1639 1640 if (length <= _spdk_bs_num_pages_to_cluster_boundary(blob, offset)) { 1641 _spdk_blob_request_submit_op_single(_channel, blob, payload, offset, length, 1642 cb_fn, cb_arg, op_type); 1643 } else { 1644 _spdk_blob_request_submit_op_split(_channel, blob, payload, offset, length, 1645 cb_fn, cb_arg, op_type); 1646 } 1647 } 1648 1649 struct rw_iov_ctx { 1650 struct spdk_blob *blob; 1651 struct spdk_io_channel *channel; 1652 spdk_blob_op_complete cb_fn; 1653 void *cb_arg; 1654 bool read; 1655 int iovcnt; 1656 struct iovec *orig_iov; 1657 uint64_t page_offset; 1658 uint64_t pages_remaining; 1659 uint64_t pages_done; 1660 struct iovec iov[0]; 1661 }; 1662 1663 static void 1664 _spdk_rw_iov_done(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1665 { 1666 assert(cb_arg == NULL); 1667 spdk_bs_sequence_finish(seq, bserrno); 1668 } 1669 1670 static void 1671 _spdk_rw_iov_split_next(void *cb_arg, int bserrno) 1672 { 1673 struct rw_iov_ctx *ctx = cb_arg; 1674 struct spdk_blob *blob = ctx->blob; 1675 struct iovec *iov, *orig_iov; 1676 int iovcnt; 1677 size_t orig_iovoff; 1678 uint64_t page_count, pages_to_boundary, page_offset; 1679 uint64_t byte_count; 1680 1681 if (bserrno != 0 || ctx->pages_remaining == 0) { 1682 ctx->cb_fn(ctx->cb_arg, bserrno); 1683 free(ctx); 1684 return; 1685 } 1686 1687 page_offset = ctx->page_offset; 1688 pages_to_boundary = _spdk_bs_num_pages_to_cluster_boundary(blob, page_offset); 1689 page_count = spdk_min(ctx->pages_remaining, pages_to_boundary); 1690 1691 /* 1692 * Get index and offset into the original iov array for our current position in the I/O sequence. 1693 * byte_count will keep track of how many bytes remaining until orig_iov and orig_iovoff will 1694 * point to the current position in the I/O sequence. 1695 */ 1696 byte_count = ctx->pages_done * sizeof(struct spdk_blob_md_page); 1697 orig_iov = &ctx->orig_iov[0]; 1698 orig_iovoff = 0; 1699 while (byte_count > 0) { 1700 if (byte_count >= orig_iov->iov_len) { 1701 byte_count -= orig_iov->iov_len; 1702 orig_iov++; 1703 } else { 1704 orig_iovoff = byte_count; 1705 byte_count = 0; 1706 } 1707 } 1708 1709 /* 1710 * Build an iov array for the next I/O in the sequence. byte_count will keep track of how many 1711 * bytes of this next I/O remain to be accounted for in the new iov array. 1712 */ 1713 byte_count = page_count * sizeof(struct spdk_blob_md_page); 1714 iov = &ctx->iov[0]; 1715 iovcnt = 0; 1716 while (byte_count > 0) { 1717 iov->iov_len = spdk_min(byte_count, orig_iov->iov_len - orig_iovoff); 1718 iov->iov_base = orig_iov->iov_base + orig_iovoff; 1719 byte_count -= iov->iov_len; 1720 orig_iovoff = 0; 1721 orig_iov++; 1722 iov++; 1723 iovcnt++; 1724 } 1725 1726 ctx->page_offset += page_count; 1727 ctx->pages_done += page_count; 1728 ctx->pages_remaining -= page_count; 1729 iov = &ctx->iov[0]; 1730 1731 if (ctx->read) { 1732 spdk_blob_io_readv(ctx->blob, ctx->channel, iov, iovcnt, page_offset, 1733 page_count, _spdk_rw_iov_split_next, ctx); 1734 } else { 1735 spdk_blob_io_writev(ctx->blob, ctx->channel, iov, iovcnt, page_offset, 1736 page_count, _spdk_rw_iov_split_next, ctx); 1737 } 1738 } 1739 1740 static void 1741 _spdk_blob_request_submit_rw_iov(struct spdk_blob *blob, struct spdk_io_channel *_channel, 1742 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 1743 spdk_blob_op_complete cb_fn, void *cb_arg, bool read) 1744 { 1745 struct spdk_bs_cpl cpl; 1746 1747 assert(blob != NULL); 1748 1749 if (!read && blob->data_ro) { 1750 cb_fn(cb_arg, -EPERM); 1751 return; 1752 } 1753 1754 if (length == 0) { 1755 cb_fn(cb_arg, 0); 1756 return; 1757 } 1758 1759 if (offset + length > blob->active.num_clusters * blob->bs->pages_per_cluster) { 1760 cb_fn(cb_arg, -EINVAL); 1761 return; 1762 } 1763 1764 /* 1765 * For now, we implement readv/writev using a sequence (instead of a batch) to account for having 1766 * to split a request that spans a cluster boundary. For I/O that do not span a cluster boundary, 1767 * there will be no noticeable difference compared to using a batch. For I/O that do span a cluster 1768 * boundary, the target LBAs (after blob offset to LBA translation) may not be contiguous, so we need 1769 * to allocate a separate iov array and split the I/O such that none of the resulting 1770 * smaller I/O cross a cluster boundary. These smaller I/O will be issued in sequence (not in parallel) 1771 * but since this case happens very infrequently, any performance impact will be negligible. 1772 * 1773 * This could be optimized in the future to allocate a big enough iov array to account for all of the iovs 1774 * for all of the smaller I/Os, pre-build all of the iov arrays for the smaller I/Os, then issue them 1775 * in a batch. That would also require creating an intermediate spdk_bs_cpl that would get called 1776 * when the batch was completed, to allow for freeing the memory for the iov arrays. 1777 */ 1778 if (spdk_likely(length <= _spdk_bs_num_pages_to_cluster_boundary(blob, offset))) { 1779 uint32_t lba_count; 1780 uint64_t lba; 1781 1782 _spdk_blob_calculate_lba_and_lba_count(blob, offset, length, &lba, &lba_count); 1783 1784 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1785 cpl.u.blob_basic.cb_fn = cb_fn; 1786 cpl.u.blob_basic.cb_arg = cb_arg; 1787 1788 if (read) { 1789 spdk_bs_sequence_t *seq; 1790 1791 seq = spdk_bs_sequence_start(_channel, &cpl); 1792 if (!seq) { 1793 cb_fn(cb_arg, -ENOMEM); 1794 return; 1795 } 1796 1797 if (_spdk_bs_page_is_allocated(blob, offset)) { 1798 spdk_bs_sequence_readv_dev(seq, iov, iovcnt, lba, lba_count, _spdk_rw_iov_done, NULL); 1799 } else { 1800 spdk_bs_sequence_readv_bs_dev(seq, blob->back_bs_dev, iov, iovcnt, lba, lba_count, 1801 _spdk_rw_iov_done, NULL); 1802 } 1803 } else { 1804 if (_spdk_bs_page_is_allocated(blob, offset)) { 1805 spdk_bs_sequence_t *seq; 1806 1807 seq = spdk_bs_sequence_start(_channel, &cpl); 1808 if (!seq) { 1809 cb_fn(cb_arg, -ENOMEM); 1810 return; 1811 } 1812 1813 spdk_bs_sequence_writev_dev(seq, iov, iovcnt, lba, lba_count, _spdk_rw_iov_done, NULL); 1814 } else { 1815 /* Queue this operation and allocate the cluster */ 1816 spdk_bs_user_op_t *op; 1817 1818 op = spdk_bs_user_op_alloc(_channel, &cpl, SPDK_BLOB_WRITEV, blob, iov, iovcnt, offset, length); 1819 if (!op) { 1820 cb_fn(cb_arg, -ENOMEM); 1821 return; 1822 } 1823 1824 _spdk_bs_allocate_and_copy_cluster(blob, _channel, offset, op); 1825 } 1826 } 1827 } else { 1828 struct rw_iov_ctx *ctx; 1829 1830 ctx = calloc(1, sizeof(struct rw_iov_ctx) + iovcnt * sizeof(struct iovec)); 1831 if (ctx == NULL) { 1832 cb_fn(cb_arg, -ENOMEM); 1833 return; 1834 } 1835 1836 ctx->blob = blob; 1837 ctx->channel = _channel; 1838 ctx->cb_fn = cb_fn; 1839 ctx->cb_arg = cb_arg; 1840 ctx->read = read; 1841 ctx->orig_iov = iov; 1842 ctx->iovcnt = iovcnt; 1843 ctx->page_offset = offset; 1844 ctx->pages_remaining = length; 1845 ctx->pages_done = 0; 1846 1847 _spdk_rw_iov_split_next(ctx, 0); 1848 } 1849 } 1850 1851 static struct spdk_blob * 1852 _spdk_blob_lookup(struct spdk_blob_store *bs, spdk_blob_id blobid) 1853 { 1854 struct spdk_blob *blob; 1855 1856 TAILQ_FOREACH(blob, &bs->blobs, link) { 1857 if (blob->id == blobid) { 1858 return blob; 1859 } 1860 } 1861 1862 return NULL; 1863 } 1864 1865 static int 1866 _spdk_bs_channel_create(void *io_device, void *ctx_buf) 1867 { 1868 struct spdk_blob_store *bs = io_device; 1869 struct spdk_bs_channel *channel = ctx_buf; 1870 struct spdk_bs_dev *dev; 1871 uint32_t max_ops = bs->max_channel_ops; 1872 uint32_t i; 1873 1874 dev = bs->dev; 1875 1876 channel->req_mem = calloc(max_ops, sizeof(struct spdk_bs_request_set)); 1877 if (!channel->req_mem) { 1878 return -1; 1879 } 1880 1881 TAILQ_INIT(&channel->reqs); 1882 1883 for (i = 0; i < max_ops; i++) { 1884 TAILQ_INSERT_TAIL(&channel->reqs, &channel->req_mem[i], link); 1885 } 1886 1887 channel->bs = bs; 1888 channel->dev = dev; 1889 channel->dev_channel = dev->create_channel(dev); 1890 1891 if (!channel->dev_channel) { 1892 SPDK_ERRLOG("Failed to create device channel.\n"); 1893 free(channel->req_mem); 1894 return -1; 1895 } 1896 1897 TAILQ_INIT(&channel->need_cluster_alloc); 1898 1899 return 0; 1900 } 1901 1902 static void 1903 _spdk_bs_channel_destroy(void *io_device, void *ctx_buf) 1904 { 1905 struct spdk_bs_channel *channel = ctx_buf; 1906 spdk_bs_user_op_t *op; 1907 1908 while (!TAILQ_EMPTY(&channel->need_cluster_alloc)) { 1909 op = TAILQ_FIRST(&channel->need_cluster_alloc); 1910 TAILQ_REMOVE(&channel->need_cluster_alloc, op, link); 1911 spdk_bs_user_op_abort(op); 1912 } 1913 1914 free(channel->req_mem); 1915 channel->dev->destroy_channel(channel->dev, channel->dev_channel); 1916 } 1917 1918 static void 1919 _spdk_bs_dev_destroy(void *io_device) 1920 { 1921 struct spdk_blob_store *bs = io_device; 1922 struct spdk_blob *blob, *blob_tmp; 1923 1924 bs->dev->destroy(bs->dev); 1925 1926 TAILQ_FOREACH_SAFE(blob, &bs->blobs, link, blob_tmp) { 1927 TAILQ_REMOVE(&bs->blobs, blob, link); 1928 _spdk_blob_free(blob); 1929 } 1930 1931 pthread_mutex_destroy(&bs->used_clusters_mutex); 1932 1933 spdk_bit_array_free(&bs->used_blobids); 1934 spdk_bit_array_free(&bs->used_md_pages); 1935 spdk_bit_array_free(&bs->used_clusters); 1936 /* 1937 * If this function is called for any reason except a successful unload, 1938 * the unload_cpl type will be NONE and this will be a nop. 1939 */ 1940 spdk_bs_call_cpl(&bs->unload_cpl, bs->unload_err); 1941 1942 free(bs); 1943 } 1944 1945 static void 1946 _spdk_bs_free(struct spdk_blob_store *bs) 1947 { 1948 spdk_bs_unregister_md_thread(bs); 1949 spdk_io_device_unregister(bs, _spdk_bs_dev_destroy); 1950 } 1951 1952 void 1953 spdk_bs_opts_init(struct spdk_bs_opts *opts) 1954 { 1955 opts->cluster_sz = SPDK_BLOB_OPTS_CLUSTER_SZ; 1956 opts->num_md_pages = SPDK_BLOB_OPTS_NUM_MD_PAGES; 1957 opts->max_md_ops = SPDK_BLOB_OPTS_MAX_MD_OPS; 1958 opts->max_channel_ops = SPDK_BLOB_OPTS_DEFAULT_CHANNEL_OPS; 1959 memset(&opts->bstype, 0, sizeof(opts->bstype)); 1960 } 1961 1962 static int 1963 _spdk_bs_opts_verify(struct spdk_bs_opts *opts) 1964 { 1965 if (opts->cluster_sz == 0 || opts->num_md_pages == 0 || opts->max_md_ops == 0 || 1966 opts->max_channel_ops == 0) { 1967 SPDK_ERRLOG("Blobstore options cannot be set to 0\n"); 1968 return -1; 1969 } 1970 1971 return 0; 1972 } 1973 1974 static struct spdk_blob_store * 1975 _spdk_bs_alloc(struct spdk_bs_dev *dev, struct spdk_bs_opts *opts) 1976 { 1977 struct spdk_blob_store *bs; 1978 uint64_t dev_size; 1979 int rc; 1980 1981 dev_size = dev->blocklen * dev->blockcnt; 1982 if (dev_size < opts->cluster_sz) { 1983 /* Device size cannot be smaller than cluster size of blobstore */ 1984 SPDK_ERRLOG("Device size %" PRIu64 " is smaller than cluster size %" PRIu32 "\n", 1985 dev_size, opts->cluster_sz); 1986 return NULL; 1987 } 1988 if (opts->cluster_sz < SPDK_BS_PAGE_SIZE) { 1989 /* Cluster size cannot be smaller than page size */ 1990 SPDK_ERRLOG("Cluster size %" PRIu32 " is smaller than page size %d\n", 1991 opts->cluster_sz, SPDK_BS_PAGE_SIZE); 1992 return NULL; 1993 } 1994 bs = calloc(1, sizeof(struct spdk_blob_store)); 1995 if (!bs) { 1996 return NULL; 1997 } 1998 1999 TAILQ_INIT(&bs->blobs); 2000 bs->dev = dev; 2001 bs->md_thread = spdk_get_thread(); 2002 assert(bs->md_thread != NULL); 2003 2004 /* 2005 * Do not use _spdk_bs_lba_to_cluster() here since blockcnt may not be an 2006 * even multiple of the cluster size. 2007 */ 2008 bs->cluster_sz = opts->cluster_sz; 2009 bs->total_clusters = dev->blockcnt / (bs->cluster_sz / dev->blocklen); 2010 bs->pages_per_cluster = bs->cluster_sz / SPDK_BS_PAGE_SIZE; 2011 bs->num_free_clusters = bs->total_clusters; 2012 bs->used_clusters = spdk_bit_array_create(bs->total_clusters); 2013 if (bs->used_clusters == NULL) { 2014 free(bs); 2015 return NULL; 2016 } 2017 2018 bs->max_channel_ops = opts->max_channel_ops; 2019 bs->super_blob = SPDK_BLOBID_INVALID; 2020 memcpy(&bs->bstype, &opts->bstype, sizeof(opts->bstype)); 2021 2022 /* The metadata is assumed to be at least 1 page */ 2023 bs->used_md_pages = spdk_bit_array_create(1); 2024 bs->used_blobids = spdk_bit_array_create(0); 2025 2026 pthread_mutex_init(&bs->used_clusters_mutex, NULL); 2027 2028 spdk_io_device_register(bs, _spdk_bs_channel_create, _spdk_bs_channel_destroy, 2029 sizeof(struct spdk_bs_channel)); 2030 rc = spdk_bs_register_md_thread(bs); 2031 if (rc == -1) { 2032 spdk_io_device_unregister(bs, NULL); 2033 pthread_mutex_destroy(&bs->used_clusters_mutex); 2034 spdk_bit_array_free(&bs->used_blobids); 2035 spdk_bit_array_free(&bs->used_md_pages); 2036 spdk_bit_array_free(&bs->used_clusters); 2037 free(bs); 2038 return NULL; 2039 } 2040 2041 return bs; 2042 } 2043 2044 /* START spdk_bs_load, spdk_bs_load_ctx will used for both load and unload. */ 2045 2046 struct spdk_bs_load_ctx { 2047 struct spdk_blob_store *bs; 2048 struct spdk_bs_super_block *super; 2049 2050 struct spdk_bs_md_mask *mask; 2051 bool in_page_chain; 2052 uint32_t page_index; 2053 uint32_t cur_page; 2054 struct spdk_blob_md_page *page; 2055 bool is_load; 2056 }; 2057 2058 static void 2059 _spdk_bs_load_ctx_fail(spdk_bs_sequence_t *seq, struct spdk_bs_load_ctx *ctx, int bserrno) 2060 { 2061 assert(bserrno != 0); 2062 2063 spdk_dma_free(ctx->super); 2064 /* 2065 * Only free the blobstore when a load fails. If an unload fails (for some reason) 2066 * we want to keep the blobstore in case the caller wants to try again. 2067 */ 2068 if (ctx->is_load) { 2069 _spdk_bs_free(ctx->bs); 2070 } 2071 free(ctx); 2072 spdk_bs_sequence_finish(seq, bserrno); 2073 } 2074 2075 static void 2076 _spdk_bs_set_mask(struct spdk_bit_array *array, struct spdk_bs_md_mask *mask) 2077 { 2078 uint32_t i = 0; 2079 2080 while (true) { 2081 i = spdk_bit_array_find_first_set(array, i); 2082 if (i >= mask->length) { 2083 break; 2084 } 2085 mask->mask[i / 8] |= 1U << (i % 8); 2086 i++; 2087 } 2088 } 2089 2090 static void 2091 _spdk_bs_write_super(spdk_bs_sequence_t *seq, struct spdk_blob_store *bs, 2092 struct spdk_bs_super_block *super, spdk_bs_sequence_cpl cb_fn, void *cb_arg) 2093 { 2094 /* Update the values in the super block */ 2095 super->super_blob = bs->super_blob; 2096 memcpy(&super->bstype, &bs->bstype, sizeof(bs->bstype)); 2097 super->crc = _spdk_blob_md_page_calc_crc(super); 2098 spdk_bs_sequence_write_dev(seq, super, _spdk_bs_page_to_lba(bs, 0), 2099 _spdk_bs_byte_to_lba(bs, sizeof(*super)), 2100 cb_fn, cb_arg); 2101 } 2102 2103 static void 2104 _spdk_bs_write_used_clusters(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 2105 { 2106 struct spdk_bs_load_ctx *ctx = arg; 2107 uint64_t mask_size, lba, lba_count; 2108 2109 /* Write out the used clusters mask */ 2110 mask_size = ctx->super->used_cluster_mask_len * SPDK_BS_PAGE_SIZE; 2111 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 2112 if (!ctx->mask) { 2113 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2114 return; 2115 } 2116 2117 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_CLUSTERS; 2118 ctx->mask->length = ctx->bs->total_clusters; 2119 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_clusters)); 2120 2121 _spdk_bs_set_mask(ctx->bs->used_clusters, ctx->mask); 2122 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_start); 2123 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_len); 2124 spdk_bs_sequence_write_dev(seq, ctx->mask, lba, lba_count, cb_fn, arg); 2125 } 2126 2127 static void 2128 _spdk_bs_write_used_md(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 2129 { 2130 struct spdk_bs_load_ctx *ctx = arg; 2131 uint64_t mask_size, lba, lba_count; 2132 2133 mask_size = ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE; 2134 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 2135 if (!ctx->mask) { 2136 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2137 return; 2138 } 2139 2140 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_PAGES; 2141 ctx->mask->length = ctx->super->md_len; 2142 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_md_pages)); 2143 2144 _spdk_bs_set_mask(ctx->bs->used_md_pages, ctx->mask); 2145 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_start); 2146 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_len); 2147 spdk_bs_sequence_write_dev(seq, ctx->mask, lba, lba_count, cb_fn, arg); 2148 } 2149 2150 static void 2151 _spdk_bs_write_used_blobids(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 2152 { 2153 struct spdk_bs_load_ctx *ctx = arg; 2154 uint64_t mask_size, lba, lba_count; 2155 2156 if (ctx->super->used_blobid_mask_len == 0) { 2157 /* 2158 * This is a pre-v3 on-disk format where the blobid mask does not get 2159 * written to disk. 2160 */ 2161 cb_fn(seq, arg, 0); 2162 return; 2163 } 2164 2165 mask_size = ctx->super->used_blobid_mask_len * SPDK_BS_PAGE_SIZE; 2166 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 2167 if (!ctx->mask) { 2168 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2169 return; 2170 } 2171 2172 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_BLOBIDS; 2173 ctx->mask->length = ctx->super->md_len; 2174 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_blobids)); 2175 2176 _spdk_bs_set_mask(ctx->bs->used_blobids, ctx->mask); 2177 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_start); 2178 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_len); 2179 spdk_bs_sequence_write_dev(seq, ctx->mask, lba, lba_count, cb_fn, arg); 2180 } 2181 2182 static void 2183 _spdk_bs_load_complete(spdk_bs_sequence_t *seq, struct spdk_bs_load_ctx *ctx, int bserrno) 2184 { 2185 spdk_dma_free(ctx->super); 2186 spdk_dma_free(ctx->mask); 2187 free(ctx); 2188 spdk_bs_sequence_finish(seq, bserrno); 2189 } 2190 2191 static void 2192 _spdk_bs_load_used_blobids_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2193 { 2194 struct spdk_bs_load_ctx *ctx = cb_arg; 2195 uint32_t i, j; 2196 int rc; 2197 2198 /* The type must be correct */ 2199 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_BLOBIDS); 2200 2201 /* The length of the mask (in bits) must not be greater than 2202 * the length of the buffer (converted to bits) */ 2203 assert(ctx->mask->length <= (ctx->super->used_blobid_mask_len * SPDK_BS_PAGE_SIZE * 8)); 2204 2205 /* The length of the mask must be exactly equal to the size 2206 * (in pages) of the metadata region */ 2207 assert(ctx->mask->length == ctx->super->md_len); 2208 2209 rc = spdk_bit_array_resize(&ctx->bs->used_blobids, ctx->mask->length); 2210 if (rc < 0) { 2211 spdk_dma_free(ctx->mask); 2212 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2213 return; 2214 } 2215 2216 for (i = 0; i < ctx->mask->length / 8; i++) { 2217 uint8_t segment = ctx->mask->mask[i]; 2218 for (j = 0; segment; j++) { 2219 if (segment & 1U) { 2220 spdk_bit_array_set(ctx->bs->used_blobids, (i * 8) + j); 2221 } 2222 segment >>= 1U; 2223 } 2224 } 2225 2226 _spdk_bs_load_complete(seq, ctx, bserrno); 2227 } 2228 2229 static void 2230 _spdk_bs_load_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2231 { 2232 struct spdk_bs_load_ctx *ctx = cb_arg; 2233 uint64_t lba, lba_count, mask_size; 2234 uint32_t i, j; 2235 int rc; 2236 2237 /* The type must be correct */ 2238 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_CLUSTERS); 2239 /* The length of the mask (in bits) must not be greater than the length of the buffer (converted to bits) */ 2240 assert(ctx->mask->length <= (ctx->super->used_cluster_mask_len * sizeof( 2241 struct spdk_blob_md_page) * 8)); 2242 /* The length of the mask must be exactly equal to the total number of clusters */ 2243 assert(ctx->mask->length == ctx->bs->total_clusters); 2244 2245 rc = spdk_bit_array_resize(&ctx->bs->used_clusters, ctx->bs->total_clusters); 2246 if (rc < 0) { 2247 spdk_dma_free(ctx->mask); 2248 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2249 return; 2250 } 2251 2252 ctx->bs->num_free_clusters = ctx->bs->total_clusters; 2253 for (i = 0; i < ctx->mask->length / 8; i++) { 2254 uint8_t segment = ctx->mask->mask[i]; 2255 for (j = 0; segment && (j < 8); j++) { 2256 if (segment & 1U) { 2257 spdk_bit_array_set(ctx->bs->used_clusters, (i * 8) + j); 2258 assert(ctx->bs->num_free_clusters > 0); 2259 ctx->bs->num_free_clusters--; 2260 } 2261 segment >>= 1U; 2262 } 2263 } 2264 2265 spdk_dma_free(ctx->mask); 2266 2267 /* Read the used blobids mask */ 2268 mask_size = ctx->super->used_blobid_mask_len * SPDK_BS_PAGE_SIZE; 2269 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 2270 if (!ctx->mask) { 2271 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2272 return; 2273 } 2274 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_start); 2275 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_blobid_mask_len); 2276 spdk_bs_sequence_read_dev(seq, ctx->mask, lba, lba_count, 2277 _spdk_bs_load_used_blobids_cpl, ctx); 2278 } 2279 2280 static void 2281 _spdk_bs_load_used_pages_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_PAGES); 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_page_mask_len * SPDK_BS_PAGE_SIZE * 2292 8)); 2293 /* The length of the mask must be exactly equal to the size (in pages) of the metadata region */ 2294 assert(ctx->mask->length == ctx->super->md_len); 2295 2296 rc = spdk_bit_array_resize(&ctx->bs->used_md_pages, ctx->mask->length); 2297 if (rc < 0) { 2298 spdk_dma_free(ctx->mask); 2299 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2300 return; 2301 } 2302 2303 for (i = 0; i < ctx->mask->length / 8; i++) { 2304 uint8_t segment = ctx->mask->mask[i]; 2305 for (j = 0; segment && (j < 8); j++) { 2306 if (segment & 1U) { 2307 spdk_bit_array_set(ctx->bs->used_md_pages, (i * 8) + j); 2308 } 2309 segment >>= 1U; 2310 } 2311 } 2312 spdk_dma_free(ctx->mask); 2313 2314 /* Read the used clusters mask */ 2315 mask_size = ctx->super->used_cluster_mask_len * SPDK_BS_PAGE_SIZE; 2316 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 2317 if (!ctx->mask) { 2318 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2319 return; 2320 } 2321 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_start); 2322 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_len); 2323 spdk_bs_sequence_read_dev(seq, ctx->mask, lba, lba_count, 2324 _spdk_bs_load_used_clusters_cpl, ctx); 2325 } 2326 2327 static void 2328 _spdk_bs_load_write_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2329 { 2330 struct spdk_bs_load_ctx *ctx = cb_arg; 2331 uint64_t lba, lba_count, mask_size; 2332 2333 /* Read the used pages mask */ 2334 mask_size = ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE; 2335 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 2336 if (!ctx->mask) { 2337 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2338 return; 2339 } 2340 2341 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_start); 2342 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_len); 2343 spdk_bs_sequence_read_dev(seq, ctx->mask, lba, lba_count, 2344 _spdk_bs_load_used_pages_cpl, ctx); 2345 } 2346 2347 static int 2348 _spdk_bs_load_replay_md_parse_page(const struct spdk_blob_md_page *page, struct spdk_blob_store *bs) 2349 { 2350 struct spdk_blob_md_descriptor *desc; 2351 size_t cur_desc = 0; 2352 2353 desc = (struct spdk_blob_md_descriptor *)page->descriptors; 2354 while (cur_desc < sizeof(page->descriptors)) { 2355 if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_PADDING) { 2356 if (desc->length == 0) { 2357 /* If padding and length are 0, this terminates the page */ 2358 break; 2359 } 2360 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_EXTENT) { 2361 struct spdk_blob_md_descriptor_extent *desc_extent; 2362 unsigned int i, j; 2363 unsigned int cluster_count = 0; 2364 2365 desc_extent = (struct spdk_blob_md_descriptor_extent *)desc; 2366 2367 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 2368 for (j = 0; j < desc_extent->extents[i].length; j++) { 2369 spdk_bit_array_set(bs->used_clusters, desc_extent->extents[i].cluster_idx + j); 2370 if (bs->num_free_clusters == 0) { 2371 return -1; 2372 } 2373 bs->num_free_clusters--; 2374 cluster_count++; 2375 } 2376 } 2377 if (cluster_count == 0) { 2378 return -1; 2379 } 2380 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR) { 2381 /* Skip this item */ 2382 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR_INTERNAL) { 2383 /* Skip this item */ 2384 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_FLAGS) { 2385 /* Skip this item */ 2386 } else { 2387 /* Error */ 2388 return -1; 2389 } 2390 /* Advance to the next descriptor */ 2391 cur_desc += sizeof(*desc) + desc->length; 2392 if (cur_desc + sizeof(*desc) > sizeof(page->descriptors)) { 2393 break; 2394 } 2395 desc = (struct spdk_blob_md_descriptor *)((uintptr_t)page->descriptors + cur_desc); 2396 } 2397 return 0; 2398 } 2399 2400 static bool _spdk_bs_load_cur_md_page_valid(struct spdk_bs_load_ctx *ctx) 2401 { 2402 uint32_t crc; 2403 2404 crc = _spdk_blob_md_page_calc_crc(ctx->page); 2405 if (crc != ctx->page->crc) { 2406 return false; 2407 } 2408 2409 if (_spdk_bs_page_to_blobid(ctx->cur_page) != ctx->page->id) { 2410 return false; 2411 } 2412 return true; 2413 } 2414 2415 static void 2416 _spdk_bs_load_replay_cur_md_page(spdk_bs_sequence_t *seq, void *cb_arg); 2417 2418 static void 2419 _spdk_bs_load_write_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2420 { 2421 struct spdk_bs_load_ctx *ctx = cb_arg; 2422 2423 _spdk_bs_load_complete(seq, ctx, bserrno); 2424 } 2425 2426 static void 2427 _spdk_bs_load_write_used_blobids_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2428 { 2429 struct spdk_bs_load_ctx *ctx = cb_arg; 2430 2431 spdk_dma_free(ctx->mask); 2432 ctx->mask = NULL; 2433 2434 _spdk_bs_write_used_clusters(seq, cb_arg, _spdk_bs_load_write_used_clusters_cpl); 2435 } 2436 2437 static void 2438 _spdk_bs_load_write_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2439 { 2440 struct spdk_bs_load_ctx *ctx = cb_arg; 2441 2442 spdk_dma_free(ctx->mask); 2443 ctx->mask = NULL; 2444 2445 _spdk_bs_write_used_blobids(seq, cb_arg, _spdk_bs_load_write_used_blobids_cpl); 2446 } 2447 2448 static void 2449 _spdk_bs_load_write_used_md(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2450 { 2451 _spdk_bs_write_used_md(seq, cb_arg, _spdk_bs_load_write_used_pages_cpl); 2452 } 2453 2454 static void 2455 _spdk_bs_load_replay_md_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2456 { 2457 struct spdk_bs_load_ctx *ctx = cb_arg; 2458 uint64_t num_md_clusters; 2459 uint64_t i; 2460 uint32_t page_num; 2461 2462 if (bserrno != 0) { 2463 _spdk_bs_load_ctx_fail(seq, ctx, bserrno); 2464 return; 2465 } 2466 2467 page_num = ctx->cur_page; 2468 if (_spdk_bs_load_cur_md_page_valid(ctx) == true) { 2469 if (ctx->page->sequence_num == 0 || ctx->in_page_chain == true) { 2470 spdk_bit_array_set(ctx->bs->used_md_pages, page_num); 2471 if (ctx->page->sequence_num == 0) { 2472 spdk_bit_array_set(ctx->bs->used_blobids, page_num); 2473 } 2474 if (_spdk_bs_load_replay_md_parse_page(ctx->page, ctx->bs)) { 2475 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 2476 return; 2477 } 2478 if (ctx->page->next != SPDK_INVALID_MD_PAGE) { 2479 ctx->in_page_chain = true; 2480 ctx->cur_page = ctx->page->next; 2481 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 2482 return; 2483 } 2484 } 2485 } 2486 2487 ctx->in_page_chain = false; 2488 2489 do { 2490 ctx->page_index++; 2491 } while (spdk_bit_array_get(ctx->bs->used_md_pages, ctx->page_index) == true); 2492 2493 if (ctx->page_index < ctx->super->md_len) { 2494 ctx->cur_page = ctx->page_index; 2495 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 2496 } else { 2497 /* Claim all of the clusters used by the metadata */ 2498 num_md_clusters = divide_round_up(ctx->super->md_len, ctx->bs->pages_per_cluster); 2499 for (i = 0; i < num_md_clusters; i++) { 2500 _spdk_bs_claim_cluster(ctx->bs, i); 2501 } 2502 spdk_dma_free(ctx->page); 2503 _spdk_bs_load_write_used_md(seq, ctx, bserrno); 2504 } 2505 } 2506 2507 static void 2508 _spdk_bs_load_replay_cur_md_page(spdk_bs_sequence_t *seq, void *cb_arg) 2509 { 2510 struct spdk_bs_load_ctx *ctx = cb_arg; 2511 uint64_t lba; 2512 2513 assert(ctx->cur_page < ctx->super->md_len); 2514 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->md_start + ctx->cur_page); 2515 spdk_bs_sequence_read_dev(seq, ctx->page, lba, 2516 _spdk_bs_byte_to_lba(ctx->bs, SPDK_BS_PAGE_SIZE), 2517 _spdk_bs_load_replay_md_cpl, ctx); 2518 } 2519 2520 static void 2521 _spdk_bs_load_replay_md(spdk_bs_sequence_t *seq, void *cb_arg) 2522 { 2523 struct spdk_bs_load_ctx *ctx = cb_arg; 2524 2525 ctx->page_index = 0; 2526 ctx->cur_page = 0; 2527 ctx->page = spdk_dma_zmalloc(SPDK_BS_PAGE_SIZE, 2528 SPDK_BS_PAGE_SIZE, 2529 NULL); 2530 if (!ctx->page) { 2531 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2532 return; 2533 } 2534 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 2535 } 2536 2537 static void 2538 _spdk_bs_recover(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2539 { 2540 struct spdk_bs_load_ctx *ctx = cb_arg; 2541 int rc; 2542 2543 if (bserrno != 0) { 2544 _spdk_bs_load_ctx_fail(seq, ctx, -EIO); 2545 return; 2546 } 2547 2548 rc = spdk_bit_array_resize(&ctx->bs->used_md_pages, ctx->super->md_len); 2549 if (rc < 0) { 2550 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2551 return; 2552 } 2553 2554 rc = spdk_bit_array_resize(&ctx->bs->used_blobids, ctx->super->md_len); 2555 if (rc < 0) { 2556 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2557 return; 2558 } 2559 2560 rc = spdk_bit_array_resize(&ctx->bs->used_clusters, ctx->bs->total_clusters); 2561 if (rc < 0) { 2562 _spdk_bs_load_ctx_fail(seq, ctx, -ENOMEM); 2563 return; 2564 } 2565 2566 ctx->bs->num_free_clusters = ctx->bs->total_clusters; 2567 _spdk_bs_load_replay_md(seq, cb_arg); 2568 } 2569 2570 static void 2571 _spdk_bs_load_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2572 { 2573 struct spdk_bs_load_ctx *ctx = cb_arg; 2574 uint32_t crc; 2575 static const char zeros[SPDK_BLOBSTORE_TYPE_LENGTH]; 2576 2577 if (ctx->super->version > SPDK_BS_VERSION || 2578 ctx->super->version < SPDK_BS_INITIAL_VERSION) { 2579 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 2580 return; 2581 } 2582 2583 if (memcmp(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 2584 sizeof(ctx->super->signature)) != 0) { 2585 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 2586 return; 2587 } 2588 2589 crc = _spdk_blob_md_page_calc_crc(ctx->super); 2590 if (crc != ctx->super->crc) { 2591 _spdk_bs_load_ctx_fail(seq, ctx, -EILSEQ); 2592 return; 2593 } 2594 2595 if (memcmp(&ctx->bs->bstype, &ctx->super->bstype, SPDK_BLOBSTORE_TYPE_LENGTH) == 0) { 2596 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Bstype matched - loading blobstore\n"); 2597 } else if (memcmp(&ctx->bs->bstype, zeros, SPDK_BLOBSTORE_TYPE_LENGTH) == 0) { 2598 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Bstype wildcard used - loading blobstore regardless bstype\n"); 2599 } else { 2600 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Unexpected bstype\n"); 2601 SPDK_TRACEDUMP(SPDK_LOG_BLOB, "Expected:", ctx->bs->bstype.bstype, SPDK_BLOBSTORE_TYPE_LENGTH); 2602 SPDK_TRACEDUMP(SPDK_LOG_BLOB, "Found:", ctx->super->bstype.bstype, SPDK_BLOBSTORE_TYPE_LENGTH); 2603 _spdk_bs_load_ctx_fail(seq, ctx, -ENXIO); 2604 return; 2605 } 2606 2607 /* Parse the super block */ 2608 ctx->bs->cluster_sz = ctx->super->cluster_size; 2609 ctx->bs->total_clusters = ctx->bs->dev->blockcnt / (ctx->bs->cluster_sz / ctx->bs->dev->blocklen); 2610 ctx->bs->pages_per_cluster = ctx->bs->cluster_sz / SPDK_BS_PAGE_SIZE; 2611 ctx->bs->md_start = ctx->super->md_start; 2612 ctx->bs->md_len = ctx->super->md_len; 2613 ctx->bs->total_data_clusters = ctx->bs->total_clusters - divide_round_up( 2614 ctx->bs->md_start + ctx->bs->md_len, ctx->bs->pages_per_cluster); 2615 ctx->bs->super_blob = ctx->super->super_blob; 2616 memcpy(&ctx->bs->bstype, &ctx->super->bstype, sizeof(ctx->super->bstype)); 2617 2618 if (ctx->super->clean == 0) { 2619 _spdk_bs_recover(seq, ctx, 0); 2620 } else if (ctx->super->used_blobid_mask_len == 0) { 2621 /* 2622 * Metadata is clean, but this is an old metadata format without 2623 * a blobid mask. Clear the clean bit and then build the masks 2624 * using _spdk_bs_recover. 2625 */ 2626 ctx->super->clean = 0; 2627 _spdk_bs_write_super(seq, ctx->bs, ctx->super, _spdk_bs_recover, ctx); 2628 } else { 2629 ctx->super->clean = 0; 2630 _spdk_bs_write_super(seq, ctx->bs, ctx->super, _spdk_bs_load_write_super_cpl, ctx); 2631 } 2632 } 2633 2634 void 2635 spdk_bs_load(struct spdk_bs_dev *dev, struct spdk_bs_opts *o, 2636 spdk_bs_op_with_handle_complete cb_fn, void *cb_arg) 2637 { 2638 struct spdk_blob_store *bs; 2639 struct spdk_bs_cpl cpl; 2640 spdk_bs_sequence_t *seq; 2641 struct spdk_bs_load_ctx *ctx; 2642 struct spdk_bs_opts opts = {}; 2643 2644 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Loading blobstore from dev %p\n", dev); 2645 2646 if (o) { 2647 opts = *o; 2648 } else { 2649 spdk_bs_opts_init(&opts); 2650 } 2651 2652 if (opts.max_md_ops == 0 || opts.max_channel_ops == 0) { 2653 cb_fn(cb_arg, NULL, -EINVAL); 2654 return; 2655 } 2656 2657 bs = _spdk_bs_alloc(dev, &opts); 2658 if (!bs) { 2659 cb_fn(cb_arg, NULL, -ENOMEM); 2660 return; 2661 } 2662 2663 ctx = calloc(1, sizeof(*ctx)); 2664 if (!ctx) { 2665 _spdk_bs_free(bs); 2666 cb_fn(cb_arg, NULL, -ENOMEM); 2667 return; 2668 } 2669 2670 ctx->bs = bs; 2671 ctx->is_load = true; 2672 2673 /* Allocate memory for the super block */ 2674 ctx->super = spdk_dma_zmalloc(sizeof(*ctx->super), 0x1000, NULL); 2675 if (!ctx->super) { 2676 free(ctx); 2677 _spdk_bs_free(bs); 2678 return; 2679 } 2680 2681 cpl.type = SPDK_BS_CPL_TYPE_BS_HANDLE; 2682 cpl.u.bs_handle.cb_fn = cb_fn; 2683 cpl.u.bs_handle.cb_arg = cb_arg; 2684 cpl.u.bs_handle.bs = bs; 2685 2686 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 2687 if (!seq) { 2688 spdk_dma_free(ctx->super); 2689 free(ctx); 2690 _spdk_bs_free(bs); 2691 cb_fn(cb_arg, NULL, -ENOMEM); 2692 return; 2693 } 2694 2695 /* Read the super block */ 2696 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 2697 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 2698 _spdk_bs_load_super_cpl, ctx); 2699 } 2700 2701 /* END spdk_bs_load */ 2702 2703 /* START spdk_bs_init */ 2704 2705 struct spdk_bs_init_ctx { 2706 struct spdk_blob_store *bs; 2707 struct spdk_bs_super_block *super; 2708 }; 2709 2710 static void 2711 _spdk_bs_init_persist_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2712 { 2713 struct spdk_bs_init_ctx *ctx = cb_arg; 2714 2715 spdk_dma_free(ctx->super); 2716 free(ctx); 2717 2718 spdk_bs_sequence_finish(seq, bserrno); 2719 } 2720 2721 static void 2722 _spdk_bs_init_trim_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2723 { 2724 struct spdk_bs_init_ctx *ctx = cb_arg; 2725 2726 /* Write super block */ 2727 spdk_bs_sequence_write_dev(seq, ctx->super, _spdk_bs_page_to_lba(ctx->bs, 0), 2728 _spdk_bs_byte_to_lba(ctx->bs, sizeof(*ctx->super)), 2729 _spdk_bs_init_persist_super_cpl, ctx); 2730 } 2731 2732 void 2733 spdk_bs_init(struct spdk_bs_dev *dev, struct spdk_bs_opts *o, 2734 spdk_bs_op_with_handle_complete cb_fn, void *cb_arg) 2735 { 2736 struct spdk_bs_init_ctx *ctx; 2737 struct spdk_blob_store *bs; 2738 struct spdk_bs_cpl cpl; 2739 spdk_bs_sequence_t *seq; 2740 spdk_bs_batch_t *batch; 2741 uint64_t num_md_lba; 2742 uint64_t num_md_pages; 2743 uint64_t num_md_clusters; 2744 uint32_t i; 2745 struct spdk_bs_opts opts = {}; 2746 int rc; 2747 2748 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Initializing blobstore on dev %p\n", dev); 2749 2750 if ((SPDK_BS_PAGE_SIZE % dev->blocklen) != 0) { 2751 SPDK_ERRLOG("unsupported dev block length of %d\n", 2752 dev->blocklen); 2753 dev->destroy(dev); 2754 cb_fn(cb_arg, NULL, -EINVAL); 2755 return; 2756 } 2757 2758 if (o) { 2759 opts = *o; 2760 } else { 2761 spdk_bs_opts_init(&opts); 2762 } 2763 2764 if (_spdk_bs_opts_verify(&opts) != 0) { 2765 dev->destroy(dev); 2766 cb_fn(cb_arg, NULL, -EINVAL); 2767 return; 2768 } 2769 2770 bs = _spdk_bs_alloc(dev, &opts); 2771 if (!bs) { 2772 dev->destroy(dev); 2773 cb_fn(cb_arg, NULL, -ENOMEM); 2774 return; 2775 } 2776 2777 if (opts.num_md_pages == SPDK_BLOB_OPTS_NUM_MD_PAGES) { 2778 /* By default, allocate 1 page per cluster. 2779 * Technically, this over-allocates metadata 2780 * because more metadata will reduce the number 2781 * of usable clusters. This can be addressed with 2782 * more complex math in the future. 2783 */ 2784 bs->md_len = bs->total_clusters; 2785 } else { 2786 bs->md_len = opts.num_md_pages; 2787 } 2788 2789 rc = spdk_bit_array_resize(&bs->used_md_pages, bs->md_len); 2790 if (rc < 0) { 2791 _spdk_bs_free(bs); 2792 cb_fn(cb_arg, NULL, -ENOMEM); 2793 return; 2794 } 2795 2796 rc = spdk_bit_array_resize(&bs->used_blobids, bs->md_len); 2797 if (rc < 0) { 2798 _spdk_bs_free(bs); 2799 cb_fn(cb_arg, NULL, -ENOMEM); 2800 return; 2801 } 2802 2803 ctx = calloc(1, sizeof(*ctx)); 2804 if (!ctx) { 2805 _spdk_bs_free(bs); 2806 cb_fn(cb_arg, NULL, -ENOMEM); 2807 return; 2808 } 2809 2810 ctx->bs = bs; 2811 2812 /* Allocate memory for the super block */ 2813 ctx->super = spdk_dma_zmalloc(sizeof(*ctx->super), 0x1000, NULL); 2814 if (!ctx->super) { 2815 free(ctx); 2816 _spdk_bs_free(bs); 2817 return; 2818 } 2819 memcpy(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 2820 sizeof(ctx->super->signature)); 2821 ctx->super->version = SPDK_BS_VERSION; 2822 ctx->super->length = sizeof(*ctx->super); 2823 ctx->super->super_blob = bs->super_blob; 2824 ctx->super->clean = 0; 2825 ctx->super->cluster_size = bs->cluster_sz; 2826 memcpy(&ctx->super->bstype, &bs->bstype, sizeof(bs->bstype)); 2827 2828 /* Calculate how many pages the metadata consumes at the front 2829 * of the disk. 2830 */ 2831 2832 /* The super block uses 1 page */ 2833 num_md_pages = 1; 2834 2835 /* The used_md_pages mask requires 1 bit per metadata page, rounded 2836 * up to the nearest page, plus a header. 2837 */ 2838 ctx->super->used_page_mask_start = num_md_pages; 2839 ctx->super->used_page_mask_len = divide_round_up(sizeof(struct spdk_bs_md_mask) + 2840 divide_round_up(bs->md_len, 8), 2841 SPDK_BS_PAGE_SIZE); 2842 num_md_pages += ctx->super->used_page_mask_len; 2843 2844 /* The used_clusters mask requires 1 bit per cluster, rounded 2845 * up to the nearest page, plus a header. 2846 */ 2847 ctx->super->used_cluster_mask_start = num_md_pages; 2848 ctx->super->used_cluster_mask_len = divide_round_up(sizeof(struct spdk_bs_md_mask) + 2849 divide_round_up(bs->total_clusters, 8), 2850 SPDK_BS_PAGE_SIZE); 2851 num_md_pages += ctx->super->used_cluster_mask_len; 2852 2853 /* The used_blobids mask requires 1 bit per metadata page, rounded 2854 * up to the nearest page, plus a header. 2855 */ 2856 ctx->super->used_blobid_mask_start = num_md_pages; 2857 ctx->super->used_blobid_mask_len = divide_round_up(sizeof(struct spdk_bs_md_mask) + 2858 divide_round_up(bs->md_len, 8), 2859 SPDK_BS_PAGE_SIZE); 2860 num_md_pages += ctx->super->used_blobid_mask_len; 2861 2862 /* The metadata region size was chosen above */ 2863 ctx->super->md_start = bs->md_start = num_md_pages; 2864 ctx->super->md_len = bs->md_len; 2865 num_md_pages += bs->md_len; 2866 2867 num_md_lba = _spdk_bs_page_to_lba(bs, num_md_pages); 2868 2869 ctx->super->crc = _spdk_blob_md_page_calc_crc(ctx->super); 2870 2871 num_md_clusters = divide_round_up(num_md_pages, bs->pages_per_cluster); 2872 if (num_md_clusters > bs->total_clusters) { 2873 SPDK_ERRLOG("Blobstore metadata cannot use more clusters than is available, " 2874 "please decrease number of pages reserved for metadata " 2875 "or increase cluster size.\n"); 2876 spdk_dma_free(ctx->super); 2877 free(ctx); 2878 _spdk_bs_free(bs); 2879 cb_fn(cb_arg, NULL, -ENOMEM); 2880 return; 2881 } 2882 /* Claim all of the clusters used by the metadata */ 2883 for (i = 0; i < num_md_clusters; i++) { 2884 _spdk_bs_claim_cluster(bs, i); 2885 } 2886 2887 bs->total_data_clusters = bs->num_free_clusters; 2888 2889 cpl.type = SPDK_BS_CPL_TYPE_BS_HANDLE; 2890 cpl.u.bs_handle.cb_fn = cb_fn; 2891 cpl.u.bs_handle.cb_arg = cb_arg; 2892 cpl.u.bs_handle.bs = bs; 2893 2894 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 2895 if (!seq) { 2896 spdk_dma_free(ctx->super); 2897 free(ctx); 2898 _spdk_bs_free(bs); 2899 cb_fn(cb_arg, NULL, -ENOMEM); 2900 return; 2901 } 2902 2903 batch = spdk_bs_sequence_to_batch(seq, _spdk_bs_init_trim_cpl, ctx); 2904 2905 /* Clear metadata space */ 2906 spdk_bs_batch_write_zeroes_dev(batch, 0, num_md_lba); 2907 /* Trim data clusters */ 2908 spdk_bs_batch_unmap_dev(batch, num_md_lba, ctx->bs->dev->blockcnt - num_md_lba); 2909 2910 spdk_bs_batch_close(batch); 2911 } 2912 2913 /* END spdk_bs_init */ 2914 2915 /* START spdk_bs_destroy */ 2916 2917 static void 2918 _spdk_bs_destroy_trim_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2919 { 2920 struct spdk_bs_init_ctx *ctx = cb_arg; 2921 struct spdk_blob_store *bs = ctx->bs; 2922 2923 /* 2924 * We need to defer calling spdk_bs_call_cpl() until after 2925 * dev destruction, so tuck these away for later use. 2926 */ 2927 bs->unload_err = bserrno; 2928 memcpy(&bs->unload_cpl, &seq->cpl, sizeof(struct spdk_bs_cpl)); 2929 seq->cpl.type = SPDK_BS_CPL_TYPE_NONE; 2930 2931 spdk_bs_sequence_finish(seq, bserrno); 2932 2933 _spdk_bs_free(bs); 2934 free(ctx); 2935 } 2936 2937 void 2938 spdk_bs_destroy(struct spdk_blob_store *bs, spdk_bs_op_complete cb_fn, 2939 void *cb_arg) 2940 { 2941 struct spdk_bs_cpl cpl; 2942 spdk_bs_sequence_t *seq; 2943 struct spdk_bs_init_ctx *ctx; 2944 2945 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Destroying blobstore\n"); 2946 2947 if (!TAILQ_EMPTY(&bs->blobs)) { 2948 SPDK_ERRLOG("Blobstore still has open blobs\n"); 2949 cb_fn(cb_arg, -EBUSY); 2950 return; 2951 } 2952 2953 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 2954 cpl.u.bs_basic.cb_fn = cb_fn; 2955 cpl.u.bs_basic.cb_arg = cb_arg; 2956 2957 ctx = calloc(1, sizeof(*ctx)); 2958 if (!ctx) { 2959 cb_fn(cb_arg, -ENOMEM); 2960 return; 2961 } 2962 2963 ctx->bs = bs; 2964 2965 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 2966 if (!seq) { 2967 free(ctx); 2968 cb_fn(cb_arg, -ENOMEM); 2969 return; 2970 } 2971 2972 /* Write zeroes to the super block */ 2973 spdk_bs_sequence_write_zeroes_dev(seq, 2974 _spdk_bs_page_to_lba(bs, 0), 2975 _spdk_bs_byte_to_lba(bs, sizeof(struct spdk_bs_super_block)), 2976 _spdk_bs_destroy_trim_cpl, ctx); 2977 } 2978 2979 /* END spdk_bs_destroy */ 2980 2981 /* START spdk_bs_unload */ 2982 2983 static void 2984 _spdk_bs_unload_write_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2985 { 2986 struct spdk_bs_load_ctx *ctx = cb_arg; 2987 2988 spdk_dma_free(ctx->super); 2989 2990 /* 2991 * We need to defer calling spdk_bs_call_cpl() until after 2992 * dev destuction, so tuck these away for later use. 2993 */ 2994 ctx->bs->unload_err = bserrno; 2995 memcpy(&ctx->bs->unload_cpl, &seq->cpl, sizeof(struct spdk_bs_cpl)); 2996 seq->cpl.type = SPDK_BS_CPL_TYPE_NONE; 2997 2998 spdk_bs_sequence_finish(seq, bserrno); 2999 3000 _spdk_bs_free(ctx->bs); 3001 free(ctx); 3002 } 3003 3004 static void 3005 _spdk_bs_unload_write_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3006 { 3007 struct spdk_bs_load_ctx *ctx = cb_arg; 3008 3009 spdk_dma_free(ctx->mask); 3010 ctx->super->clean = 1; 3011 3012 _spdk_bs_write_super(seq, ctx->bs, ctx->super, _spdk_bs_unload_write_super_cpl, ctx); 3013 } 3014 3015 static void 3016 _spdk_bs_unload_write_used_blobids_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3017 { 3018 struct spdk_bs_load_ctx *ctx = cb_arg; 3019 3020 spdk_dma_free(ctx->mask); 3021 ctx->mask = NULL; 3022 3023 _spdk_bs_write_used_clusters(seq, cb_arg, _spdk_bs_unload_write_used_clusters_cpl); 3024 } 3025 3026 static void 3027 _spdk_bs_unload_write_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3028 { 3029 struct spdk_bs_load_ctx *ctx = cb_arg; 3030 3031 spdk_dma_free(ctx->mask); 3032 ctx->mask = NULL; 3033 3034 _spdk_bs_write_used_blobids(seq, cb_arg, _spdk_bs_unload_write_used_blobids_cpl); 3035 } 3036 3037 static void 3038 _spdk_bs_unload_read_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3039 { 3040 _spdk_bs_write_used_md(seq, cb_arg, _spdk_bs_unload_write_used_pages_cpl); 3041 } 3042 3043 void 3044 spdk_bs_unload(struct spdk_blob_store *bs, spdk_bs_op_complete cb_fn, void *cb_arg) 3045 { 3046 struct spdk_bs_cpl cpl; 3047 spdk_bs_sequence_t *seq; 3048 struct spdk_bs_load_ctx *ctx; 3049 3050 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Syncing blobstore\n"); 3051 3052 if (!TAILQ_EMPTY(&bs->blobs)) { 3053 SPDK_ERRLOG("Blobstore still has open blobs\n"); 3054 cb_fn(cb_arg, -EBUSY); 3055 return; 3056 } 3057 3058 ctx = calloc(1, sizeof(*ctx)); 3059 if (!ctx) { 3060 cb_fn(cb_arg, -ENOMEM); 3061 return; 3062 } 3063 3064 ctx->bs = bs; 3065 ctx->is_load = false; 3066 3067 ctx->super = spdk_dma_zmalloc(sizeof(*ctx->super), 0x1000, NULL); 3068 if (!ctx->super) { 3069 free(ctx); 3070 cb_fn(cb_arg, -ENOMEM); 3071 return; 3072 } 3073 3074 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 3075 cpl.u.bs_basic.cb_fn = cb_fn; 3076 cpl.u.bs_basic.cb_arg = cb_arg; 3077 3078 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3079 if (!seq) { 3080 spdk_dma_free(ctx->super); 3081 free(ctx); 3082 cb_fn(cb_arg, -ENOMEM); 3083 return; 3084 } 3085 3086 /* Read super block */ 3087 spdk_bs_sequence_read_dev(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 3088 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 3089 _spdk_bs_unload_read_super_cpl, ctx); 3090 } 3091 3092 /* END spdk_bs_unload */ 3093 3094 void 3095 spdk_bs_set_super(struct spdk_blob_store *bs, spdk_blob_id blobid, 3096 spdk_bs_op_complete cb_fn, void *cb_arg) 3097 { 3098 bs->super_blob = blobid; 3099 cb_fn(cb_arg, 0); 3100 } 3101 3102 void 3103 spdk_bs_get_super(struct spdk_blob_store *bs, 3104 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 3105 { 3106 if (bs->super_blob == SPDK_BLOBID_INVALID) { 3107 cb_fn(cb_arg, SPDK_BLOBID_INVALID, -ENOENT); 3108 } else { 3109 cb_fn(cb_arg, bs->super_blob, 0); 3110 } 3111 } 3112 3113 uint64_t 3114 spdk_bs_get_cluster_size(struct spdk_blob_store *bs) 3115 { 3116 return bs->cluster_sz; 3117 } 3118 3119 uint64_t 3120 spdk_bs_get_page_size(struct spdk_blob_store *bs) 3121 { 3122 return SPDK_BS_PAGE_SIZE; 3123 } 3124 3125 uint64_t 3126 spdk_bs_free_cluster_count(struct spdk_blob_store *bs) 3127 { 3128 return bs->num_free_clusters; 3129 } 3130 3131 uint64_t 3132 spdk_bs_total_data_cluster_count(struct spdk_blob_store *bs) 3133 { 3134 return bs->total_data_clusters; 3135 } 3136 3137 static int 3138 spdk_bs_register_md_thread(struct spdk_blob_store *bs) 3139 { 3140 bs->md_channel = spdk_get_io_channel(bs); 3141 if (!bs->md_channel) { 3142 SPDK_ERRLOG("Failed to get IO channel.\n"); 3143 return -1; 3144 } 3145 3146 return 0; 3147 } 3148 3149 static int 3150 spdk_bs_unregister_md_thread(struct spdk_blob_store *bs) 3151 { 3152 spdk_put_io_channel(bs->md_channel); 3153 3154 return 0; 3155 } 3156 3157 spdk_blob_id spdk_blob_get_id(struct spdk_blob *blob) 3158 { 3159 assert(blob != NULL); 3160 3161 return blob->id; 3162 } 3163 3164 uint64_t spdk_blob_get_num_pages(struct spdk_blob *blob) 3165 { 3166 assert(blob != NULL); 3167 3168 return _spdk_bs_cluster_to_page(blob->bs, blob->active.num_clusters); 3169 } 3170 3171 uint64_t spdk_blob_get_num_clusters(struct spdk_blob *blob) 3172 { 3173 assert(blob != NULL); 3174 3175 return blob->active.num_clusters; 3176 } 3177 3178 /* START spdk_bs_create_blob */ 3179 3180 static void 3181 _spdk_bs_create_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3182 { 3183 struct spdk_blob *blob = cb_arg; 3184 3185 _spdk_blob_free(blob); 3186 3187 spdk_bs_sequence_finish(seq, bserrno); 3188 } 3189 3190 static int 3191 _spdk_blob_set_xattrs(struct spdk_blob *blob, const struct spdk_blob_xattr_opts *xattrs, 3192 bool internal) 3193 { 3194 uint64_t i; 3195 size_t value_len = 0; 3196 int rc; 3197 const void *value = NULL; 3198 if (xattrs->count > 0 && xattrs->get_value == NULL) { 3199 return -EINVAL; 3200 } 3201 for (i = 0; i < xattrs->count; i++) { 3202 xattrs->get_value(xattrs->ctx, xattrs->names[i], &value, &value_len); 3203 if (value == NULL || value_len == 0) { 3204 return -EINVAL; 3205 } 3206 rc = _spdk_blob_set_xattr(blob, xattrs->names[i], value, value_len, internal); 3207 if (rc < 0) { 3208 return rc; 3209 } 3210 } 3211 return 0; 3212 } 3213 3214 static void 3215 _spdk_blob_set_thin_provision(struct spdk_blob *blob) 3216 { 3217 _spdk_blob_verify_md_op(blob); 3218 blob->invalid_flags |= SPDK_BLOB_THIN_PROV; 3219 blob->state = SPDK_BLOB_STATE_DIRTY; 3220 } 3221 3222 void spdk_bs_create_blob_ext(struct spdk_blob_store *bs, const struct spdk_blob_opts *opts, 3223 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 3224 { 3225 struct spdk_blob *blob; 3226 uint32_t page_idx; 3227 struct spdk_bs_cpl cpl; 3228 struct spdk_blob_opts opts_default; 3229 spdk_bs_sequence_t *seq; 3230 spdk_blob_id id; 3231 int rc; 3232 3233 assert(spdk_get_thread() == bs->md_thread); 3234 3235 page_idx = spdk_bit_array_find_first_clear(bs->used_md_pages, 0); 3236 if (page_idx >= spdk_bit_array_capacity(bs->used_md_pages)) { 3237 cb_fn(cb_arg, 0, -ENOMEM); 3238 return; 3239 } 3240 spdk_bit_array_set(bs->used_blobids, page_idx); 3241 spdk_bit_array_set(bs->used_md_pages, page_idx); 3242 3243 id = _spdk_bs_page_to_blobid(page_idx); 3244 3245 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Creating blob with id %lu at page %u\n", id, page_idx); 3246 3247 blob = _spdk_blob_alloc(bs, id); 3248 if (!blob) { 3249 cb_fn(cb_arg, 0, -ENOMEM); 3250 return; 3251 } 3252 3253 if (!opts) { 3254 spdk_blob_opts_init(&opts_default); 3255 opts = &opts_default; 3256 } 3257 3258 rc = _spdk_blob_set_xattrs(blob, &opts->xattrs, false); 3259 if (rc < 0) { 3260 _spdk_blob_free(blob); 3261 cb_fn(cb_arg, 0, rc); 3262 return; 3263 } 3264 if (opts->thin_provision) { 3265 _spdk_blob_set_thin_provision(blob); 3266 } 3267 3268 rc = spdk_blob_resize(blob, opts->num_clusters); 3269 if (rc < 0) { 3270 _spdk_blob_free(blob); 3271 cb_fn(cb_arg, 0, rc); 3272 return; 3273 } 3274 cpl.type = SPDK_BS_CPL_TYPE_BLOBID; 3275 cpl.u.blobid.cb_fn = cb_fn; 3276 cpl.u.blobid.cb_arg = cb_arg; 3277 cpl.u.blobid.blobid = blob->id; 3278 3279 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3280 if (!seq) { 3281 _spdk_blob_free(blob); 3282 cb_fn(cb_arg, 0, -ENOMEM); 3283 return; 3284 } 3285 3286 _spdk_blob_persist(seq, blob, _spdk_bs_create_blob_cpl, blob); 3287 } 3288 3289 void spdk_bs_create_blob(struct spdk_blob_store *bs, 3290 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 3291 { 3292 spdk_bs_create_blob_ext(bs, NULL, cb_fn, cb_arg); 3293 } 3294 3295 /* END spdk_bs_create_blob */ 3296 3297 /* START spdk_blob_resize */ 3298 int 3299 spdk_blob_resize(struct spdk_blob *blob, uint64_t sz) 3300 { 3301 int rc; 3302 3303 _spdk_blob_verify_md_op(blob); 3304 3305 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Resizing blob %lu to %lu clusters\n", blob->id, sz); 3306 3307 if (blob->md_ro) { 3308 return -EPERM; 3309 } 3310 3311 if (sz == blob->active.num_clusters) { 3312 return 0; 3313 } 3314 3315 rc = _spdk_resize_blob(blob, sz); 3316 if (rc < 0) { 3317 return rc; 3318 } 3319 3320 return 0; 3321 } 3322 3323 /* END spdk_blob_resize */ 3324 3325 3326 /* START spdk_bs_delete_blob */ 3327 3328 static void 3329 _spdk_bs_delete_close_cpl(void *cb_arg, int bserrno) 3330 { 3331 spdk_bs_sequence_t *seq = cb_arg; 3332 3333 spdk_bs_sequence_finish(seq, bserrno); 3334 } 3335 3336 static void 3337 _spdk_bs_delete_persist_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3338 { 3339 struct spdk_blob *blob = cb_arg; 3340 3341 if (bserrno != 0) { 3342 /* 3343 * We already removed this blob from the blobstore tailq, so 3344 * we need to free it here since this is the last reference 3345 * to it. 3346 */ 3347 _spdk_blob_free(blob); 3348 _spdk_bs_delete_close_cpl(seq, bserrno); 3349 return; 3350 } 3351 3352 /* 3353 * This will immediately decrement the ref_count and call 3354 * the completion routine since the metadata state is clean. 3355 * By calling spdk_blob_close, we reduce the number of call 3356 * points into code that touches the blob->open_ref count 3357 * and the blobstore's blob list. 3358 */ 3359 spdk_blob_close(blob, _spdk_bs_delete_close_cpl, seq); 3360 } 3361 3362 static void 3363 _spdk_bs_delete_open_cpl(void *cb_arg, struct spdk_blob *blob, int bserrno) 3364 { 3365 spdk_bs_sequence_t *seq = cb_arg; 3366 uint32_t page_num; 3367 3368 if (bserrno != 0) { 3369 spdk_bs_sequence_finish(seq, bserrno); 3370 return; 3371 } 3372 3373 _spdk_blob_verify_md_op(blob); 3374 3375 if (blob->open_ref > 1) { 3376 /* 3377 * Someone has this blob open (besides this delete context). 3378 * Decrement the ref count directly and return -EBUSY. 3379 */ 3380 blob->open_ref--; 3381 spdk_bs_sequence_finish(seq, -EBUSY); 3382 return; 3383 } 3384 3385 /* 3386 * Remove the blob from the blob_store list now, to ensure it does not 3387 * get returned after this point by _spdk_blob_lookup(). 3388 */ 3389 TAILQ_REMOVE(&blob->bs->blobs, blob, link); 3390 page_num = _spdk_bs_blobid_to_page(blob->id); 3391 spdk_bit_array_clear(blob->bs->used_blobids, page_num); 3392 blob->state = SPDK_BLOB_STATE_DIRTY; 3393 blob->active.num_pages = 0; 3394 _spdk_resize_blob(blob, 0); 3395 3396 _spdk_blob_persist(seq, blob, _spdk_bs_delete_persist_cpl, blob); 3397 } 3398 3399 void 3400 spdk_bs_delete_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 3401 spdk_blob_op_complete cb_fn, void *cb_arg) 3402 { 3403 struct spdk_bs_cpl cpl; 3404 spdk_bs_sequence_t *seq; 3405 3406 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Deleting blob %lu\n", blobid); 3407 3408 assert(spdk_get_thread() == bs->md_thread); 3409 3410 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 3411 cpl.u.blob_basic.cb_fn = cb_fn; 3412 cpl.u.blob_basic.cb_arg = cb_arg; 3413 3414 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3415 if (!seq) { 3416 cb_fn(cb_arg, -ENOMEM); 3417 return; 3418 } 3419 3420 spdk_bs_open_blob(bs, blobid, _spdk_bs_delete_open_cpl, seq); 3421 } 3422 3423 /* END spdk_bs_delete_blob */ 3424 3425 /* START spdk_bs_open_blob */ 3426 3427 static void 3428 _spdk_bs_open_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3429 { 3430 struct spdk_blob *blob = cb_arg; 3431 3432 /* If the blob have crc error, we just return NULL. */ 3433 if (blob == NULL) { 3434 seq->cpl.u.blob_handle.blob = NULL; 3435 spdk_bs_sequence_finish(seq, bserrno); 3436 return; 3437 } 3438 3439 blob->open_ref++; 3440 3441 TAILQ_INSERT_HEAD(&blob->bs->blobs, blob, link); 3442 3443 spdk_bs_sequence_finish(seq, bserrno); 3444 } 3445 3446 void spdk_bs_open_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 3447 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 3448 { 3449 struct spdk_blob *blob; 3450 struct spdk_bs_cpl cpl; 3451 spdk_bs_sequence_t *seq; 3452 uint32_t page_num; 3453 3454 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Opening blob %lu\n", blobid); 3455 assert(spdk_get_thread() == bs->md_thread); 3456 3457 page_num = _spdk_bs_blobid_to_page(blobid); 3458 if (spdk_bit_array_get(bs->used_blobids, page_num) == false) { 3459 /* Invalid blobid */ 3460 cb_fn(cb_arg, NULL, -ENOENT); 3461 return; 3462 } 3463 3464 blob = _spdk_blob_lookup(bs, blobid); 3465 if (blob) { 3466 blob->open_ref++; 3467 cb_fn(cb_arg, blob, 0); 3468 return; 3469 } 3470 3471 blob = _spdk_blob_alloc(bs, blobid); 3472 if (!blob) { 3473 cb_fn(cb_arg, NULL, -ENOMEM); 3474 return; 3475 } 3476 3477 cpl.type = SPDK_BS_CPL_TYPE_BLOB_HANDLE; 3478 cpl.u.blob_handle.cb_fn = cb_fn; 3479 cpl.u.blob_handle.cb_arg = cb_arg; 3480 cpl.u.blob_handle.blob = blob; 3481 3482 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 3483 if (!seq) { 3484 _spdk_blob_free(blob); 3485 cb_fn(cb_arg, NULL, -ENOMEM); 3486 return; 3487 } 3488 3489 _spdk_blob_load(seq, blob, _spdk_bs_open_blob_cpl, blob); 3490 } 3491 /* END spdk_bs_open_blob */ 3492 3493 /* START spdk_blob_set_read_only */ 3494 int spdk_blob_set_read_only(struct spdk_blob *blob) 3495 { 3496 _spdk_blob_verify_md_op(blob); 3497 3498 blob->data_ro_flags |= SPDK_BLOB_READ_ONLY; 3499 3500 blob->state = SPDK_BLOB_STATE_DIRTY; 3501 return 0; 3502 } 3503 /* END spdk_blob_set_read_only */ 3504 3505 /* START spdk_blob_sync_md */ 3506 3507 static void 3508 _spdk_blob_sync_md_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3509 { 3510 struct spdk_blob *blob = cb_arg; 3511 3512 if (bserrno == 0 && (blob->data_ro_flags & SPDK_BLOB_READ_ONLY)) { 3513 blob->data_ro = true; 3514 blob->md_ro = true; 3515 } 3516 3517 spdk_bs_sequence_finish(seq, bserrno); 3518 } 3519 3520 static void 3521 _spdk_blob_sync_md(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 3522 { 3523 struct spdk_bs_cpl cpl; 3524 spdk_bs_sequence_t *seq; 3525 3526 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 3527 cpl.u.blob_basic.cb_fn = cb_fn; 3528 cpl.u.blob_basic.cb_arg = cb_arg; 3529 3530 seq = spdk_bs_sequence_start(blob->bs->md_channel, &cpl); 3531 if (!seq) { 3532 cb_fn(cb_arg, -ENOMEM); 3533 return; 3534 } 3535 3536 _spdk_blob_persist(seq, blob, _spdk_blob_sync_md_cpl, blob); 3537 } 3538 3539 void 3540 spdk_blob_sync_md(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 3541 { 3542 _spdk_blob_verify_md_op(blob); 3543 3544 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Syncing blob %lu\n", blob->id); 3545 3546 if (blob->md_ro) { 3547 assert(blob->state == SPDK_BLOB_STATE_CLEAN); 3548 cb_fn(cb_arg, 0); 3549 return; 3550 } 3551 3552 _spdk_blob_sync_md(blob, cb_fn, cb_arg); 3553 } 3554 3555 /* END spdk_blob_sync_md */ 3556 3557 struct spdk_blob_insert_cluster_ctx { 3558 struct spdk_thread *thread; 3559 struct spdk_blob *blob; 3560 uint32_t cluster_num; /* cluster index in blob */ 3561 uint32_t cluster; /* cluster on disk */ 3562 int rc; 3563 spdk_blob_op_complete cb_fn; 3564 void *cb_arg; 3565 }; 3566 3567 static void 3568 _spdk_blob_insert_cluster_msg_cpl(void *arg) 3569 { 3570 struct spdk_blob_insert_cluster_ctx *ctx = arg; 3571 3572 ctx->cb_fn(ctx->cb_arg, ctx->rc); 3573 free(ctx); 3574 } 3575 3576 static void 3577 _spdk_blob_insert_cluster_msg_cb(void *arg, int bserrno) 3578 { 3579 struct spdk_blob_insert_cluster_ctx *ctx = arg; 3580 3581 ctx->rc = bserrno; 3582 spdk_thread_send_msg(ctx->thread, _spdk_blob_insert_cluster_msg_cpl, ctx); 3583 } 3584 3585 static void 3586 _spdk_blob_insert_cluster_msg(void *arg) 3587 { 3588 struct spdk_blob_insert_cluster_ctx *ctx = arg; 3589 3590 ctx->rc = _spdk_blob_insert_cluster(ctx->blob, ctx->cluster_num, ctx->cluster); 3591 if (ctx->rc != 0) { 3592 spdk_thread_send_msg(ctx->thread, _spdk_blob_insert_cluster_msg_cpl, ctx); 3593 return; 3594 } 3595 3596 ctx->blob->state = SPDK_BLOB_STATE_DIRTY; 3597 _spdk_blob_sync_md(ctx->blob, _spdk_blob_insert_cluster_msg_cb, ctx); 3598 } 3599 3600 void 3601 _spdk_blob_insert_cluster_on_md_thread(struct spdk_blob *blob, uint32_t cluster_num, 3602 uint64_t cluster, spdk_blob_op_complete cb_fn, void *cb_arg) 3603 { 3604 struct spdk_blob_insert_cluster_ctx *ctx; 3605 3606 ctx = calloc(1, sizeof(*ctx)); 3607 if (ctx == NULL) { 3608 cb_fn(cb_arg, -ENOMEM); 3609 return; 3610 } 3611 3612 ctx->thread = spdk_get_thread(); 3613 ctx->blob = blob; 3614 ctx->cluster_num = cluster_num; 3615 ctx->cluster = cluster; 3616 ctx->cb_fn = cb_fn; 3617 ctx->cb_arg = cb_arg; 3618 3619 spdk_thread_send_msg(blob->bs->md_thread, _spdk_blob_insert_cluster_msg, ctx); 3620 } 3621 3622 /* START spdk_blob_close */ 3623 3624 static void 3625 _spdk_blob_close_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 3626 { 3627 struct spdk_blob *blob = cb_arg; 3628 3629 if (bserrno == 0) { 3630 blob->open_ref--; 3631 if (blob->open_ref == 0) { 3632 /* 3633 * Blobs with active.num_pages == 0 are deleted blobs. 3634 * these blobs are removed from the blob_store list 3635 * when the deletion process starts - so don't try to 3636 * remove them again. 3637 */ 3638 if (blob->active.num_pages > 0) { 3639 TAILQ_REMOVE(&blob->bs->blobs, blob, link); 3640 } 3641 _spdk_blob_free(blob); 3642 } 3643 } 3644 3645 spdk_bs_sequence_finish(seq, bserrno); 3646 } 3647 3648 void spdk_blob_close(struct spdk_blob *blob, spdk_blob_op_complete cb_fn, void *cb_arg) 3649 { 3650 struct spdk_bs_cpl cpl; 3651 spdk_bs_sequence_t *seq; 3652 3653 _spdk_blob_verify_md_op(blob); 3654 3655 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "Closing blob %lu\n", blob->id); 3656 3657 if (blob->open_ref == 0) { 3658 cb_fn(cb_arg, -EBADF); 3659 return; 3660 } 3661 3662 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 3663 cpl.u.blob_basic.cb_fn = cb_fn; 3664 cpl.u.blob_basic.cb_arg = cb_arg; 3665 3666 seq = spdk_bs_sequence_start(blob->bs->md_channel, &cpl); 3667 if (!seq) { 3668 cb_fn(cb_arg, -ENOMEM); 3669 return; 3670 } 3671 3672 /* Sync metadata */ 3673 _spdk_blob_persist(seq, blob, _spdk_blob_close_cpl, blob); 3674 } 3675 3676 /* END spdk_blob_close */ 3677 3678 struct spdk_io_channel *spdk_bs_alloc_io_channel(struct spdk_blob_store *bs) 3679 { 3680 return spdk_get_io_channel(bs); 3681 } 3682 3683 void spdk_bs_free_io_channel(struct spdk_io_channel *channel) 3684 { 3685 spdk_put_io_channel(channel); 3686 } 3687 3688 void spdk_blob_io_unmap(struct spdk_blob *blob, struct spdk_io_channel *channel, 3689 uint64_t offset, uint64_t length, spdk_blob_op_complete cb_fn, void *cb_arg) 3690 { 3691 _spdk_blob_request_submit_op(blob, channel, NULL, offset, length, cb_fn, cb_arg, 3692 SPDK_BLOB_UNMAP); 3693 } 3694 3695 void spdk_blob_io_write_zeroes(struct spdk_blob *blob, struct spdk_io_channel *channel, 3696 uint64_t offset, uint64_t length, spdk_blob_op_complete cb_fn, void *cb_arg) 3697 { 3698 _spdk_blob_request_submit_op(blob, channel, NULL, offset, length, cb_fn, cb_arg, 3699 SPDK_BLOB_WRITE_ZEROES); 3700 } 3701 3702 void spdk_blob_io_write(struct spdk_blob *blob, struct spdk_io_channel *channel, 3703 void *payload, uint64_t offset, uint64_t length, 3704 spdk_blob_op_complete cb_fn, void *cb_arg) 3705 { 3706 _spdk_blob_request_submit_op(blob, channel, payload, offset, length, cb_fn, cb_arg, 3707 SPDK_BLOB_WRITE); 3708 } 3709 3710 void spdk_blob_io_read(struct spdk_blob *blob, struct spdk_io_channel *channel, 3711 void *payload, uint64_t offset, uint64_t length, 3712 spdk_blob_op_complete cb_fn, void *cb_arg) 3713 { 3714 _spdk_blob_request_submit_op(blob, channel, payload, offset, length, cb_fn, cb_arg, 3715 SPDK_BLOB_READ); 3716 } 3717 3718 void spdk_blob_io_writev(struct spdk_blob *blob, struct spdk_io_channel *channel, 3719 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 3720 spdk_blob_op_complete cb_fn, void *cb_arg) 3721 { 3722 _spdk_blob_request_submit_rw_iov(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg, false); 3723 } 3724 3725 void spdk_blob_io_readv(struct spdk_blob *blob, struct spdk_io_channel *channel, 3726 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 3727 spdk_blob_op_complete cb_fn, void *cb_arg) 3728 { 3729 _spdk_blob_request_submit_rw_iov(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg, true); 3730 } 3731 3732 void spdk_bs_io_unmap_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 3733 uint64_t offset, uint64_t length, spdk_blob_op_complete cb_fn, void *cb_arg) 3734 { 3735 spdk_blob_io_unmap(blob, channel, offset, length, cb_fn, cb_arg); 3736 } 3737 3738 void spdk_bs_io_write_zeroes_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 3739 uint64_t offset, uint64_t length, spdk_blob_op_complete cb_fn, void *cb_arg) 3740 { 3741 spdk_blob_io_write_zeroes(blob, channel, offset, length, cb_fn, cb_arg); 3742 } 3743 3744 void spdk_bs_io_write_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 3745 void *payload, uint64_t offset, uint64_t length, 3746 spdk_blob_op_complete cb_fn, void *cb_arg) 3747 { 3748 spdk_blob_io_write(blob, channel, payload, offset, length, cb_fn, cb_arg); 3749 } 3750 3751 void spdk_bs_io_read_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 3752 void *payload, uint64_t offset, uint64_t length, 3753 spdk_blob_op_complete cb_fn, void *cb_arg) 3754 { 3755 spdk_blob_io_read(blob, channel, payload, offset, length, cb_fn, cb_arg); 3756 } 3757 3758 void spdk_bs_io_writev_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 3759 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 3760 spdk_blob_op_complete cb_fn, void *cb_arg) 3761 { 3762 spdk_blob_io_writev(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg); 3763 } 3764 3765 void spdk_bs_io_readv_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 3766 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 3767 spdk_blob_op_complete cb_fn, void *cb_arg) 3768 { 3769 spdk_blob_io_readv(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg); 3770 } 3771 3772 struct spdk_bs_iter_ctx { 3773 int64_t page_num; 3774 struct spdk_blob_store *bs; 3775 3776 spdk_blob_op_with_handle_complete cb_fn; 3777 void *cb_arg; 3778 }; 3779 3780 static void 3781 _spdk_bs_iter_cpl(void *cb_arg, struct spdk_blob *_blob, int bserrno) 3782 { 3783 struct spdk_bs_iter_ctx *ctx = cb_arg; 3784 struct spdk_blob_store *bs = ctx->bs; 3785 spdk_blob_id id; 3786 3787 if (bserrno == 0) { 3788 ctx->cb_fn(ctx->cb_arg, _blob, bserrno); 3789 free(ctx); 3790 return; 3791 } 3792 3793 ctx->page_num++; 3794 ctx->page_num = spdk_bit_array_find_first_set(bs->used_blobids, ctx->page_num); 3795 if (ctx->page_num >= spdk_bit_array_capacity(bs->used_blobids)) { 3796 ctx->cb_fn(ctx->cb_arg, NULL, -ENOENT); 3797 free(ctx); 3798 return; 3799 } 3800 3801 id = _spdk_bs_page_to_blobid(ctx->page_num); 3802 3803 spdk_bs_open_blob(bs, id, _spdk_bs_iter_cpl, ctx); 3804 } 3805 3806 void 3807 spdk_bs_iter_first(struct spdk_blob_store *bs, 3808 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 3809 { 3810 struct spdk_bs_iter_ctx *ctx; 3811 3812 ctx = calloc(1, sizeof(*ctx)); 3813 if (!ctx) { 3814 cb_fn(cb_arg, NULL, -ENOMEM); 3815 return; 3816 } 3817 3818 ctx->page_num = -1; 3819 ctx->bs = bs; 3820 ctx->cb_fn = cb_fn; 3821 ctx->cb_arg = cb_arg; 3822 3823 _spdk_bs_iter_cpl(ctx, NULL, -1); 3824 } 3825 3826 static void 3827 _spdk_bs_iter_close_cpl(void *cb_arg, int bserrno) 3828 { 3829 struct spdk_bs_iter_ctx *ctx = cb_arg; 3830 3831 _spdk_bs_iter_cpl(ctx, NULL, -1); 3832 } 3833 3834 void 3835 spdk_bs_iter_next(struct spdk_blob_store *bs, struct spdk_blob *blob, 3836 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 3837 { 3838 struct spdk_bs_iter_ctx *ctx; 3839 3840 assert(blob != NULL); 3841 3842 ctx = calloc(1, sizeof(*ctx)); 3843 if (!ctx) { 3844 cb_fn(cb_arg, NULL, -ENOMEM); 3845 return; 3846 } 3847 3848 ctx->page_num = _spdk_bs_blobid_to_page(blob->id); 3849 ctx->bs = bs; 3850 ctx->cb_fn = cb_fn; 3851 ctx->cb_arg = cb_arg; 3852 3853 /* Close the existing blob */ 3854 spdk_blob_close(blob, _spdk_bs_iter_close_cpl, ctx); 3855 } 3856 3857 static int 3858 _spdk_blob_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 3859 uint16_t value_len, bool internal) 3860 { 3861 struct spdk_xattr_tailq *xattrs; 3862 struct spdk_xattr *xattr; 3863 3864 _spdk_blob_verify_md_op(blob); 3865 3866 if (blob->md_ro) { 3867 return -EPERM; 3868 } 3869 3870 if (internal) { 3871 xattrs = &blob->xattrs_internal; 3872 blob->invalid_flags |= SPDK_BLOB_INTERNAL_XATTR; 3873 } else { 3874 xattrs = &blob->xattrs; 3875 } 3876 3877 TAILQ_FOREACH(xattr, xattrs, link) { 3878 if (!strcmp(name, xattr->name)) { 3879 free(xattr->value); 3880 xattr->value_len = value_len; 3881 xattr->value = malloc(value_len); 3882 memcpy(xattr->value, value, value_len); 3883 3884 blob->state = SPDK_BLOB_STATE_DIRTY; 3885 3886 return 0; 3887 } 3888 } 3889 3890 xattr = calloc(1, sizeof(*xattr)); 3891 if (!xattr) { 3892 return -1; 3893 } 3894 xattr->name = strdup(name); 3895 xattr->value_len = value_len; 3896 xattr->value = malloc(value_len); 3897 memcpy(xattr->value, value, value_len); 3898 TAILQ_INSERT_TAIL(xattrs, xattr, link); 3899 3900 blob->state = SPDK_BLOB_STATE_DIRTY; 3901 3902 return 0; 3903 } 3904 3905 int 3906 spdk_blob_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 3907 uint16_t value_len) 3908 { 3909 return _spdk_blob_set_xattr(blob, name, value, value_len, false); 3910 } 3911 3912 static int 3913 _spdk_blob_remove_xattr(struct spdk_blob *blob, const char *name, bool internal) 3914 { 3915 struct spdk_xattr_tailq *xattrs; 3916 struct spdk_xattr *xattr; 3917 3918 _spdk_blob_verify_md_op(blob); 3919 3920 if (blob->md_ro) { 3921 return -EPERM; 3922 } 3923 xattrs = internal ? &blob->xattrs_internal : &blob->xattrs; 3924 3925 TAILQ_FOREACH(xattr, xattrs, link) { 3926 if (!strcmp(name, xattr->name)) { 3927 TAILQ_REMOVE(xattrs, xattr, link); 3928 free(xattr->value); 3929 free(xattr->name); 3930 free(xattr); 3931 3932 if (internal && TAILQ_EMPTY(&blob->xattrs_internal)) { 3933 blob->invalid_flags &= ~SPDK_BLOB_INTERNAL_XATTR; 3934 } 3935 blob->state = SPDK_BLOB_STATE_DIRTY; 3936 3937 return 0; 3938 } 3939 } 3940 3941 return -ENOENT; 3942 } 3943 3944 int 3945 spdk_blob_remove_xattr(struct spdk_blob *blob, const char *name) 3946 { 3947 return _spdk_blob_remove_xattr(blob, name, false); 3948 } 3949 3950 static int 3951 _spdk_blob_get_xattr_value(struct spdk_blob *blob, const char *name, 3952 const void **value, size_t *value_len, bool internal) 3953 { 3954 struct spdk_xattr *xattr; 3955 struct spdk_xattr_tailq *xattrs; 3956 3957 _spdk_blob_verify_md_op(blob); 3958 3959 xattrs = internal ? &blob->xattrs_internal : &blob->xattrs; 3960 3961 TAILQ_FOREACH(xattr, xattrs, link) { 3962 if (!strcmp(name, xattr->name)) { 3963 *value = xattr->value; 3964 *value_len = xattr->value_len; 3965 return 0; 3966 } 3967 } 3968 return -ENOENT; 3969 } 3970 3971 int 3972 spdk_blob_get_xattr_value(struct spdk_blob *blob, const char *name, 3973 const void **value, size_t *value_len) 3974 { 3975 return _spdk_blob_get_xattr_value(blob, name, value, value_len, false); 3976 } 3977 3978 struct spdk_xattr_names { 3979 uint32_t count; 3980 const char *names[0]; 3981 }; 3982 3983 static int 3984 _spdk_blob_get_xattr_names(struct spdk_xattr_tailq *xattrs, struct spdk_xattr_names **names) 3985 { 3986 struct spdk_xattr *xattr; 3987 int count = 0; 3988 3989 TAILQ_FOREACH(xattr, xattrs, link) { 3990 count++; 3991 } 3992 3993 *names = calloc(1, sizeof(struct spdk_xattr_names) + count * sizeof(char *)); 3994 if (*names == NULL) { 3995 return -ENOMEM; 3996 } 3997 3998 TAILQ_FOREACH(xattr, xattrs, link) { 3999 (*names)->names[(*names)->count++] = xattr->name; 4000 } 4001 4002 return 0; 4003 } 4004 4005 int 4006 spdk_blob_get_xattr_names(struct spdk_blob *blob, struct spdk_xattr_names **names) 4007 { 4008 _spdk_blob_verify_md_op(blob); 4009 4010 return _spdk_blob_get_xattr_names(&blob->xattrs, names); 4011 } 4012 4013 uint32_t 4014 spdk_xattr_names_get_count(struct spdk_xattr_names *names) 4015 { 4016 assert(names != NULL); 4017 4018 return names->count; 4019 } 4020 4021 const char * 4022 spdk_xattr_names_get_name(struct spdk_xattr_names *names, uint32_t index) 4023 { 4024 if (index >= names->count) { 4025 return NULL; 4026 } 4027 4028 return names->names[index]; 4029 } 4030 4031 void 4032 spdk_xattr_names_free(struct spdk_xattr_names *names) 4033 { 4034 free(names); 4035 } 4036 4037 struct spdk_bs_type 4038 spdk_bs_get_bstype(struct spdk_blob_store *bs) 4039 { 4040 return bs->bstype; 4041 } 4042 4043 void 4044 spdk_bs_set_bstype(struct spdk_blob_store *bs, struct spdk_bs_type bstype) 4045 { 4046 memcpy(&bs->bstype, &bstype, sizeof(bstype)); 4047 } 4048 4049 SPDK_LOG_REGISTER_COMPONENT("blob", SPDK_LOG_BLOB) 4050