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 inline size_t 51 divide_round_up(size_t num, size_t divisor) 52 { 53 return (num + divisor - 1) / divisor; 54 } 55 56 static void 57 _spdk_bs_claim_cluster(struct spdk_blob_store *bs, uint32_t cluster_num) 58 { 59 assert(cluster_num < spdk_bit_array_capacity(bs->used_clusters)); 60 assert(spdk_bit_array_get(bs->used_clusters, cluster_num) == false); 61 assert(bs->num_free_clusters > 0); 62 63 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Claiming cluster %u\n", cluster_num); 64 65 spdk_bit_array_set(bs->used_clusters, cluster_num); 66 bs->num_free_clusters--; 67 } 68 69 static void 70 _spdk_bs_release_cluster(struct spdk_blob_store *bs, uint32_t cluster_num) 71 { 72 assert(cluster_num < spdk_bit_array_capacity(bs->used_clusters)); 73 assert(spdk_bit_array_get(bs->used_clusters, cluster_num) == true); 74 assert(bs->num_free_clusters < bs->total_clusters); 75 76 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Releasing cluster %u\n", cluster_num); 77 78 spdk_bit_array_clear(bs->used_clusters, cluster_num); 79 bs->num_free_clusters++; 80 } 81 82 static struct spdk_blob * 83 _spdk_blob_alloc(struct spdk_blob_store *bs, spdk_blob_id id) 84 { 85 struct spdk_blob *blob; 86 87 blob = calloc(1, sizeof(*blob)); 88 if (!blob) { 89 return NULL; 90 } 91 92 blob->id = id; 93 blob->bs = bs; 94 95 blob->state = SPDK_BLOB_STATE_DIRTY; 96 blob->active.num_pages = 1; 97 blob->active.pages = calloc(1, sizeof(*blob->active.pages)); 98 if (!blob->active.pages) { 99 free(blob); 100 return NULL; 101 } 102 103 blob->active.pages[0] = _spdk_bs_blobid_to_page(id); 104 105 TAILQ_INIT(&blob->xattrs); 106 107 return blob; 108 } 109 110 static void 111 _spdk_blob_free(struct spdk_blob *blob) 112 { 113 struct spdk_xattr *xattr, *xattr_tmp; 114 115 assert(blob != NULL); 116 117 free(blob->active.clusters); 118 free(blob->clean.clusters); 119 free(blob->active.pages); 120 free(blob->clean.pages); 121 122 TAILQ_FOREACH_SAFE(xattr, &blob->xattrs, link, xattr_tmp) { 123 TAILQ_REMOVE(&blob->xattrs, xattr, link); 124 free(xattr->name); 125 free(xattr->value); 126 free(xattr); 127 } 128 129 free(blob); 130 } 131 132 static int 133 _spdk_blob_mark_clean(struct spdk_blob *blob) 134 { 135 uint64_t *clusters = NULL; 136 uint32_t *pages = NULL; 137 138 assert(blob != NULL); 139 assert(blob->state == SPDK_BLOB_STATE_LOADING || 140 blob->state == SPDK_BLOB_STATE_SYNCING); 141 142 if (blob->active.num_clusters) { 143 assert(blob->active.clusters); 144 clusters = calloc(blob->active.num_clusters, sizeof(*blob->active.clusters)); 145 if (!clusters) { 146 return -1; 147 } 148 memcpy(clusters, blob->active.clusters, blob->active.num_clusters * sizeof(*clusters)); 149 } 150 151 if (blob->active.num_pages) { 152 assert(blob->active.pages); 153 pages = calloc(blob->active.num_pages, sizeof(*blob->active.pages)); 154 if (!pages) { 155 free(clusters); 156 return -1; 157 } 158 memcpy(pages, blob->active.pages, blob->active.num_pages * sizeof(*pages)); 159 } 160 161 free(blob->clean.clusters); 162 free(blob->clean.pages); 163 164 blob->clean.num_clusters = blob->active.num_clusters; 165 blob->clean.clusters = blob->active.clusters; 166 blob->clean.num_pages = blob->active.num_pages; 167 blob->clean.pages = blob->active.pages; 168 169 blob->active.clusters = clusters; 170 blob->active.pages = pages; 171 172 blob->state = SPDK_BLOB_STATE_CLEAN; 173 174 return 0; 175 } 176 177 static void 178 _spdk_blob_parse_page(const struct spdk_blob_md_page *page, struct spdk_blob *blob) 179 { 180 struct spdk_blob_md_descriptor *desc; 181 size_t cur_desc = 0; 182 void *tmp; 183 184 desc = (struct spdk_blob_md_descriptor *)page->descriptors; 185 while (cur_desc < sizeof(page->descriptors)) { 186 if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_PADDING) { 187 if (desc->length == 0) { 188 /* If padding and length are 0, this terminates the page */ 189 break; 190 } 191 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_EXTENT) { 192 struct spdk_blob_md_descriptor_extent *desc_extent; 193 unsigned int i, j; 194 unsigned int cluster_count = blob->active.num_clusters; 195 196 desc_extent = (struct spdk_blob_md_descriptor_extent *)desc; 197 198 assert(desc_extent->length > 0); 199 assert(desc_extent->length % sizeof(desc_extent->extents[0]) == 0); 200 201 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 202 for (j = 0; j < desc_extent->extents[i].length; j++) { 203 assert(spdk_bit_array_get(blob->bs->used_clusters, desc_extent->extents[i].cluster_idx + j)); 204 cluster_count++; 205 } 206 } 207 208 assert(cluster_count > 0); 209 tmp = realloc(blob->active.clusters, cluster_count * sizeof(uint64_t)); 210 assert(tmp != NULL); 211 blob->active.clusters = tmp; 212 blob->active.cluster_array_size = cluster_count; 213 214 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 215 for (j = 0; j < desc_extent->extents[i].length; j++) { 216 blob->active.clusters[blob->active.num_clusters++] = _spdk_bs_cluster_to_lba(blob->bs, 217 desc_extent->extents[i].cluster_idx + j); 218 } 219 } 220 221 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR) { 222 struct spdk_blob_md_descriptor_xattr *desc_xattr; 223 struct spdk_xattr *xattr; 224 225 desc_xattr = (struct spdk_blob_md_descriptor_xattr *)desc; 226 227 assert(desc_xattr->length == sizeof(desc_xattr->name_length) + 228 sizeof(desc_xattr->value_length) + 229 desc_xattr->name_length + desc_xattr->value_length); 230 231 xattr = calloc(1, sizeof(*xattr)); 232 assert(xattr != NULL); 233 234 xattr->name = malloc(desc_xattr->name_length + 1); 235 assert(xattr->name); 236 strncpy(xattr->name, desc_xattr->name, desc_xattr->name_length); 237 xattr->name[desc_xattr->name_length] = '\0'; 238 239 xattr->value = malloc(desc_xattr->value_length); 240 assert(xattr->value != NULL); 241 xattr->value_len = desc_xattr->value_length; 242 memcpy(xattr->value, 243 (void *)((uintptr_t)desc_xattr->name + desc_xattr->name_length), 244 desc_xattr->value_length); 245 246 TAILQ_INSERT_TAIL(&blob->xattrs, xattr, link); 247 } else { 248 /* Error */ 249 break; 250 } 251 252 /* Advance to the next descriptor */ 253 cur_desc += sizeof(*desc) + desc->length; 254 if (cur_desc + sizeof(*desc) > sizeof(page->descriptors)) { 255 break; 256 } 257 desc = (struct spdk_blob_md_descriptor *)((uintptr_t)page->descriptors + cur_desc); 258 } 259 } 260 261 static int 262 _spdk_blob_parse(const struct spdk_blob_md_page *pages, uint32_t page_count, 263 struct spdk_blob *blob) 264 { 265 const struct spdk_blob_md_page *page; 266 uint32_t i; 267 268 assert(page_count > 0); 269 assert(pages[0].sequence_num == 0); 270 assert(blob != NULL); 271 assert(blob->state == SPDK_BLOB_STATE_LOADING); 272 assert(blob->active.clusters == NULL); 273 assert(blob->state == SPDK_BLOB_STATE_LOADING); 274 275 /* The blobid provided doesn't match what's in the MD, this can 276 * happen for example if a bogus blobid is passed in through open. 277 */ 278 if (blob->id != pages[0].id) { 279 SPDK_ERRLOG("Blobid (%lu) doesn't match what's in metadata (%lu)\n", 280 blob->id, pages[0].id); 281 return -ENOENT; 282 } 283 284 for (i = 0; i < page_count; i++) { 285 page = &pages[i]; 286 287 assert(page->id == blob->id); 288 assert(page->sequence_num == i); 289 290 _spdk_blob_parse_page(page, blob); 291 } 292 293 return 0; 294 } 295 296 static int 297 _spdk_blob_serialize_add_page(const struct spdk_blob *blob, 298 struct spdk_blob_md_page **pages, 299 uint32_t *page_count, 300 struct spdk_blob_md_page **last_page) 301 { 302 struct spdk_blob_md_page *page; 303 304 assert(pages != NULL); 305 assert(page_count != NULL); 306 307 if (*page_count == 0) { 308 assert(*pages == NULL); 309 *page_count = 1; 310 *pages = spdk_dma_malloc(SPDK_BS_PAGE_SIZE, 311 SPDK_BS_PAGE_SIZE, 312 NULL); 313 } else { 314 assert(*pages != NULL); 315 (*page_count)++; 316 *pages = spdk_dma_realloc(*pages, 317 SPDK_BS_PAGE_SIZE * (*page_count), 318 SPDK_BS_PAGE_SIZE, 319 NULL); 320 } 321 322 if (*pages == NULL) { 323 *page_count = 0; 324 *last_page = NULL; 325 return -ENOMEM; 326 } 327 328 page = &(*pages)[*page_count - 1]; 329 memset(page, 0, sizeof(*page)); 330 page->id = blob->id; 331 page->sequence_num = *page_count - 1; 332 page->next = SPDK_INVALID_MD_PAGE; 333 *last_page = page; 334 335 return 0; 336 } 337 338 /* Transform the in-memory representation 'xattr' into an on-disk xattr descriptor. 339 * Update required_sz on both success and failure. 340 * 341 */ 342 static int 343 _spdk_blob_serialize_xattr(const struct spdk_xattr *xattr, 344 uint8_t *buf, size_t buf_sz, 345 size_t *required_sz) 346 { 347 struct spdk_blob_md_descriptor_xattr *desc; 348 349 *required_sz = sizeof(struct spdk_blob_md_descriptor_xattr) + 350 strlen(xattr->name) + 351 xattr->value_len; 352 353 if (buf_sz < *required_sz) { 354 return -1; 355 } 356 357 desc = (struct spdk_blob_md_descriptor_xattr *)buf; 358 359 desc->type = SPDK_MD_DESCRIPTOR_TYPE_XATTR; 360 desc->length = sizeof(desc->name_length) + 361 sizeof(desc->value_length) + 362 strlen(xattr->name) + 363 xattr->value_len; 364 desc->name_length = strlen(xattr->name); 365 desc->value_length = xattr->value_len; 366 367 memcpy(desc->name, xattr->name, desc->name_length); 368 memcpy((void *)((uintptr_t)desc->name + desc->name_length), 369 xattr->value, 370 desc->value_length); 371 372 return 0; 373 } 374 375 static void 376 _spdk_blob_serialize_extent(const struct spdk_blob *blob, 377 uint64_t start_cluster, uint64_t *next_cluster, 378 uint8_t *buf, size_t buf_sz) 379 { 380 struct spdk_blob_md_descriptor_extent *desc; 381 size_t cur_sz; 382 uint64_t i, extent_idx; 383 uint32_t lba, lba_per_cluster, lba_count; 384 385 /* The buffer must have room for at least one extent */ 386 cur_sz = sizeof(struct spdk_blob_md_descriptor) + sizeof(desc->extents[0]); 387 if (buf_sz < cur_sz) { 388 *next_cluster = start_cluster; 389 return; 390 } 391 392 desc = (struct spdk_blob_md_descriptor_extent *)buf; 393 desc->type = SPDK_MD_DESCRIPTOR_TYPE_EXTENT; 394 395 lba_per_cluster = _spdk_bs_cluster_to_lba(blob->bs, 1); 396 397 lba = blob->active.clusters[start_cluster]; 398 lba_count = lba_per_cluster; 399 extent_idx = 0; 400 for (i = start_cluster + 1; i < blob->active.num_clusters; i++) { 401 if ((lba + lba_count) == blob->active.clusters[i]) { 402 lba_count += lba_per_cluster; 403 continue; 404 } 405 desc->extents[extent_idx].cluster_idx = lba / lba_per_cluster; 406 desc->extents[extent_idx].length = lba_count / lba_per_cluster; 407 extent_idx++; 408 409 cur_sz += sizeof(desc->extents[extent_idx]); 410 411 if (buf_sz < cur_sz) { 412 /* If we ran out of buffer space, return */ 413 desc->length = sizeof(desc->extents[0]) * extent_idx; 414 *next_cluster = i; 415 return; 416 } 417 418 lba = blob->active.clusters[i]; 419 lba_count = lba_per_cluster; 420 } 421 422 desc->extents[extent_idx].cluster_idx = lba / lba_per_cluster; 423 desc->extents[extent_idx].length = lba_count / lba_per_cluster; 424 extent_idx++; 425 426 desc->length = sizeof(desc->extents[0]) * extent_idx; 427 *next_cluster = blob->active.num_clusters; 428 429 return; 430 } 431 432 static int 433 _spdk_blob_serialize(const struct spdk_blob *blob, struct spdk_blob_md_page **pages, 434 uint32_t *page_count) 435 { 436 struct spdk_blob_md_page *cur_page; 437 const struct spdk_xattr *xattr; 438 int rc; 439 uint8_t *buf; 440 size_t remaining_sz; 441 uint64_t last_cluster; 442 443 assert(pages != NULL); 444 assert(page_count != NULL); 445 assert(blob != NULL); 446 assert(blob->state == SPDK_BLOB_STATE_SYNCING); 447 448 *pages = NULL; 449 *page_count = 0; 450 451 /* A blob always has at least 1 page, even if it has no descriptors */ 452 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, &cur_page); 453 if (rc < 0) { 454 return rc; 455 } 456 457 buf = (uint8_t *)cur_page->descriptors; 458 remaining_sz = sizeof(cur_page->descriptors); 459 460 /* Serialize xattrs */ 461 TAILQ_FOREACH(xattr, &blob->xattrs, link) { 462 size_t required_sz = 0; 463 rc = _spdk_blob_serialize_xattr(xattr, 464 buf, remaining_sz, 465 &required_sz); 466 if (rc < 0) { 467 /* Need to add a new page to the chain */ 468 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, 469 &cur_page); 470 if (rc < 0) { 471 spdk_dma_free(*pages); 472 *pages = NULL; 473 *page_count = 0; 474 return rc; 475 } 476 477 buf = (uint8_t *)cur_page->descriptors; 478 remaining_sz = sizeof(cur_page->descriptors); 479 480 /* Try again */ 481 required_sz = 0; 482 rc = _spdk_blob_serialize_xattr(xattr, 483 buf, remaining_sz, 484 &required_sz); 485 486 if (rc < 0) { 487 spdk_dma_free(*pages); 488 *pages = NULL; 489 *page_count = 0; 490 return -1; 491 } 492 } 493 494 remaining_sz -= required_sz; 495 buf += required_sz; 496 } 497 498 /* Serialize extents */ 499 last_cluster = 0; 500 while (last_cluster < blob->active.num_clusters) { 501 _spdk_blob_serialize_extent(blob, last_cluster, &last_cluster, 502 buf, remaining_sz); 503 504 if (last_cluster == blob->active.num_clusters) { 505 break; 506 } 507 508 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, 509 &cur_page); 510 if (rc < 0) { 511 return rc; 512 } 513 514 buf = (uint8_t *)cur_page->descriptors; 515 remaining_sz = sizeof(cur_page->descriptors); 516 } 517 518 return 0; 519 } 520 521 struct spdk_blob_load_ctx { 522 struct spdk_blob *blob; 523 524 struct spdk_blob_md_page *pages; 525 uint32_t num_pages; 526 527 spdk_bs_sequence_cpl cb_fn; 528 void *cb_arg; 529 }; 530 531 static uint32_t 532 _spdk_blob_md_page_calc_crc(void *page) 533 { 534 uint32_t crc; 535 536 crc = BLOB_CRC32C_INITIAL; 537 crc = spdk_crc32c_update(page, SPDK_BS_PAGE_SIZE - 4, crc); 538 crc ^= BLOB_CRC32C_INITIAL; 539 540 return crc; 541 542 } 543 544 static void 545 _spdk_blob_load_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 546 { 547 struct spdk_blob_load_ctx *ctx = cb_arg; 548 struct spdk_blob *blob = ctx->blob; 549 struct spdk_blob_md_page *page; 550 int rc; 551 uint32_t crc; 552 553 page = &ctx->pages[ctx->num_pages - 1]; 554 crc = _spdk_blob_md_page_calc_crc(page); 555 if (crc != page->crc) { 556 SPDK_ERRLOG("Metadata page %d crc mismatch\n", ctx->num_pages); 557 _spdk_blob_free(blob); 558 ctx->cb_fn(seq, NULL, -EINVAL); 559 spdk_dma_free(ctx->pages); 560 free(ctx); 561 return; 562 } 563 564 if (page->next != SPDK_INVALID_MD_PAGE) { 565 uint32_t next_page = page->next; 566 uint64_t next_lba = _spdk_bs_page_to_lba(blob->bs, blob->bs->md_start + next_page); 567 568 569 assert(next_lba < (blob->bs->md_start + blob->bs->md_len)); 570 571 /* Read the next page */ 572 ctx->num_pages++; 573 ctx->pages = spdk_dma_realloc(ctx->pages, (sizeof(*page) * ctx->num_pages), 574 sizeof(*page), NULL); 575 if (ctx->pages == NULL) { 576 ctx->cb_fn(seq, ctx->cb_arg, -ENOMEM); 577 free(ctx); 578 return; 579 } 580 581 spdk_bs_sequence_read(seq, &ctx->pages[ctx->num_pages - 1], 582 next_lba, 583 _spdk_bs_byte_to_lba(blob->bs, sizeof(*page)), 584 _spdk_blob_load_cpl, ctx); 585 return; 586 } 587 588 /* Parse the pages */ 589 rc = _spdk_blob_parse(ctx->pages, ctx->num_pages, blob); 590 if (rc) { 591 _spdk_blob_free(blob); 592 ctx->cb_fn(seq, NULL, rc); 593 spdk_dma_free(ctx->pages); 594 free(ctx); 595 return; 596 } 597 598 _spdk_blob_mark_clean(blob); 599 600 ctx->cb_fn(seq, ctx->cb_arg, rc); 601 602 /* Free the memory */ 603 spdk_dma_free(ctx->pages); 604 free(ctx); 605 } 606 607 /* Load a blob from disk given a blobid */ 608 static void 609 _spdk_blob_load(spdk_bs_sequence_t *seq, struct spdk_blob *blob, 610 spdk_bs_sequence_cpl cb_fn, void *cb_arg) 611 { 612 struct spdk_blob_load_ctx *ctx; 613 struct spdk_blob_store *bs; 614 uint32_t page_num; 615 uint64_t lba; 616 617 assert(blob != NULL); 618 assert(blob->state == SPDK_BLOB_STATE_CLEAN || 619 blob->state == SPDK_BLOB_STATE_DIRTY); 620 621 bs = blob->bs; 622 623 ctx = calloc(1, sizeof(*ctx)); 624 if (!ctx) { 625 cb_fn(seq, cb_arg, -ENOMEM); 626 return; 627 } 628 629 ctx->blob = blob; 630 ctx->pages = spdk_dma_realloc(ctx->pages, SPDK_BS_PAGE_SIZE, 631 SPDK_BS_PAGE_SIZE, NULL); 632 if (!ctx->pages) { 633 free(ctx); 634 cb_fn(seq, cb_arg, -ENOMEM); 635 return; 636 } 637 ctx->num_pages = 1; 638 ctx->cb_fn = cb_fn; 639 ctx->cb_arg = cb_arg; 640 641 page_num = _spdk_bs_blobid_to_page(blob->id); 642 lba = _spdk_bs_page_to_lba(blob->bs, bs->md_start + page_num); 643 644 blob->state = SPDK_BLOB_STATE_LOADING; 645 646 spdk_bs_sequence_read(seq, &ctx->pages[0], lba, 647 _spdk_bs_byte_to_lba(bs, SPDK_BS_PAGE_SIZE), 648 _spdk_blob_load_cpl, ctx); 649 } 650 651 struct spdk_blob_persist_ctx { 652 struct spdk_blob *blob; 653 654 struct spdk_blob_md_page *pages; 655 656 uint64_t idx; 657 658 spdk_bs_sequence_cpl cb_fn; 659 void *cb_arg; 660 }; 661 662 static void 663 _spdk_blob_persist_complete(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 664 { 665 struct spdk_blob_persist_ctx *ctx = cb_arg; 666 struct spdk_blob *blob = ctx->blob; 667 668 if (bserrno == 0) { 669 _spdk_blob_mark_clean(blob); 670 } 671 672 /* Call user callback */ 673 ctx->cb_fn(seq, ctx->cb_arg, bserrno); 674 675 /* Free the memory */ 676 spdk_dma_free(ctx->pages); 677 free(ctx); 678 } 679 680 static void 681 _spdk_blob_persist_unmap_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 682 { 683 struct spdk_blob_persist_ctx *ctx = cb_arg; 684 struct spdk_blob *blob = ctx->blob; 685 struct spdk_blob_store *bs = blob->bs; 686 void *tmp; 687 size_t i; 688 689 /* Release all clusters that were truncated */ 690 for (i = blob->active.num_clusters; i < blob->active.cluster_array_size; i++) { 691 uint32_t cluster_num = _spdk_bs_lba_to_cluster(bs, blob->active.clusters[i]); 692 693 _spdk_bs_release_cluster(bs, cluster_num); 694 } 695 696 if (blob->active.num_clusters == 0) { 697 free(blob->active.clusters); 698 blob->active.clusters = NULL; 699 blob->active.cluster_array_size = 0; 700 } else { 701 tmp = realloc(blob->active.clusters, sizeof(uint64_t) * blob->active.num_clusters); 702 assert(tmp != NULL); 703 blob->active.clusters = tmp; 704 blob->active.cluster_array_size = blob->active.num_clusters; 705 } 706 707 _spdk_blob_persist_complete(seq, ctx, bserrno); 708 } 709 710 static void 711 _spdk_blob_persist_unmap_clusters(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 712 { 713 struct spdk_blob_persist_ctx *ctx = cb_arg; 714 struct spdk_blob *blob = ctx->blob; 715 struct spdk_blob_store *bs = blob->bs; 716 spdk_bs_batch_t *batch; 717 size_t i; 718 uint64_t lba; 719 uint32_t lba_count; 720 721 /* Clusters don't move around in blobs. The list shrinks or grows 722 * at the end, but no changes ever occur in the middle of the list. 723 */ 724 725 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_unmap_clusters_cpl, ctx); 726 727 /* Unmap all clusters that were truncated */ 728 lba = 0; 729 lba_count = 0; 730 for (i = blob->active.num_clusters; i < blob->active.cluster_array_size; i++) { 731 uint64_t next_lba = blob->active.clusters[i]; 732 uint32_t next_lba_count = _spdk_bs_cluster_to_lba(bs, 1); 733 734 if ((lba + lba_count) == next_lba) { 735 /* This cluster is contiguous with the previous one. */ 736 lba_count += next_lba_count; 737 continue; 738 } 739 740 /* This cluster is not contiguous with the previous one. */ 741 742 /* If a run of LBAs previously existing, send them 743 * as an unmap. 744 */ 745 if (lba_count > 0) { 746 spdk_bs_batch_unmap(batch, lba, lba_count); 747 } 748 749 /* Start building the next batch */ 750 lba = next_lba; 751 lba_count = next_lba_count; 752 } 753 754 /* If we ended with a contiguous set of LBAs, send the unmap now */ 755 if (lba_count > 0) { 756 spdk_bs_batch_unmap(batch, lba, lba_count); 757 } 758 759 spdk_bs_batch_close(batch); 760 } 761 762 static void 763 _spdk_blob_persist_unmap_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 764 { 765 struct spdk_blob_persist_ctx *ctx = cb_arg; 766 struct spdk_blob *blob = ctx->blob; 767 struct spdk_blob_store *bs = blob->bs; 768 size_t i; 769 770 /* This loop starts at 1 because the first page is special and handled 771 * below. The pages (except the first) are never written in place, 772 * so any pages in the clean list must be unmapped. 773 */ 774 for (i = 1; i < blob->clean.num_pages; i++) { 775 spdk_bit_array_clear(bs->used_md_pages, blob->clean.pages[i]); 776 } 777 778 if (blob->active.num_pages == 0) { 779 uint32_t page_num; 780 781 page_num = _spdk_bs_blobid_to_page(blob->id); 782 spdk_bit_array_clear(bs->used_md_pages, page_num); 783 } 784 785 /* Move on to unmapping clusters */ 786 _spdk_blob_persist_unmap_clusters(seq, ctx, 0); 787 } 788 789 static void 790 _spdk_blob_persist_unmap_pages(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 791 { 792 struct spdk_blob_persist_ctx *ctx = cb_arg; 793 struct spdk_blob *blob = ctx->blob; 794 struct spdk_blob_store *bs = blob->bs; 795 uint64_t lba; 796 uint32_t lba_count; 797 spdk_bs_batch_t *batch; 798 size_t i; 799 800 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_unmap_pages_cpl, ctx); 801 802 lba_count = _spdk_bs_byte_to_lba(bs, SPDK_BS_PAGE_SIZE); 803 804 /* This loop starts at 1 because the first page is special and handled 805 * below. The pages (except the first) are never written in place, 806 * so any pages in the clean list must be unmapped. 807 */ 808 for (i = 1; i < blob->clean.num_pages; i++) { 809 lba = _spdk_bs_page_to_lba(bs, bs->md_start + blob->clean.pages[i]); 810 811 spdk_bs_batch_unmap(batch, lba, lba_count); 812 } 813 814 /* The first page will only be unmapped if this is a delete. */ 815 if (blob->active.num_pages == 0) { 816 uint32_t page_num; 817 818 /* The first page in the metadata goes where the blobid indicates */ 819 page_num = _spdk_bs_blobid_to_page(blob->id); 820 lba = _spdk_bs_page_to_lba(bs, bs->md_start + page_num); 821 822 spdk_bs_batch_unmap(batch, lba, lba_count); 823 } 824 825 spdk_bs_batch_close(batch); 826 } 827 828 static void 829 _spdk_blob_persist_write_page_root(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 830 { 831 struct spdk_blob_persist_ctx *ctx = cb_arg; 832 struct spdk_blob *blob = ctx->blob; 833 struct spdk_blob_store *bs = blob->bs; 834 uint64_t lba; 835 uint32_t lba_count; 836 struct spdk_blob_md_page *page; 837 838 if (blob->active.num_pages == 0) { 839 /* Move on to the next step */ 840 _spdk_blob_persist_unmap_pages(seq, ctx, 0); 841 return; 842 } 843 844 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(*page)); 845 846 page = &ctx->pages[0]; 847 /* The first page in the metadata goes where the blobid indicates */ 848 lba = _spdk_bs_page_to_lba(bs, bs->md_start + _spdk_bs_blobid_to_page(blob->id)); 849 850 spdk_bs_sequence_write(seq, page, lba, lba_count, 851 _spdk_blob_persist_unmap_pages, ctx); 852 } 853 854 static void 855 _spdk_blob_persist_write_page_chain(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 856 { 857 struct spdk_blob_persist_ctx *ctx = cb_arg; 858 struct spdk_blob *blob = ctx->blob; 859 struct spdk_blob_store *bs = blob->bs; 860 uint64_t lba; 861 uint32_t lba_count; 862 struct spdk_blob_md_page *page; 863 spdk_bs_batch_t *batch; 864 size_t i; 865 866 /* Clusters don't move around in blobs. The list shrinks or grows 867 * at the end, but no changes ever occur in the middle of the list. 868 */ 869 870 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(*page)); 871 872 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_write_page_root, ctx); 873 874 /* This starts at 1. The root page is not written until 875 * all of the others are finished 876 */ 877 for (i = 1; i < blob->active.num_pages; i++) { 878 page = &ctx->pages[i]; 879 assert(page->sequence_num == i); 880 881 lba = _spdk_bs_page_to_lba(bs, bs->md_start + blob->active.pages[i]); 882 883 spdk_bs_batch_write(batch, page, lba, lba_count); 884 } 885 886 spdk_bs_batch_close(batch); 887 } 888 889 static int 890 _spdk_resize_blob(struct spdk_blob *blob, uint64_t sz) 891 { 892 uint64_t i; 893 uint64_t *tmp; 894 uint64_t lfc; /* lowest free cluster */ 895 struct spdk_blob_store *bs; 896 897 bs = blob->bs; 898 899 assert(blob->state != SPDK_BLOB_STATE_LOADING && 900 blob->state != SPDK_BLOB_STATE_SYNCING); 901 902 if (blob->active.num_clusters == sz) { 903 return 0; 904 } 905 906 if (blob->active.num_clusters < blob->active.cluster_array_size) { 907 /* If this blob was resized to be larger, then smaller, then 908 * larger without syncing, then the cluster array already 909 * contains spare assigned clusters we can use. 910 */ 911 blob->active.num_clusters = spdk_min(blob->active.cluster_array_size, 912 sz); 913 } 914 915 blob->state = SPDK_BLOB_STATE_DIRTY; 916 917 /* Do two passes - one to verify that we can obtain enough clusters 918 * and another to actually claim them. 919 */ 920 921 lfc = 0; 922 for (i = blob->active.num_clusters; i < sz; i++) { 923 lfc = spdk_bit_array_find_first_clear(bs->used_clusters, lfc); 924 if (lfc >= bs->total_clusters) { 925 /* No more free clusters. Cannot satisfy the request */ 926 assert(false); 927 return -1; 928 } 929 lfc++; 930 } 931 932 if (sz > blob->active.num_clusters) { 933 /* Expand the cluster array if necessary. 934 * We only shrink the array when persisting. 935 */ 936 tmp = realloc(blob->active.clusters, sizeof(uint64_t) * sz); 937 if (sz > 0 && tmp == NULL) { 938 assert(false); 939 return -1; 940 } 941 blob->active.clusters = tmp; 942 blob->active.cluster_array_size = sz; 943 } 944 945 lfc = 0; 946 for (i = blob->active.num_clusters; i < sz; i++) { 947 lfc = spdk_bit_array_find_first_clear(bs->used_clusters, lfc); 948 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Claiming cluster %lu for blob %lu\n", lfc, blob->id); 949 _spdk_bs_claim_cluster(bs, lfc); 950 blob->active.clusters[i] = _spdk_bs_cluster_to_lba(bs, lfc); 951 lfc++; 952 } 953 954 blob->active.num_clusters = sz; 955 956 return 0; 957 } 958 959 /* Write a blob to disk */ 960 static void 961 _spdk_blob_persist(spdk_bs_sequence_t *seq, struct spdk_blob *blob, 962 spdk_bs_sequence_cpl cb_fn, void *cb_arg) 963 { 964 struct spdk_blob_persist_ctx *ctx; 965 int rc; 966 uint64_t i; 967 uint32_t page_num; 968 struct spdk_blob_store *bs; 969 970 assert(blob != NULL); 971 assert(blob->state == SPDK_BLOB_STATE_CLEAN || 972 blob->state == SPDK_BLOB_STATE_DIRTY); 973 974 if (blob->state == SPDK_BLOB_STATE_CLEAN) { 975 cb_fn(seq, cb_arg, 0); 976 return; 977 } 978 979 bs = blob->bs; 980 981 ctx = calloc(1, sizeof(*ctx)); 982 if (!ctx) { 983 cb_fn(seq, cb_arg, -ENOMEM); 984 return; 985 } 986 ctx->blob = blob; 987 ctx->cb_fn = cb_fn; 988 ctx->cb_arg = cb_arg; 989 990 blob->state = SPDK_BLOB_STATE_SYNCING; 991 992 if (blob->active.num_pages == 0) { 993 /* This is the signal that the blob should be deleted. 994 * Immediately jump to the clean up routine. */ 995 assert(blob->clean.num_pages > 0); 996 ctx->idx = blob->clean.num_pages - 1; 997 _spdk_blob_persist_unmap_pages(seq, ctx, 0); 998 return; 999 1000 } 1001 1002 /* Generate the new metadata */ 1003 rc = _spdk_blob_serialize(blob, &ctx->pages, &blob->active.num_pages); 1004 if (rc < 0) { 1005 free(ctx); 1006 cb_fn(seq, cb_arg, rc); 1007 return; 1008 } 1009 1010 assert(blob->active.num_pages >= 1); 1011 1012 /* Resize the cache of page indices */ 1013 blob->active.pages = realloc(blob->active.pages, 1014 blob->active.num_pages * sizeof(*blob->active.pages)); 1015 if (!blob->active.pages) { 1016 free(ctx); 1017 cb_fn(seq, cb_arg, -ENOMEM); 1018 return; 1019 } 1020 1021 /* Assign this metadata to pages. This requires two passes - 1022 * one to verify that there are enough pages and a second 1023 * to actually claim them. */ 1024 page_num = 0; 1025 /* Note that this loop starts at one. The first page location is fixed by the blobid. */ 1026 for (i = 1; i < blob->active.num_pages; i++) { 1027 page_num = spdk_bit_array_find_first_clear(bs->used_md_pages, page_num); 1028 if (page_num >= spdk_bit_array_capacity(bs->used_md_pages)) { 1029 spdk_dma_free(ctx->pages); 1030 free(ctx); 1031 blob->state = SPDK_BLOB_STATE_DIRTY; 1032 cb_fn(seq, cb_arg, -ENOMEM); 1033 return; 1034 } 1035 page_num++; 1036 } 1037 1038 page_num = 0; 1039 blob->active.pages[0] = _spdk_bs_blobid_to_page(blob->id); 1040 for (i = 1; i < blob->active.num_pages; i++) { 1041 page_num = spdk_bit_array_find_first_clear(bs->used_md_pages, page_num); 1042 ctx->pages[i - 1].next = page_num; 1043 /* Now that previous metadata page is complete, calculate the crc for it. */ 1044 ctx->pages[i - 1].crc = _spdk_blob_md_page_calc_crc(&ctx->pages[i - 1]); 1045 blob->active.pages[i] = page_num; 1046 spdk_bit_array_set(bs->used_md_pages, page_num); 1047 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Claiming page %u for blob %lu\n", page_num, blob->id); 1048 page_num++; 1049 } 1050 ctx->pages[i - 1].crc = _spdk_blob_md_page_calc_crc(&ctx->pages[i - 1]); 1051 /* Start writing the metadata from last page to first */ 1052 ctx->idx = blob->active.num_pages - 1; 1053 _spdk_blob_persist_write_page_chain(seq, ctx, 0); 1054 } 1055 1056 static void 1057 _spdk_blob_request_submit_rw(struct spdk_blob *blob, struct spdk_io_channel *_channel, 1058 void *payload, uint64_t offset, uint64_t length, 1059 spdk_blob_op_complete cb_fn, void *cb_arg, bool read) 1060 { 1061 spdk_bs_batch_t *batch; 1062 struct spdk_bs_cpl cpl; 1063 uint64_t lba; 1064 uint32_t lba_count; 1065 uint8_t *buf; 1066 uint64_t page; 1067 1068 assert(blob != NULL); 1069 1070 if (offset + length > blob->active.num_clusters * blob->bs->pages_per_cluster) { 1071 cb_fn(cb_arg, -EINVAL); 1072 return; 1073 } 1074 1075 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1076 cpl.u.blob_basic.cb_fn = cb_fn; 1077 cpl.u.blob_basic.cb_arg = cb_arg; 1078 1079 batch = spdk_bs_batch_open(_channel, &cpl); 1080 if (!batch) { 1081 cb_fn(cb_arg, -ENOMEM); 1082 return; 1083 } 1084 1085 length = _spdk_bs_page_to_lba(blob->bs, length); 1086 page = offset; 1087 buf = payload; 1088 while (length > 0) { 1089 lba = _spdk_bs_blob_page_to_lba(blob, page); 1090 lba_count = spdk_min(length, 1091 _spdk_bs_page_to_lba(blob->bs, 1092 _spdk_bs_num_pages_to_cluster_boundary(blob, page))); 1093 1094 if (read) { 1095 spdk_bs_batch_read(batch, buf, lba, lba_count); 1096 } else { 1097 spdk_bs_batch_write(batch, buf, lba, lba_count); 1098 } 1099 1100 length -= lba_count; 1101 buf += _spdk_bs_lba_to_byte(blob->bs, lba_count); 1102 page += _spdk_bs_lba_to_page(blob->bs, lba_count); 1103 } 1104 1105 spdk_bs_batch_close(batch); 1106 } 1107 1108 struct rw_iov_ctx { 1109 struct spdk_blob *blob; 1110 bool read; 1111 int iovcnt; 1112 struct iovec *orig_iov; 1113 uint64_t page_offset; 1114 uint64_t pages_remaining; 1115 uint64_t pages_done; 1116 struct iovec iov[0]; 1117 }; 1118 1119 static void 1120 _spdk_rw_iov_done(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1121 { 1122 assert(cb_arg == NULL); 1123 spdk_bs_sequence_finish(seq, bserrno); 1124 } 1125 1126 static void 1127 _spdk_rw_iov_split_next(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1128 { 1129 struct rw_iov_ctx *ctx = cb_arg; 1130 struct iovec *iov, *orig_iov; 1131 int iovcnt; 1132 size_t orig_iovoff; 1133 uint64_t lba; 1134 uint64_t page_count, pages_to_boundary; 1135 uint32_t lba_count; 1136 uint64_t byte_count; 1137 1138 if (bserrno != 0 || ctx->pages_remaining == 0) { 1139 free(ctx); 1140 spdk_bs_sequence_finish(seq, bserrno); 1141 return; 1142 } 1143 1144 pages_to_boundary = _spdk_bs_num_pages_to_cluster_boundary(ctx->blob, ctx->page_offset); 1145 page_count = spdk_min(ctx->pages_remaining, pages_to_boundary); 1146 lba = _spdk_bs_blob_page_to_lba(ctx->blob, ctx->page_offset); 1147 lba_count = _spdk_bs_page_to_lba(ctx->blob->bs, page_count); 1148 1149 /* 1150 * Get index and offset into the original iov array for our current position in the I/O sequence. 1151 * byte_count will keep track of how many bytes remaining until orig_iov and orig_iovoff will 1152 * point to the current position in the I/O sequence. 1153 */ 1154 byte_count = ctx->pages_done * sizeof(struct spdk_blob_md_page); 1155 orig_iov = &ctx->orig_iov[0]; 1156 orig_iovoff = 0; 1157 while (byte_count > 0) { 1158 if (byte_count >= orig_iov->iov_len) { 1159 byte_count -= orig_iov->iov_len; 1160 orig_iov++; 1161 } else { 1162 orig_iovoff = byte_count; 1163 byte_count = 0; 1164 } 1165 } 1166 1167 /* 1168 * Build an iov array for the next I/O in the sequence. byte_count will keep track of how many 1169 * bytes of this next I/O remain to be accounted for in the new iov array. 1170 */ 1171 byte_count = page_count * sizeof(struct spdk_blob_md_page); 1172 iov = &ctx->iov[0]; 1173 iovcnt = 0; 1174 while (byte_count > 0) { 1175 iov->iov_len = spdk_min(byte_count, orig_iov->iov_len - orig_iovoff); 1176 iov->iov_base = orig_iov->iov_base + orig_iovoff; 1177 byte_count -= iov->iov_len; 1178 orig_iovoff = 0; 1179 orig_iov++; 1180 iov++; 1181 iovcnt++; 1182 } 1183 1184 ctx->page_offset += page_count; 1185 ctx->pages_done += page_count; 1186 ctx->pages_remaining -= page_count; 1187 iov = &ctx->iov[0]; 1188 1189 if (ctx->read) { 1190 spdk_bs_sequence_readv(seq, iov, iovcnt, lba, lba_count, _spdk_rw_iov_split_next, ctx); 1191 } else { 1192 spdk_bs_sequence_writev(seq, iov, iovcnt, lba, lba_count, _spdk_rw_iov_split_next, ctx); 1193 } 1194 } 1195 1196 static void 1197 _spdk_blob_request_submit_rw_iov(struct spdk_blob *blob, struct spdk_io_channel *_channel, 1198 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 1199 spdk_blob_op_complete cb_fn, void *cb_arg, bool read) 1200 { 1201 spdk_bs_sequence_t *seq; 1202 struct spdk_bs_cpl cpl; 1203 1204 assert(blob != NULL); 1205 1206 if (length == 0) { 1207 cb_fn(cb_arg, 0); 1208 return; 1209 } 1210 1211 if (offset + length > blob->active.num_clusters * blob->bs->pages_per_cluster) { 1212 cb_fn(cb_arg, -EINVAL); 1213 return; 1214 } 1215 1216 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1217 cpl.u.blob_basic.cb_fn = cb_fn; 1218 cpl.u.blob_basic.cb_arg = cb_arg; 1219 1220 /* 1221 * For now, we implement readv/writev using a sequence (instead of a batch) to account for having 1222 * to split a request that spans a cluster boundary. For I/O that do not span a cluster boundary, 1223 * there will be no noticeable difference compared to using a batch. For I/O that do span a cluster 1224 * boundary, the target LBAs (after blob offset to LBA translation) may not be contiguous, so we need 1225 * to allocate a separate iov array and split the I/O such that none of the resulting 1226 * smaller I/O cross a cluster boundary. These smaller I/O will be issued in sequence (not in parallel) 1227 * but since this case happens very infrequently, any performance impact will be negligible. 1228 * 1229 * This could be optimized in the future to allocate a big enough iov array to account for all of the iovs 1230 * for all of the smaller I/Os, pre-build all of the iov arrays for the smaller I/Os, then issue them 1231 * in a batch. That would also require creating an intermediate spdk_bs_cpl that would get called 1232 * when the batch was completed, to allow for freeing the memory for the iov arrays. 1233 */ 1234 seq = spdk_bs_sequence_start(_channel, &cpl); 1235 if (!seq) { 1236 cb_fn(cb_arg, -ENOMEM); 1237 return; 1238 } 1239 1240 if (spdk_likely(length <= _spdk_bs_num_pages_to_cluster_boundary(blob, offset))) { 1241 uint64_t lba = _spdk_bs_blob_page_to_lba(blob, offset); 1242 uint32_t lba_count = _spdk_bs_page_to_lba(blob->bs, length); 1243 1244 if (read) { 1245 spdk_bs_sequence_readv(seq, iov, iovcnt, lba, lba_count, _spdk_rw_iov_done, NULL); 1246 } else { 1247 spdk_bs_sequence_writev(seq, iov, iovcnt, lba, lba_count, _spdk_rw_iov_done, NULL); 1248 } 1249 } else { 1250 struct rw_iov_ctx *ctx; 1251 1252 ctx = calloc(1, sizeof(struct rw_iov_ctx) + iovcnt * sizeof(struct iovec)); 1253 if (ctx == NULL) { 1254 spdk_bs_sequence_finish(seq, -ENOMEM); 1255 return; 1256 } 1257 1258 ctx->blob = blob; 1259 ctx->read = read; 1260 ctx->orig_iov = iov; 1261 ctx->iovcnt = iovcnt; 1262 ctx->page_offset = offset; 1263 ctx->pages_remaining = length; 1264 ctx->pages_done = 0; 1265 1266 _spdk_rw_iov_split_next(seq, ctx, 0); 1267 } 1268 } 1269 1270 static struct spdk_blob * 1271 _spdk_blob_lookup(struct spdk_blob_store *bs, spdk_blob_id blobid) 1272 { 1273 struct spdk_blob *blob; 1274 1275 TAILQ_FOREACH(blob, &bs->blobs, link) { 1276 if (blob->id == blobid) { 1277 return blob; 1278 } 1279 } 1280 1281 return NULL; 1282 } 1283 1284 static int 1285 _spdk_bs_channel_create(struct spdk_blob_store *bs, struct spdk_bs_channel *channel, 1286 uint32_t max_ops) 1287 { 1288 struct spdk_bs_dev *dev; 1289 uint32_t i; 1290 1291 dev = bs->dev; 1292 1293 channel->req_mem = calloc(max_ops, sizeof(struct spdk_bs_request_set)); 1294 if (!channel->req_mem) { 1295 return -1; 1296 } 1297 1298 TAILQ_INIT(&channel->reqs); 1299 1300 for (i = 0; i < max_ops; i++) { 1301 TAILQ_INSERT_TAIL(&channel->reqs, &channel->req_mem[i], link); 1302 } 1303 1304 channel->bs = bs; 1305 channel->dev = dev; 1306 channel->dev_channel = dev->create_channel(dev); 1307 1308 return 0; 1309 } 1310 1311 static int 1312 _spdk_bs_md_channel_create(void *io_device, void *ctx_buf) 1313 { 1314 struct spdk_blob_store *bs; 1315 struct spdk_bs_channel *channel = ctx_buf; 1316 1317 bs = SPDK_CONTAINEROF(io_device, struct spdk_blob_store, md_target); 1318 1319 return _spdk_bs_channel_create(bs, channel, bs->md_target.max_md_ops); 1320 } 1321 1322 static int 1323 _spdk_bs_io_channel_create(void *io_device, void *ctx_buf) 1324 { 1325 struct spdk_blob_store *bs; 1326 struct spdk_bs_channel *channel = ctx_buf; 1327 1328 bs = SPDK_CONTAINEROF(io_device, struct spdk_blob_store, io_target); 1329 1330 return _spdk_bs_channel_create(bs, channel, bs->io_target.max_channel_ops); 1331 } 1332 1333 1334 static void 1335 _spdk_bs_channel_destroy(void *io_device, void *ctx_buf) 1336 { 1337 struct spdk_bs_channel *channel = ctx_buf; 1338 1339 free(channel->req_mem); 1340 channel->dev->destroy_channel(channel->dev, channel->dev_channel); 1341 } 1342 1343 static void 1344 _spdk_bs_dev_destroy(void *io_device) 1345 { 1346 struct spdk_blob_store *bs; 1347 struct spdk_blob *blob, *blob_tmp; 1348 1349 bs = SPDK_CONTAINEROF(io_device, struct spdk_blob_store, md_target); 1350 bs->dev->destroy(bs->dev); 1351 1352 TAILQ_FOREACH_SAFE(blob, &bs->blobs, link, blob_tmp) { 1353 TAILQ_REMOVE(&bs->blobs, blob, link); 1354 _spdk_blob_free(blob); 1355 } 1356 1357 spdk_bit_array_free(&bs->used_md_pages); 1358 spdk_bit_array_free(&bs->used_clusters); 1359 /* 1360 * If this function is called for any reason except a successful unload, 1361 * the unload_cpl type will be NONE and this will be a nop. 1362 */ 1363 spdk_bs_call_cpl(&bs->unload_cpl, bs->unload_err); 1364 1365 free(bs); 1366 } 1367 1368 static void 1369 _spdk_bs_free(struct spdk_blob_store *bs) 1370 { 1371 spdk_bs_unregister_md_thread(bs); 1372 spdk_io_device_unregister(&bs->io_target, NULL); 1373 spdk_io_device_unregister(&bs->md_target, _spdk_bs_dev_destroy); 1374 } 1375 1376 void 1377 spdk_bs_opts_init(struct spdk_bs_opts *opts) 1378 { 1379 opts->cluster_sz = SPDK_BLOB_OPTS_CLUSTER_SZ; 1380 opts->num_md_pages = SPDK_BLOB_OPTS_NUM_MD_PAGES; 1381 opts->max_md_ops = SPDK_BLOB_OPTS_MAX_MD_OPS; 1382 opts->max_channel_ops = SPDK_BLOB_OPTS_MAX_CHANNEL_OPS; 1383 memset(&opts->bstype, 0, sizeof(opts->bstype)); 1384 } 1385 1386 static int 1387 _spdk_bs_opts_verify(struct spdk_bs_opts *opts) 1388 { 1389 if (opts->cluster_sz == 0 || opts->num_md_pages == 0 || opts->max_md_ops == 0 || 1390 opts->max_channel_ops == 0) { 1391 SPDK_ERRLOG("Blobstore options cannot be set to 0\n"); 1392 return -1; 1393 } 1394 1395 return 0; 1396 } 1397 1398 static struct spdk_blob_store * 1399 _spdk_bs_alloc(struct spdk_bs_dev *dev, struct spdk_bs_opts *opts) 1400 { 1401 struct spdk_blob_store *bs; 1402 uint64_t dev_size; 1403 1404 dev_size = dev->blocklen * dev->blockcnt; 1405 if (dev_size < opts->cluster_sz) { 1406 /* Device size cannot be smaller than cluster size of blobstore */ 1407 SPDK_ERRLOG("Device size %" PRIu64 " is smaller than cluster size %d\n", dev_size, 1408 opts->cluster_sz); 1409 return NULL; 1410 } 1411 bs = calloc(1, sizeof(struct spdk_blob_store)); 1412 if (!bs) { 1413 return NULL; 1414 } 1415 1416 TAILQ_INIT(&bs->blobs); 1417 bs->dev = dev; 1418 1419 /* 1420 * Do not use _spdk_bs_lba_to_cluster() here since blockcnt may not be an 1421 * even multiple of the cluster size. 1422 */ 1423 bs->cluster_sz = opts->cluster_sz; 1424 bs->total_clusters = dev->blockcnt / (bs->cluster_sz / dev->blocklen); 1425 bs->pages_per_cluster = bs->cluster_sz / SPDK_BS_PAGE_SIZE; 1426 bs->num_free_clusters = bs->total_clusters; 1427 bs->used_clusters = spdk_bit_array_create(bs->total_clusters); 1428 if (bs->used_clusters == NULL) { 1429 free(bs); 1430 return NULL; 1431 } 1432 1433 bs->md_target.max_md_ops = opts->max_md_ops; 1434 bs->io_target.max_channel_ops = opts->max_channel_ops; 1435 bs->super_blob = SPDK_BLOBID_INVALID; 1436 memcpy(&bs->bstype, &opts->bstype, sizeof(opts->bstype)); 1437 1438 /* The metadata is assumed to be at least 1 page */ 1439 bs->used_md_pages = spdk_bit_array_create(1); 1440 1441 spdk_io_device_register(&bs->md_target, _spdk_bs_md_channel_create, _spdk_bs_channel_destroy, 1442 sizeof(struct spdk_bs_channel)); 1443 spdk_bs_register_md_thread(bs); 1444 1445 spdk_io_device_register(&bs->io_target, _spdk_bs_io_channel_create, _spdk_bs_channel_destroy, 1446 sizeof(struct spdk_bs_channel)); 1447 1448 return bs; 1449 } 1450 1451 /* START spdk_bs_load, spdk_bs_load_ctx will used for both load and unload. */ 1452 1453 struct spdk_bs_load_ctx { 1454 struct spdk_blob_store *bs; 1455 struct spdk_bs_super_block *super; 1456 1457 struct spdk_bs_md_mask *mask; 1458 bool in_page_chain; 1459 uint32_t page_index; 1460 uint32_t cur_page; 1461 struct spdk_blob_md_page *page; 1462 }; 1463 1464 static void 1465 _spdk_bs_set_mask(struct spdk_bit_array *array, struct spdk_bs_md_mask *mask) 1466 { 1467 uint32_t i = 0; 1468 1469 while (true) { 1470 i = spdk_bit_array_find_first_set(array, i); 1471 if (i >= mask->length) { 1472 break; 1473 } 1474 mask->mask[i / 8] |= 1U << (i % 8); 1475 i++; 1476 } 1477 } 1478 1479 static void 1480 _spdk_bs_write_super(spdk_bs_sequence_t *seq, struct spdk_blob_store *bs, 1481 struct spdk_bs_super_block *super, spdk_bs_sequence_cpl cb_fn, void *cb_arg) 1482 { 1483 /* Update the values in the super block */ 1484 super->super_blob = bs->super_blob; 1485 memcpy(&super->bstype, &bs->bstype, sizeof(bs->bstype)); 1486 super->crc = _spdk_blob_md_page_calc_crc(super); 1487 spdk_bs_sequence_write(seq, super, _spdk_bs_page_to_lba(bs, 0), 1488 _spdk_bs_byte_to_lba(bs, sizeof(*super)), 1489 cb_fn, cb_arg); 1490 } 1491 1492 static void 1493 _spdk_bs_write_used_clusters(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 1494 { 1495 struct spdk_bs_load_ctx *ctx = arg; 1496 uint64_t mask_size, lba, lba_count; 1497 1498 /* Write out the used clusters mask */ 1499 mask_size = ctx->super->used_cluster_mask_len * SPDK_BS_PAGE_SIZE; 1500 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 1501 if (!ctx->mask) { 1502 spdk_dma_free(ctx->super); 1503 free(ctx); 1504 spdk_bs_sequence_finish(seq, -ENOMEM); 1505 return; 1506 } 1507 1508 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_CLUSTERS; 1509 ctx->mask->length = ctx->bs->total_clusters; 1510 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_clusters)); 1511 1512 _spdk_bs_set_mask(ctx->bs->used_clusters, ctx->mask); 1513 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_start); 1514 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_len); 1515 spdk_bs_sequence_write(seq, ctx->mask, lba, lba_count, cb_fn, arg); 1516 } 1517 1518 static void 1519 _spdk_bs_write_used_md(spdk_bs_sequence_t *seq, void *arg, spdk_bs_sequence_cpl cb_fn) 1520 { 1521 struct spdk_bs_load_ctx *ctx = arg; 1522 uint64_t mask_size, lba, lba_count; 1523 1524 mask_size = ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE; 1525 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 1526 if (!ctx->mask) { 1527 spdk_dma_free(ctx->super); 1528 free(ctx); 1529 spdk_bs_sequence_finish(seq, -ENOMEM); 1530 return; 1531 } 1532 1533 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_PAGES; 1534 ctx->mask->length = ctx->super->md_len; 1535 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_md_pages)); 1536 1537 _spdk_bs_set_mask(ctx->bs->used_md_pages, ctx->mask); 1538 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_start); 1539 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_len); 1540 spdk_bs_sequence_write(seq, ctx->mask, lba, lba_count, cb_fn, arg); 1541 } 1542 1543 static void 1544 _spdk_bs_load_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1545 { 1546 struct spdk_bs_load_ctx *ctx = cb_arg; 1547 uint32_t i, j; 1548 int rc; 1549 1550 /* The type must be correct */ 1551 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_CLUSTERS); 1552 /* The length of the mask (in bits) must not be greater than the length of the buffer (converted to bits) */ 1553 assert(ctx->mask->length <= (ctx->super->used_cluster_mask_len * sizeof( 1554 struct spdk_blob_md_page) * 8)); 1555 /* The length of the mask must be exactly equal to the total number of clusters */ 1556 assert(ctx->mask->length == ctx->bs->total_clusters); 1557 1558 rc = spdk_bit_array_resize(&ctx->bs->used_clusters, ctx->bs->total_clusters); 1559 if (rc < 0) { 1560 spdk_dma_free(ctx->super); 1561 spdk_dma_free(ctx->mask); 1562 _spdk_bs_free(ctx->bs); 1563 free(ctx); 1564 spdk_bs_sequence_finish(seq, -ENOMEM); 1565 return; 1566 } 1567 1568 ctx->bs->num_free_clusters = ctx->bs->total_clusters; 1569 for (i = 0; i < ctx->mask->length / 8; i++) { 1570 uint8_t segment = ctx->mask->mask[i]; 1571 for (j = 0; segment && (j < 8); j++) { 1572 if (segment & 1U) { 1573 spdk_bit_array_set(ctx->bs->used_clusters, (i * 8) + j); 1574 assert(ctx->bs->num_free_clusters > 0); 1575 ctx->bs->num_free_clusters--; 1576 } 1577 segment >>= 1U; 1578 } 1579 } 1580 1581 spdk_dma_free(ctx->super); 1582 spdk_dma_free(ctx->mask); 1583 free(ctx); 1584 1585 spdk_bs_sequence_finish(seq, bserrno); 1586 } 1587 1588 static void 1589 _spdk_bs_load_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1590 { 1591 struct spdk_bs_load_ctx *ctx = cb_arg; 1592 uint64_t lba, lba_count, mask_size; 1593 uint32_t i, j; 1594 int rc; 1595 1596 /* The type must be correct */ 1597 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_PAGES); 1598 /* The length of the mask (in bits) must not be greater than the length of the buffer (converted to bits) */ 1599 assert(ctx->mask->length <= (ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE * 1600 8)); 1601 /* The length of the mask must be exactly equal to the size (in pages) of the metadata region */ 1602 assert(ctx->mask->length == ctx->super->md_len); 1603 1604 rc = spdk_bit_array_resize(&ctx->bs->used_md_pages, ctx->mask->length); 1605 if (rc < 0) { 1606 spdk_dma_free(ctx->super); 1607 spdk_dma_free(ctx->mask); 1608 _spdk_bs_free(ctx->bs); 1609 free(ctx); 1610 spdk_bs_sequence_finish(seq, -ENOMEM); 1611 return; 1612 } 1613 1614 for (i = 0; i < ctx->mask->length / 8; i++) { 1615 uint8_t segment = ctx->mask->mask[i]; 1616 for (j = 0; segment && (j < 8); j++) { 1617 if (segment & 1U) { 1618 spdk_bit_array_set(ctx->bs->used_md_pages, (i * 8) + j); 1619 } 1620 segment >>= 1U; 1621 } 1622 } 1623 spdk_dma_free(ctx->mask); 1624 1625 /* Read the used clusters mask */ 1626 mask_size = ctx->super->used_cluster_mask_len * SPDK_BS_PAGE_SIZE; 1627 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 1628 if (!ctx->mask) { 1629 spdk_dma_free(ctx->super); 1630 _spdk_bs_free(ctx->bs); 1631 free(ctx); 1632 spdk_bs_sequence_finish(seq, -ENOMEM); 1633 return; 1634 } 1635 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_start); 1636 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_len); 1637 spdk_bs_sequence_read(seq, ctx->mask, lba, lba_count, 1638 _spdk_bs_load_used_clusters_cpl, ctx); 1639 } 1640 1641 static void 1642 _spdk_bs_load_write_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1643 { 1644 struct spdk_bs_load_ctx *ctx = cb_arg; 1645 uint64_t lba, lba_count, mask_size; 1646 1647 /* Read the used pages mask */ 1648 mask_size = ctx->super->used_page_mask_len * SPDK_BS_PAGE_SIZE; 1649 ctx->mask = spdk_dma_zmalloc(mask_size, 0x1000, NULL); 1650 if (!ctx->mask) { 1651 spdk_dma_free(ctx->super); 1652 _spdk_bs_free(ctx->bs); 1653 free(ctx); 1654 spdk_bs_sequence_finish(seq, -ENOMEM); 1655 return; 1656 } 1657 1658 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_start); 1659 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_len); 1660 spdk_bs_sequence_read(seq, ctx->mask, lba, lba_count, 1661 _spdk_bs_load_used_pages_cpl, ctx); 1662 } 1663 1664 static int 1665 _spdk_bs_load_replay_md_parse_page(const struct spdk_blob_md_page *page, struct spdk_blob_store *bs) 1666 { 1667 struct spdk_blob_md_descriptor *desc; 1668 size_t cur_desc = 0; 1669 1670 desc = (struct spdk_blob_md_descriptor *)page->descriptors; 1671 while (cur_desc < sizeof(page->descriptors)) { 1672 if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_PADDING) { 1673 if (desc->length == 0) { 1674 /* If padding and length are 0, this terminates the page */ 1675 break; 1676 } 1677 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_EXTENT) { 1678 struct spdk_blob_md_descriptor_extent *desc_extent; 1679 unsigned int i, j; 1680 unsigned int cluster_count = 0; 1681 1682 desc_extent = (struct spdk_blob_md_descriptor_extent *)desc; 1683 1684 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 1685 for (j = 0; j < desc_extent->extents[i].length; j++) { 1686 spdk_bit_array_set(bs->used_clusters, desc_extent->extents[i].cluster_idx + j); 1687 if (bs->num_free_clusters == 0) { 1688 return -1; 1689 } 1690 bs->num_free_clusters--; 1691 cluster_count++; 1692 } 1693 } 1694 if (cluster_count == 0) { 1695 return -1; 1696 } 1697 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR) { 1698 /* Skip this item */ 1699 } else { 1700 /* Error */ 1701 return -1; 1702 } 1703 /* Advance to the next descriptor */ 1704 cur_desc += sizeof(*desc) + desc->length; 1705 if (cur_desc + sizeof(*desc) > sizeof(page->descriptors)) { 1706 break; 1707 } 1708 desc = (struct spdk_blob_md_descriptor *)((uintptr_t)page->descriptors + cur_desc); 1709 } 1710 return 0; 1711 } 1712 1713 static bool _spdk_bs_load_cur_md_page_valid(struct spdk_bs_load_ctx *ctx) 1714 { 1715 uint32_t crc; 1716 1717 crc = _spdk_blob_md_page_calc_crc(ctx->page); 1718 if (crc != ctx->page->crc) { 1719 return false; 1720 } 1721 1722 if (_spdk_bs_page_to_blobid(ctx->cur_page) != ctx->page->id) { 1723 return false; 1724 } 1725 return true; 1726 } 1727 1728 static void 1729 _spdk_bs_load_replay_cur_md_page(spdk_bs_sequence_t *seq, void *cb_arg); 1730 1731 static void 1732 _spdk_bs_load_write_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1733 { 1734 struct spdk_bs_load_ctx *ctx = cb_arg; 1735 1736 spdk_dma_free(ctx->mask); 1737 spdk_dma_free(ctx->super); 1738 spdk_bs_sequence_finish(seq, bserrno); 1739 free(ctx); 1740 } 1741 1742 static void 1743 _spdk_bs_load_write_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1744 { 1745 struct spdk_bs_load_ctx *ctx = cb_arg; 1746 1747 spdk_dma_free(ctx->mask); 1748 1749 _spdk_bs_write_used_clusters(seq, cb_arg, _spdk_bs_load_write_used_clusters_cpl); 1750 } 1751 1752 static void 1753 _spdk_bs_load_write_used_md(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1754 { 1755 _spdk_bs_write_used_md(seq, cb_arg, _spdk_bs_load_write_used_pages_cpl); 1756 } 1757 1758 static void 1759 _spdk_bs_load_replay_md_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1760 { 1761 struct spdk_bs_load_ctx *ctx = cb_arg; 1762 uint32_t page_num; 1763 1764 if (bserrno != 0) { 1765 spdk_dma_free(ctx->super); 1766 _spdk_bs_free(ctx->bs); 1767 free(ctx); 1768 spdk_bs_sequence_finish(seq, bserrno); 1769 return; 1770 } 1771 1772 page_num = ctx->cur_page; 1773 if (_spdk_bs_load_cur_md_page_valid(ctx) == true) { 1774 if (ctx->page->sequence_num == 0 || ctx->in_page_chain == true) { 1775 spdk_bit_array_set(ctx->bs->used_md_pages, page_num); 1776 if (_spdk_bs_load_replay_md_parse_page(ctx->page, ctx->bs)) { 1777 spdk_dma_free(ctx->super); 1778 _spdk_bs_free(ctx->bs); 1779 free(ctx); 1780 spdk_bs_sequence_finish(seq, -EILSEQ); 1781 return; 1782 } 1783 if (ctx->page->next != SPDK_INVALID_MD_PAGE) { 1784 ctx->in_page_chain = true; 1785 ctx->cur_page = ctx->page->next; 1786 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 1787 return; 1788 } 1789 } 1790 } 1791 1792 ctx->in_page_chain = false; 1793 1794 do { 1795 ctx->page_index++; 1796 } while (spdk_bit_array_get(ctx->bs->used_md_pages, ctx->page_index) == true); 1797 1798 if (ctx->page_index < ctx->super->md_len) { 1799 ctx->cur_page = ctx->page_index; 1800 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 1801 } else { 1802 spdk_dma_free(ctx->page); 1803 _spdk_bs_load_write_used_md(seq, ctx, bserrno); 1804 } 1805 } 1806 1807 static void 1808 _spdk_bs_load_replay_cur_md_page(spdk_bs_sequence_t *seq, void *cb_arg) 1809 { 1810 struct spdk_bs_load_ctx *ctx = cb_arg; 1811 uint64_t lba; 1812 1813 assert(ctx->cur_page < ctx->super->md_len); 1814 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->md_start + ctx->cur_page); 1815 spdk_bs_sequence_read(seq, ctx->page, lba, 1816 _spdk_bs_byte_to_lba(ctx->bs, SPDK_BS_PAGE_SIZE), 1817 _spdk_bs_load_replay_md_cpl, ctx); 1818 } 1819 1820 static void 1821 _spdk_bs_load_replay_md(spdk_bs_sequence_t *seq, void *cb_arg) 1822 { 1823 struct spdk_bs_load_ctx *ctx = cb_arg; 1824 1825 ctx->page_index = 0; 1826 ctx->cur_page = 0; 1827 ctx->page = spdk_dma_zmalloc(SPDK_BS_PAGE_SIZE, 1828 SPDK_BS_PAGE_SIZE, 1829 NULL); 1830 if (!ctx->page) { 1831 spdk_dma_free(ctx->super); 1832 _spdk_bs_free(ctx->bs); 1833 free(ctx); 1834 spdk_bs_sequence_finish(seq, -ENOMEM); 1835 return; 1836 } 1837 _spdk_bs_load_replay_cur_md_page(seq, cb_arg); 1838 } 1839 1840 static void 1841 _spdk_bs_recover(spdk_bs_sequence_t *seq, void *cb_arg) 1842 { 1843 struct spdk_bs_load_ctx *ctx = cb_arg; 1844 int rc; 1845 1846 rc = spdk_bit_array_resize(&ctx->bs->used_md_pages, ctx->super->md_len); 1847 if (rc < 0) { 1848 spdk_dma_free(ctx->super); 1849 _spdk_bs_free(ctx->bs); 1850 free(ctx); 1851 spdk_bs_sequence_finish(seq, -ENOMEM); 1852 return; 1853 } 1854 1855 rc = spdk_bit_array_resize(&ctx->bs->used_clusters, ctx->bs->total_clusters); 1856 if (rc < 0) { 1857 spdk_dma_free(ctx->super); 1858 _spdk_bs_free(ctx->bs); 1859 free(ctx); 1860 spdk_bs_sequence_finish(seq, -ENOMEM); 1861 return; 1862 } 1863 1864 ctx->bs->num_free_clusters = ctx->bs->total_clusters; 1865 _spdk_bs_load_replay_md(seq, cb_arg); 1866 } 1867 1868 static void 1869 _spdk_bs_load_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1870 { 1871 struct spdk_bs_load_ctx *ctx = cb_arg; 1872 uint32_t crc; 1873 static const char zeros[SPDK_BLOBSTORE_TYPE_LENGTH]; 1874 1875 if (ctx->super->version > SPDK_BS_VERSION || 1876 ctx->super->version < SPDK_BS_INITIAL_VERSION) { 1877 spdk_dma_free(ctx->super); 1878 _spdk_bs_free(ctx->bs); 1879 free(ctx); 1880 spdk_bs_sequence_finish(seq, -EILSEQ); 1881 return; 1882 } 1883 1884 if (memcmp(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 1885 sizeof(ctx->super->signature)) != 0) { 1886 spdk_dma_free(ctx->super); 1887 _spdk_bs_free(ctx->bs); 1888 free(ctx); 1889 spdk_bs_sequence_finish(seq, -EILSEQ); 1890 return; 1891 } 1892 1893 crc = _spdk_blob_md_page_calc_crc(ctx->super); 1894 if (crc != ctx->super->crc) { 1895 spdk_dma_free(ctx->super); 1896 _spdk_bs_free(ctx->bs); 1897 free(ctx); 1898 spdk_bs_sequence_finish(seq, -EILSEQ); 1899 return; 1900 } 1901 1902 if (memcmp(&ctx->bs->bstype, &ctx->super->bstype, SPDK_BLOBSTORE_TYPE_LENGTH) == 0) { 1903 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Bstype matched - loading blobstore\n"); 1904 } else if (memcmp(&ctx->bs->bstype, zeros, SPDK_BLOBSTORE_TYPE_LENGTH) == 0) { 1905 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Bstype wildcard used - loading blobstore regardless bstype\n"); 1906 } else { 1907 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Unexpected bstype\n"); 1908 SPDK_TRACEDUMP(SPDK_TRACE_BLOB, "Expected:", ctx->bs->bstype.bstype, SPDK_BLOBSTORE_TYPE_LENGTH); 1909 SPDK_TRACEDUMP(SPDK_TRACE_BLOB, "Found:", ctx->super->bstype.bstype, SPDK_BLOBSTORE_TYPE_LENGTH); 1910 spdk_dma_free(ctx->super); 1911 _spdk_bs_free(ctx->bs); 1912 free(ctx); 1913 spdk_bs_sequence_finish(seq, -ENXIO); 1914 return; 1915 } 1916 1917 /* Parse the super block */ 1918 ctx->bs->cluster_sz = ctx->super->cluster_size; 1919 ctx->bs->total_clusters = ctx->bs->dev->blockcnt / (ctx->bs->cluster_sz / ctx->bs->dev->blocklen); 1920 ctx->bs->pages_per_cluster = ctx->bs->cluster_sz / SPDK_BS_PAGE_SIZE; 1921 ctx->bs->md_start = ctx->super->md_start; 1922 ctx->bs->md_len = ctx->super->md_len; 1923 ctx->bs->super_blob = ctx->super->super_blob; 1924 memcpy(&ctx->bs->bstype, &ctx->super->bstype, sizeof(ctx->super->bstype)); 1925 1926 if (ctx->super->clean == 1) { 1927 ctx->super->clean = 0; 1928 _spdk_bs_write_super(seq, ctx->bs, ctx->super, _spdk_bs_load_write_super_cpl, ctx); 1929 } else { 1930 _spdk_bs_recover(seq, ctx); 1931 } 1932 } 1933 1934 void 1935 spdk_bs_load(struct spdk_bs_dev *dev, struct spdk_bs_opts *o, 1936 spdk_bs_op_with_handle_complete cb_fn, void *cb_arg) 1937 { 1938 struct spdk_blob_store *bs; 1939 struct spdk_bs_cpl cpl; 1940 spdk_bs_sequence_t *seq; 1941 struct spdk_bs_load_ctx *ctx; 1942 struct spdk_bs_opts opts = {}; 1943 1944 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Loading blobstore from dev %p\n", dev); 1945 1946 if (o) { 1947 opts = *o; 1948 } else { 1949 spdk_bs_opts_init(&opts); 1950 } 1951 1952 if (opts.max_md_ops == 0 || opts.max_channel_ops == 0) { 1953 cb_fn(cb_arg, NULL, -EINVAL); 1954 return; 1955 } 1956 1957 bs = _spdk_bs_alloc(dev, &opts); 1958 if (!bs) { 1959 cb_fn(cb_arg, NULL, -ENOMEM); 1960 return; 1961 } 1962 1963 ctx = calloc(1, sizeof(*ctx)); 1964 if (!ctx) { 1965 _spdk_bs_free(bs); 1966 cb_fn(cb_arg, NULL, -ENOMEM); 1967 return; 1968 } 1969 1970 ctx->bs = bs; 1971 1972 /* Allocate memory for the super block */ 1973 ctx->super = spdk_dma_zmalloc(sizeof(*ctx->super), 0x1000, NULL); 1974 if (!ctx->super) { 1975 free(ctx); 1976 _spdk_bs_free(bs); 1977 return; 1978 } 1979 1980 cpl.type = SPDK_BS_CPL_TYPE_BS_HANDLE; 1981 cpl.u.bs_handle.cb_fn = cb_fn; 1982 cpl.u.bs_handle.cb_arg = cb_arg; 1983 cpl.u.bs_handle.bs = bs; 1984 1985 seq = spdk_bs_sequence_start(bs->md_target.md_channel, &cpl); 1986 if (!seq) { 1987 spdk_dma_free(ctx->super); 1988 free(ctx); 1989 _spdk_bs_free(bs); 1990 cb_fn(cb_arg, NULL, -ENOMEM); 1991 return; 1992 } 1993 1994 /* Read the super block */ 1995 spdk_bs_sequence_read(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 1996 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 1997 _spdk_bs_load_super_cpl, ctx); 1998 } 1999 2000 /* END spdk_bs_load */ 2001 2002 /* START spdk_bs_init */ 2003 2004 struct spdk_bs_init_ctx { 2005 struct spdk_blob_store *bs; 2006 struct spdk_bs_super_block *super; 2007 }; 2008 2009 static void 2010 _spdk_bs_init_persist_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2011 { 2012 struct spdk_bs_init_ctx *ctx = cb_arg; 2013 2014 spdk_dma_free(ctx->super); 2015 free(ctx); 2016 2017 spdk_bs_sequence_finish(seq, bserrno); 2018 } 2019 2020 static void 2021 _spdk_bs_init_trim_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2022 { 2023 struct spdk_bs_init_ctx *ctx = cb_arg; 2024 2025 /* Write super block */ 2026 spdk_bs_sequence_write(seq, ctx->super, _spdk_bs_page_to_lba(ctx->bs, 0), 2027 _spdk_bs_byte_to_lba(ctx->bs, sizeof(*ctx->super)), 2028 _spdk_bs_init_persist_super_cpl, ctx); 2029 } 2030 2031 void 2032 spdk_bs_init(struct spdk_bs_dev *dev, struct spdk_bs_opts *o, 2033 spdk_bs_op_with_handle_complete cb_fn, void *cb_arg) 2034 { 2035 struct spdk_bs_init_ctx *ctx; 2036 struct spdk_blob_store *bs; 2037 struct spdk_bs_cpl cpl; 2038 spdk_bs_sequence_t *seq; 2039 uint64_t num_md_pages; 2040 uint64_t num_md_clusters; 2041 uint32_t i; 2042 struct spdk_bs_opts opts = {}; 2043 int rc; 2044 2045 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Initializing blobstore on dev %p\n", dev); 2046 2047 if ((SPDK_BS_PAGE_SIZE % dev->blocklen) != 0) { 2048 SPDK_ERRLOG("unsupported dev block length of %d\n", 2049 dev->blocklen); 2050 dev->destroy(dev); 2051 cb_fn(cb_arg, NULL, -EINVAL); 2052 return; 2053 } 2054 2055 if (o) { 2056 opts = *o; 2057 } else { 2058 spdk_bs_opts_init(&opts); 2059 } 2060 2061 if (_spdk_bs_opts_verify(&opts) != 0) { 2062 dev->destroy(dev); 2063 cb_fn(cb_arg, NULL, -EINVAL); 2064 return; 2065 } 2066 2067 bs = _spdk_bs_alloc(dev, &opts); 2068 if (!bs) { 2069 dev->destroy(dev); 2070 cb_fn(cb_arg, NULL, -ENOMEM); 2071 return; 2072 } 2073 2074 if (opts.num_md_pages == SPDK_BLOB_OPTS_NUM_MD_PAGES) { 2075 /* By default, allocate 1 page per cluster. 2076 * Technically, this over-allocates metadata 2077 * because more metadata will reduce the number 2078 * of usable clusters. This can be addressed with 2079 * more complex math in the future. 2080 */ 2081 bs->md_len = bs->total_clusters; 2082 } else { 2083 bs->md_len = opts.num_md_pages; 2084 } 2085 2086 rc = spdk_bit_array_resize(&bs->used_md_pages, bs->md_len); 2087 if (rc < 0) { 2088 _spdk_bs_free(bs); 2089 cb_fn(cb_arg, NULL, -ENOMEM); 2090 return; 2091 } 2092 2093 ctx = calloc(1, sizeof(*ctx)); 2094 if (!ctx) { 2095 _spdk_bs_free(bs); 2096 cb_fn(cb_arg, NULL, -ENOMEM); 2097 return; 2098 } 2099 2100 ctx->bs = bs; 2101 2102 /* Allocate memory for the super block */ 2103 ctx->super = spdk_dma_zmalloc(sizeof(*ctx->super), 0x1000, NULL); 2104 if (!ctx->super) { 2105 free(ctx); 2106 _spdk_bs_free(bs); 2107 return; 2108 } 2109 memcpy(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 2110 sizeof(ctx->super->signature)); 2111 ctx->super->version = SPDK_BS_VERSION; 2112 ctx->super->length = sizeof(*ctx->super); 2113 ctx->super->super_blob = bs->super_blob; 2114 ctx->super->clean = 0; 2115 ctx->super->cluster_size = bs->cluster_sz; 2116 memcpy(&ctx->super->bstype, &bs->bstype, sizeof(bs->bstype)); 2117 2118 /* Calculate how many pages the metadata consumes at the front 2119 * of the disk. 2120 */ 2121 2122 /* The super block uses 1 page */ 2123 num_md_pages = 1; 2124 2125 /* The used_md_pages mask requires 1 bit per metadata page, rounded 2126 * up to the nearest page, plus a header. 2127 */ 2128 ctx->super->used_page_mask_start = num_md_pages; 2129 ctx->super->used_page_mask_len = divide_round_up(sizeof(struct spdk_bs_md_mask) + 2130 divide_round_up(bs->md_len, 8), 2131 SPDK_BS_PAGE_SIZE); 2132 num_md_pages += ctx->super->used_page_mask_len; 2133 2134 /* The used_clusters mask requires 1 bit per cluster, rounded 2135 * up to the nearest page, plus a header. 2136 */ 2137 ctx->super->used_cluster_mask_start = num_md_pages; 2138 ctx->super->used_cluster_mask_len = divide_round_up(sizeof(struct spdk_bs_md_mask) + 2139 divide_round_up(bs->total_clusters, 8), 2140 SPDK_BS_PAGE_SIZE); 2141 num_md_pages += ctx->super->used_cluster_mask_len; 2142 2143 /* The metadata region size was chosen above */ 2144 ctx->super->md_start = bs->md_start = num_md_pages; 2145 ctx->super->md_len = bs->md_len; 2146 num_md_pages += bs->md_len; 2147 2148 ctx->super->crc = _spdk_blob_md_page_calc_crc(ctx->super); 2149 2150 num_md_clusters = divide_round_up(num_md_pages, bs->pages_per_cluster); 2151 if (num_md_clusters > bs->total_clusters) { 2152 SPDK_ERRLOG("Blobstore metadata cannot use more clusters than is available, " 2153 "please decrease number of pages reserved for metadata " 2154 "or increase cluster size.\n"); 2155 spdk_dma_free(ctx->super); 2156 free(ctx); 2157 _spdk_bs_free(bs); 2158 cb_fn(cb_arg, NULL, -ENOMEM); 2159 return; 2160 } 2161 /* Claim all of the clusters used by the metadata */ 2162 for (i = 0; i < num_md_clusters; i++) { 2163 _spdk_bs_claim_cluster(bs, i); 2164 } 2165 2166 cpl.type = SPDK_BS_CPL_TYPE_BS_HANDLE; 2167 cpl.u.bs_handle.cb_fn = cb_fn; 2168 cpl.u.bs_handle.cb_arg = cb_arg; 2169 cpl.u.bs_handle.bs = bs; 2170 2171 seq = spdk_bs_sequence_start(bs->md_target.md_channel, &cpl); 2172 if (!seq) { 2173 spdk_dma_free(ctx->super); 2174 free(ctx); 2175 _spdk_bs_free(bs); 2176 cb_fn(cb_arg, NULL, -ENOMEM); 2177 return; 2178 } 2179 2180 /* TRIM the entire device */ 2181 spdk_bs_sequence_unmap(seq, 0, bs->dev->blockcnt, _spdk_bs_init_trim_cpl, ctx); 2182 } 2183 2184 /* END spdk_bs_init */ 2185 2186 /* START spdk_bs_unload */ 2187 2188 static void 2189 _spdk_bs_unload_write_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2190 { 2191 struct spdk_bs_load_ctx *ctx = cb_arg; 2192 2193 spdk_dma_free(ctx->super); 2194 2195 /* 2196 * We need to defer calling spdk_bs_call_cpl() until after 2197 * dev destuction, so tuck these away for later use. 2198 */ 2199 ctx->bs->unload_err = bserrno; 2200 memcpy(&ctx->bs->unload_cpl, &seq->cpl, sizeof(struct spdk_bs_cpl)); 2201 seq->cpl.type = SPDK_BS_CPL_TYPE_NONE; 2202 2203 spdk_bs_sequence_finish(seq, bserrno); 2204 2205 _spdk_bs_free(ctx->bs); 2206 free(ctx); 2207 } 2208 2209 static void 2210 _spdk_bs_unload_write_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2211 { 2212 struct spdk_bs_load_ctx *ctx = cb_arg; 2213 2214 spdk_dma_free(ctx->mask); 2215 ctx->super->clean = 1; 2216 2217 _spdk_bs_write_super(seq, ctx->bs, ctx->super, _spdk_bs_unload_write_super_cpl, ctx); 2218 } 2219 2220 static void 2221 _spdk_bs_unload_write_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2222 { 2223 struct spdk_bs_load_ctx *ctx = cb_arg; 2224 2225 spdk_dma_free(ctx->mask); 2226 2227 _spdk_bs_write_used_clusters(seq, cb_arg, _spdk_bs_unload_write_used_clusters_cpl); 2228 } 2229 2230 static void 2231 _spdk_bs_unload_read_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2232 { 2233 _spdk_bs_write_used_md(seq, cb_arg, _spdk_bs_unload_write_used_pages_cpl); 2234 } 2235 2236 void 2237 spdk_bs_unload(struct spdk_blob_store *bs, spdk_bs_op_complete cb_fn, void *cb_arg) 2238 { 2239 struct spdk_bs_cpl cpl; 2240 spdk_bs_sequence_t *seq; 2241 struct spdk_bs_load_ctx *ctx; 2242 2243 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Syncing blobstore\n"); 2244 2245 ctx = calloc(1, sizeof(*ctx)); 2246 if (!ctx) { 2247 cb_fn(cb_arg, -ENOMEM); 2248 return; 2249 } 2250 2251 ctx->bs = bs; 2252 2253 ctx->super = spdk_dma_zmalloc(sizeof(*ctx->super), 0x1000, NULL); 2254 if (!ctx->super) { 2255 free(ctx); 2256 cb_fn(cb_arg, -ENOMEM); 2257 return; 2258 } 2259 2260 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 2261 cpl.u.bs_basic.cb_fn = cb_fn; 2262 cpl.u.bs_basic.cb_arg = cb_arg; 2263 2264 seq = spdk_bs_sequence_start(bs->md_target.md_channel, &cpl); 2265 if (!seq) { 2266 spdk_dma_free(ctx->super); 2267 free(ctx); 2268 cb_fn(cb_arg, -ENOMEM); 2269 return; 2270 } 2271 2272 assert(TAILQ_EMPTY(&bs->blobs)); 2273 2274 /* Read super block */ 2275 spdk_bs_sequence_read(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 2276 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 2277 _spdk_bs_unload_read_super_cpl, ctx); 2278 } 2279 2280 /* END spdk_bs_unload */ 2281 2282 void 2283 spdk_bs_set_super(struct spdk_blob_store *bs, spdk_blob_id blobid, 2284 spdk_bs_op_complete cb_fn, void *cb_arg) 2285 { 2286 bs->super_blob = blobid; 2287 cb_fn(cb_arg, 0); 2288 } 2289 2290 void 2291 spdk_bs_get_super(struct spdk_blob_store *bs, 2292 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 2293 { 2294 if (bs->super_blob == SPDK_BLOBID_INVALID) { 2295 cb_fn(cb_arg, SPDK_BLOBID_INVALID, -ENOENT); 2296 } else { 2297 cb_fn(cb_arg, bs->super_blob, 0); 2298 } 2299 } 2300 2301 uint64_t 2302 spdk_bs_get_cluster_size(struct spdk_blob_store *bs) 2303 { 2304 return bs->cluster_sz; 2305 } 2306 2307 uint64_t 2308 spdk_bs_get_page_size(struct spdk_blob_store *bs) 2309 { 2310 return SPDK_BS_PAGE_SIZE; 2311 } 2312 2313 uint64_t 2314 spdk_bs_free_cluster_count(struct spdk_blob_store *bs) 2315 { 2316 return bs->num_free_clusters; 2317 } 2318 2319 int spdk_bs_register_md_thread(struct spdk_blob_store *bs) 2320 { 2321 bs->md_target.md_channel = spdk_get_io_channel(&bs->md_target); 2322 2323 return 0; 2324 } 2325 2326 int spdk_bs_unregister_md_thread(struct spdk_blob_store *bs) 2327 { 2328 spdk_put_io_channel(bs->md_target.md_channel); 2329 2330 return 0; 2331 } 2332 2333 spdk_blob_id spdk_blob_get_id(struct spdk_blob *blob) 2334 { 2335 assert(blob != NULL); 2336 2337 return blob->id; 2338 } 2339 2340 uint64_t spdk_blob_get_num_pages(struct spdk_blob *blob) 2341 { 2342 assert(blob != NULL); 2343 2344 return _spdk_bs_cluster_to_page(blob->bs, blob->active.num_clusters); 2345 } 2346 2347 uint64_t spdk_blob_get_num_clusters(struct spdk_blob *blob) 2348 { 2349 assert(blob != NULL); 2350 2351 return blob->active.num_clusters; 2352 } 2353 2354 /* START spdk_bs_md_create_blob */ 2355 2356 static void 2357 _spdk_bs_md_create_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2358 { 2359 struct spdk_blob *blob = cb_arg; 2360 2361 _spdk_blob_free(blob); 2362 2363 spdk_bs_sequence_finish(seq, bserrno); 2364 } 2365 2366 void spdk_bs_md_create_blob(struct spdk_blob_store *bs, 2367 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 2368 { 2369 struct spdk_blob *blob; 2370 uint32_t page_idx; 2371 struct spdk_bs_cpl cpl; 2372 spdk_bs_sequence_t *seq; 2373 spdk_blob_id id; 2374 2375 page_idx = spdk_bit_array_find_first_clear(bs->used_md_pages, 0); 2376 if (page_idx >= spdk_bit_array_capacity(bs->used_md_pages)) { 2377 cb_fn(cb_arg, 0, -ENOMEM); 2378 return; 2379 } 2380 spdk_bit_array_set(bs->used_md_pages, page_idx); 2381 2382 id = _spdk_bs_page_to_blobid(page_idx); 2383 2384 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Creating blob with id %lu at page %u\n", id, page_idx); 2385 2386 blob = _spdk_blob_alloc(bs, id); 2387 if (!blob) { 2388 cb_fn(cb_arg, 0, -ENOMEM); 2389 return; 2390 } 2391 2392 cpl.type = SPDK_BS_CPL_TYPE_BLOBID; 2393 cpl.u.blobid.cb_fn = cb_fn; 2394 cpl.u.blobid.cb_arg = cb_arg; 2395 cpl.u.blobid.blobid = blob->id; 2396 2397 seq = spdk_bs_sequence_start(bs->md_target.md_channel, &cpl); 2398 if (!seq) { 2399 _spdk_blob_free(blob); 2400 cb_fn(cb_arg, 0, -ENOMEM); 2401 return; 2402 } 2403 2404 _spdk_blob_persist(seq, blob, _spdk_bs_md_create_blob_cpl, blob); 2405 } 2406 2407 /* END spdk_bs_md_create_blob */ 2408 2409 /* START spdk_bs_md_resize_blob */ 2410 int 2411 spdk_bs_md_resize_blob(struct spdk_blob *blob, uint64_t sz) 2412 { 2413 int rc; 2414 2415 assert(blob != NULL); 2416 2417 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Resizing blob %lu to %lu clusters\n", blob->id, sz); 2418 2419 if (sz == blob->active.num_clusters) { 2420 return 0; 2421 } 2422 2423 rc = _spdk_resize_blob(blob, sz); 2424 if (rc < 0) { 2425 return rc; 2426 } 2427 2428 return 0; 2429 } 2430 2431 /* END spdk_bs_md_resize_blob */ 2432 2433 2434 /* START spdk_bs_md_delete_blob */ 2435 2436 static void 2437 _spdk_bs_md_delete_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2438 { 2439 struct spdk_blob *blob = cb_arg; 2440 2441 _spdk_blob_free(blob); 2442 2443 spdk_bs_sequence_finish(seq, bserrno); 2444 } 2445 2446 static void 2447 _spdk_bs_md_delete_open_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2448 { 2449 struct spdk_blob *blob = cb_arg; 2450 2451 /* If the blob have crc error, we just return NULL. */ 2452 if (blob == NULL) { 2453 spdk_bs_sequence_finish(seq, bserrno); 2454 return; 2455 } 2456 blob->state = SPDK_BLOB_STATE_DIRTY; 2457 blob->active.num_pages = 0; 2458 _spdk_resize_blob(blob, 0); 2459 2460 _spdk_blob_persist(seq, blob, _spdk_bs_md_delete_blob_cpl, blob); 2461 } 2462 2463 void 2464 spdk_bs_md_delete_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 2465 spdk_blob_op_complete cb_fn, void *cb_arg) 2466 { 2467 struct spdk_blob *blob; 2468 struct spdk_bs_cpl cpl; 2469 spdk_bs_sequence_t *seq; 2470 2471 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Deleting blob %lu\n", blobid); 2472 2473 blob = _spdk_blob_lookup(bs, blobid); 2474 if (blob) { 2475 assert(blob->open_ref > 0); 2476 cb_fn(cb_arg, -EINVAL); 2477 return; 2478 } 2479 2480 blob = _spdk_blob_alloc(bs, blobid); 2481 if (!blob) { 2482 cb_fn(cb_arg, -ENOMEM); 2483 return; 2484 } 2485 2486 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 2487 cpl.u.blob_basic.cb_fn = cb_fn; 2488 cpl.u.blob_basic.cb_arg = cb_arg; 2489 2490 seq = spdk_bs_sequence_start(bs->md_target.md_channel, &cpl); 2491 if (!seq) { 2492 _spdk_blob_free(blob); 2493 cb_fn(cb_arg, -ENOMEM); 2494 return; 2495 } 2496 2497 _spdk_blob_load(seq, blob, _spdk_bs_md_delete_open_cpl, blob); 2498 } 2499 2500 /* END spdk_bs_md_delete_blob */ 2501 2502 /* START spdk_bs_md_open_blob */ 2503 2504 static void 2505 _spdk_bs_md_open_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2506 { 2507 struct spdk_blob *blob = cb_arg; 2508 2509 /* If the blob have crc error, we just return NULL. */ 2510 if (blob == NULL) { 2511 seq->cpl.u.blob_handle.blob = NULL; 2512 spdk_bs_sequence_finish(seq, bserrno); 2513 return; 2514 } 2515 2516 blob->open_ref++; 2517 2518 TAILQ_INSERT_HEAD(&blob->bs->blobs, blob, link); 2519 2520 spdk_bs_sequence_finish(seq, bserrno); 2521 } 2522 2523 void spdk_bs_md_open_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 2524 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 2525 { 2526 struct spdk_blob *blob; 2527 struct spdk_bs_cpl cpl; 2528 spdk_bs_sequence_t *seq; 2529 uint32_t page_num; 2530 2531 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Opening blob %lu\n", blobid); 2532 2533 blob = _spdk_blob_lookup(bs, blobid); 2534 if (blob) { 2535 blob->open_ref++; 2536 cb_fn(cb_arg, blob, 0); 2537 return; 2538 } 2539 2540 page_num = _spdk_bs_blobid_to_page(blobid); 2541 if (spdk_bit_array_get(bs->used_md_pages, page_num) == false) { 2542 /* Invalid blobid */ 2543 cb_fn(cb_arg, NULL, -ENOENT); 2544 return; 2545 } 2546 2547 blob = _spdk_blob_alloc(bs, blobid); 2548 if (!blob) { 2549 cb_fn(cb_arg, NULL, -ENOMEM); 2550 return; 2551 } 2552 2553 cpl.type = SPDK_BS_CPL_TYPE_BLOB_HANDLE; 2554 cpl.u.blob_handle.cb_fn = cb_fn; 2555 cpl.u.blob_handle.cb_arg = cb_arg; 2556 cpl.u.blob_handle.blob = blob; 2557 2558 seq = spdk_bs_sequence_start(bs->md_target.md_channel, &cpl); 2559 if (!seq) { 2560 _spdk_blob_free(blob); 2561 cb_fn(cb_arg, NULL, -ENOMEM); 2562 return; 2563 } 2564 2565 _spdk_blob_load(seq, blob, _spdk_bs_md_open_blob_cpl, blob); 2566 } 2567 2568 /* START spdk_bs_md_sync_blob */ 2569 static void 2570 _spdk_blob_sync_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2571 { 2572 spdk_bs_sequence_finish(seq, bserrno); 2573 } 2574 2575 void spdk_bs_md_sync_blob(struct spdk_blob *blob, 2576 spdk_blob_op_complete cb_fn, void *cb_arg) 2577 { 2578 struct spdk_bs_cpl cpl; 2579 spdk_bs_sequence_t *seq; 2580 2581 assert(blob != NULL); 2582 2583 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Syncing blob %lu\n", blob->id); 2584 2585 assert(blob->state != SPDK_BLOB_STATE_LOADING && 2586 blob->state != SPDK_BLOB_STATE_SYNCING); 2587 2588 if (blob->state == SPDK_BLOB_STATE_CLEAN) { 2589 cb_fn(cb_arg, 0); 2590 return; 2591 } 2592 2593 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 2594 cpl.u.blob_basic.cb_fn = cb_fn; 2595 cpl.u.blob_basic.cb_arg = cb_arg; 2596 2597 seq = spdk_bs_sequence_start(blob->bs->md_target.md_channel, &cpl); 2598 if (!seq) { 2599 cb_fn(cb_arg, -ENOMEM); 2600 return; 2601 } 2602 2603 _spdk_blob_persist(seq, blob, _spdk_blob_sync_cpl, blob); 2604 } 2605 2606 /* END spdk_bs_md_sync_blob */ 2607 2608 /* START spdk_bs_md_close_blob */ 2609 2610 static void 2611 _spdk_blob_close_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2612 { 2613 struct spdk_blob **blob = cb_arg; 2614 2615 if ((*blob)->open_ref == 0) { 2616 TAILQ_REMOVE(&(*blob)->bs->blobs, (*blob), link); 2617 _spdk_blob_free((*blob)); 2618 } 2619 2620 *blob = NULL; 2621 2622 spdk_bs_sequence_finish(seq, bserrno); 2623 } 2624 2625 void spdk_bs_md_close_blob(struct spdk_blob **b, 2626 spdk_blob_op_complete cb_fn, void *cb_arg) 2627 { 2628 struct spdk_bs_cpl cpl; 2629 struct spdk_blob *blob; 2630 spdk_bs_sequence_t *seq; 2631 2632 assert(b != NULL); 2633 blob = *b; 2634 assert(blob != NULL); 2635 2636 SPDK_DEBUGLOG(SPDK_TRACE_BLOB, "Closing blob %lu\n", blob->id); 2637 2638 assert(blob->state != SPDK_BLOB_STATE_LOADING && 2639 blob->state != SPDK_BLOB_STATE_SYNCING); 2640 2641 if (blob->open_ref == 0) { 2642 cb_fn(cb_arg, -EBADF); 2643 return; 2644 } 2645 2646 blob->open_ref--; 2647 2648 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 2649 cpl.u.blob_basic.cb_fn = cb_fn; 2650 cpl.u.blob_basic.cb_arg = cb_arg; 2651 2652 seq = spdk_bs_sequence_start(blob->bs->md_target.md_channel, &cpl); 2653 if (!seq) { 2654 cb_fn(cb_arg, -ENOMEM); 2655 return; 2656 } 2657 2658 if (blob->state == SPDK_BLOB_STATE_CLEAN) { 2659 _spdk_blob_close_cpl(seq, b, 0); 2660 return; 2661 } 2662 2663 /* Sync metadata */ 2664 _spdk_blob_persist(seq, blob, _spdk_blob_close_cpl, b); 2665 } 2666 2667 /* END spdk_bs_md_close_blob */ 2668 2669 struct spdk_io_channel *spdk_bs_alloc_io_channel(struct spdk_blob_store *bs) 2670 { 2671 return spdk_get_io_channel(&bs->io_target); 2672 } 2673 2674 void spdk_bs_free_io_channel(struct spdk_io_channel *channel) 2675 { 2676 spdk_put_io_channel(channel); 2677 } 2678 2679 void spdk_bs_io_flush_channel(struct spdk_io_channel *channel, 2680 spdk_blob_op_complete cb_fn, void *cb_arg) 2681 { 2682 /* Flush is synchronous right now */ 2683 cb_fn(cb_arg, 0); 2684 } 2685 2686 void spdk_bs_io_write_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 2687 void *payload, uint64_t offset, uint64_t length, 2688 spdk_blob_op_complete cb_fn, void *cb_arg) 2689 { 2690 _spdk_blob_request_submit_rw(blob, channel, payload, offset, length, cb_fn, cb_arg, false); 2691 } 2692 2693 void spdk_bs_io_read_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 2694 void *payload, uint64_t offset, uint64_t length, 2695 spdk_blob_op_complete cb_fn, void *cb_arg) 2696 { 2697 _spdk_blob_request_submit_rw(blob, channel, payload, offset, length, cb_fn, cb_arg, true); 2698 } 2699 2700 void spdk_bs_io_writev_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 2701 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 2702 spdk_blob_op_complete cb_fn, void *cb_arg) 2703 { 2704 _spdk_blob_request_submit_rw_iov(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg, false); 2705 } 2706 2707 void spdk_bs_io_readv_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 2708 struct iovec *iov, int iovcnt, uint64_t offset, uint64_t length, 2709 spdk_blob_op_complete cb_fn, void *cb_arg) 2710 { 2711 _spdk_blob_request_submit_rw_iov(blob, channel, iov, iovcnt, offset, length, cb_fn, cb_arg, true); 2712 } 2713 2714 struct spdk_bs_iter_ctx { 2715 int64_t page_num; 2716 struct spdk_blob_store *bs; 2717 2718 spdk_blob_op_with_handle_complete cb_fn; 2719 void *cb_arg; 2720 }; 2721 2722 static void 2723 _spdk_bs_iter_cpl(void *cb_arg, struct spdk_blob *blob, int bserrno) 2724 { 2725 struct spdk_bs_iter_ctx *ctx = cb_arg; 2726 struct spdk_blob_store *bs = ctx->bs; 2727 spdk_blob_id id; 2728 2729 if (bserrno == 0) { 2730 ctx->cb_fn(ctx->cb_arg, blob, bserrno); 2731 free(ctx); 2732 return; 2733 } 2734 2735 ctx->page_num++; 2736 ctx->page_num = spdk_bit_array_find_first_set(bs->used_md_pages, ctx->page_num); 2737 if (ctx->page_num >= spdk_bit_array_capacity(bs->used_md_pages)) { 2738 ctx->cb_fn(ctx->cb_arg, NULL, -ENOENT); 2739 free(ctx); 2740 return; 2741 } 2742 2743 id = _spdk_bs_page_to_blobid(ctx->page_num); 2744 2745 blob = _spdk_blob_lookup(bs, id); 2746 if (blob) { 2747 blob->open_ref++; 2748 ctx->cb_fn(ctx->cb_arg, blob, 0); 2749 free(ctx); 2750 return; 2751 } 2752 2753 spdk_bs_md_open_blob(bs, id, _spdk_bs_iter_cpl, ctx); 2754 } 2755 2756 void 2757 spdk_bs_md_iter_first(struct spdk_blob_store *bs, 2758 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 2759 { 2760 struct spdk_bs_iter_ctx *ctx; 2761 2762 ctx = calloc(1, sizeof(*ctx)); 2763 if (!ctx) { 2764 cb_fn(cb_arg, NULL, -ENOMEM); 2765 return; 2766 } 2767 2768 ctx->page_num = -1; 2769 ctx->bs = bs; 2770 ctx->cb_fn = cb_fn; 2771 ctx->cb_arg = cb_arg; 2772 2773 _spdk_bs_iter_cpl(ctx, NULL, -1); 2774 } 2775 2776 static void 2777 _spdk_bs_iter_close_cpl(void *cb_arg, int bserrno) 2778 { 2779 struct spdk_bs_iter_ctx *ctx = cb_arg; 2780 2781 _spdk_bs_iter_cpl(ctx, NULL, -1); 2782 } 2783 2784 void 2785 spdk_bs_md_iter_next(struct spdk_blob_store *bs, struct spdk_blob **b, 2786 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 2787 { 2788 struct spdk_bs_iter_ctx *ctx; 2789 struct spdk_blob *blob; 2790 2791 assert(b != NULL); 2792 blob = *b; 2793 assert(blob != NULL); 2794 2795 ctx = calloc(1, sizeof(*ctx)); 2796 if (!ctx) { 2797 cb_fn(cb_arg, NULL, -ENOMEM); 2798 return; 2799 } 2800 2801 ctx->page_num = _spdk_bs_blobid_to_page(blob->id); 2802 ctx->bs = bs; 2803 ctx->cb_fn = cb_fn; 2804 ctx->cb_arg = cb_arg; 2805 2806 /* Close the existing blob */ 2807 spdk_bs_md_close_blob(b, _spdk_bs_iter_close_cpl, ctx); 2808 } 2809 2810 int 2811 spdk_blob_md_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 2812 uint16_t value_len) 2813 { 2814 struct spdk_xattr *xattr; 2815 2816 assert(blob != NULL); 2817 2818 assert(blob->state != SPDK_BLOB_STATE_LOADING && 2819 blob->state != SPDK_BLOB_STATE_SYNCING); 2820 2821 TAILQ_FOREACH(xattr, &blob->xattrs, link) { 2822 if (!strcmp(name, xattr->name)) { 2823 free(xattr->value); 2824 xattr->value_len = value_len; 2825 xattr->value = malloc(value_len); 2826 memcpy(xattr->value, value, value_len); 2827 2828 blob->state = SPDK_BLOB_STATE_DIRTY; 2829 2830 return 0; 2831 } 2832 } 2833 2834 xattr = calloc(1, sizeof(*xattr)); 2835 if (!xattr) { 2836 return -1; 2837 } 2838 xattr->name = strdup(name); 2839 xattr->value_len = value_len; 2840 xattr->value = malloc(value_len); 2841 memcpy(xattr->value, value, value_len); 2842 TAILQ_INSERT_TAIL(&blob->xattrs, xattr, link); 2843 2844 blob->state = SPDK_BLOB_STATE_DIRTY; 2845 2846 return 0; 2847 } 2848 2849 int 2850 spdk_blob_md_remove_xattr(struct spdk_blob *blob, const char *name) 2851 { 2852 struct spdk_xattr *xattr; 2853 2854 assert(blob != NULL); 2855 2856 assert(blob->state != SPDK_BLOB_STATE_LOADING && 2857 blob->state != SPDK_BLOB_STATE_SYNCING); 2858 2859 TAILQ_FOREACH(xattr, &blob->xattrs, link) { 2860 if (!strcmp(name, xattr->name)) { 2861 TAILQ_REMOVE(&blob->xattrs, xattr, link); 2862 free(xattr->value); 2863 free(xattr->name); 2864 free(xattr); 2865 2866 blob->state = SPDK_BLOB_STATE_DIRTY; 2867 2868 return 0; 2869 } 2870 } 2871 2872 return -ENOENT; 2873 } 2874 2875 int 2876 spdk_bs_md_get_xattr_value(struct spdk_blob *blob, const char *name, 2877 const void **value, size_t *value_len) 2878 { 2879 struct spdk_xattr *xattr; 2880 2881 TAILQ_FOREACH(xattr, &blob->xattrs, link) { 2882 if (!strcmp(name, xattr->name)) { 2883 *value = xattr->value; 2884 *value_len = xattr->value_len; 2885 return 0; 2886 } 2887 } 2888 2889 return -ENOENT; 2890 } 2891 2892 struct spdk_xattr_names { 2893 uint32_t count; 2894 const char *names[0]; 2895 }; 2896 2897 int 2898 spdk_bs_md_get_xattr_names(struct spdk_blob *blob, 2899 struct spdk_xattr_names **names) 2900 { 2901 struct spdk_xattr *xattr; 2902 int count = 0; 2903 2904 TAILQ_FOREACH(xattr, &blob->xattrs, link) { 2905 count++; 2906 } 2907 2908 *names = calloc(1, sizeof(struct spdk_xattr_names) + count * sizeof(char *)); 2909 if (*names == NULL) { 2910 return -ENOMEM; 2911 } 2912 2913 TAILQ_FOREACH(xattr, &blob->xattrs, link) { 2914 (*names)->names[(*names)->count++] = xattr->name; 2915 } 2916 2917 return 0; 2918 } 2919 2920 uint32_t 2921 spdk_xattr_names_get_count(struct spdk_xattr_names *names) 2922 { 2923 assert(names != NULL); 2924 2925 return names->count; 2926 } 2927 2928 const char * 2929 spdk_xattr_names_get_name(struct spdk_xattr_names *names, uint32_t index) 2930 { 2931 if (index >= names->count) { 2932 return NULL; 2933 } 2934 2935 return names->names[index]; 2936 } 2937 2938 void 2939 spdk_xattr_names_free(struct spdk_xattr_names *names) 2940 { 2941 free(names); 2942 } 2943 2944 struct spdk_bs_type 2945 spdk_bs_get_bstype(struct spdk_blob_store *bs) 2946 { 2947 return bs->bstype; 2948 } 2949 2950 void 2951 spdk_bs_set_bstype(struct spdk_blob_store *bs, struct spdk_bs_type bstype) 2952 { 2953 memcpy(&bs->bstype, &bstype, sizeof(bstype)); 2954 } 2955 2956 SPDK_LOG_REGISTER_TRACE_FLAG("blob", SPDK_TRACE_BLOB); 2957