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