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