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/env.h" 38 #include "spdk/queue.h" 39 #include "spdk/io_channel.h" 40 #include "spdk/bit_array.h" 41 42 #include "spdk_internal/log.h" 43 44 #include "blobstore.h" 45 #include "request.h" 46 47 static inline size_t 48 divide_round_up(size_t num, size_t divisor) 49 { 50 return (num + divisor - 1) / divisor; 51 } 52 53 static void 54 _spdk_bs_claim_cluster(struct spdk_blob_store *bs, uint32_t cluster_num) 55 { 56 assert(cluster_num < spdk_bit_array_capacity(bs->used_clusters)); 57 assert(spdk_bit_array_get(bs->used_clusters, cluster_num) == false); 58 assert(bs->num_free_clusters > 0); 59 60 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Claiming cluster %u\n", cluster_num); 61 62 spdk_bit_array_set(bs->used_clusters, cluster_num); 63 bs->num_free_clusters--; 64 } 65 66 static void 67 _spdk_bs_release_cluster(struct spdk_blob_store *bs, uint32_t cluster_num) 68 { 69 assert(cluster_num < spdk_bit_array_capacity(bs->used_clusters)); 70 assert(spdk_bit_array_get(bs->used_clusters, cluster_num) == true); 71 assert(bs->num_free_clusters < bs->total_clusters); 72 73 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Releasing cluster %u\n", cluster_num); 74 75 spdk_bit_array_clear(bs->used_clusters, cluster_num); 76 bs->num_free_clusters++; 77 } 78 79 static struct spdk_blob * 80 _spdk_blob_alloc(struct spdk_blob_store *bs, spdk_blob_id id) 81 { 82 struct spdk_blob *blob; 83 84 blob = calloc(1, sizeof(*blob)); 85 if (!blob) { 86 return NULL; 87 } 88 89 blob->id = id; 90 blob->bs = bs; 91 92 blob->state = SPDK_BLOB_STATE_DIRTY; 93 blob->active.num_pages = 1; 94 blob->active.pages = calloc(1, sizeof(*blob->active.pages)); 95 if (!blob->active.pages) { 96 free(blob); 97 return NULL; 98 } 99 100 blob->active.pages[0] = _spdk_bs_blobid_to_page(id); 101 102 TAILQ_INIT(&blob->xattrs); 103 104 return blob; 105 } 106 107 static void 108 _spdk_blob_free(struct spdk_blob *blob) 109 { 110 struct spdk_xattr *xattr, *xattr_tmp; 111 112 assert(blob != NULL); 113 114 free(blob->active.clusters); 115 free(blob->clean.clusters); 116 free(blob->active.pages); 117 free(blob->clean.pages); 118 119 TAILQ_FOREACH_SAFE(xattr, &blob->xattrs, link, xattr_tmp) { 120 TAILQ_REMOVE(&blob->xattrs, xattr, link); 121 free(xattr->name); 122 free(xattr->value); 123 free(xattr); 124 } 125 126 free(blob); 127 } 128 129 static int 130 _spdk_blob_mark_clean(struct spdk_blob *blob) 131 { 132 uint64_t *clusters = NULL; 133 uint32_t *pages = NULL; 134 135 assert(blob != NULL); 136 assert(blob->state == SPDK_BLOB_STATE_LOADING || 137 blob->state == SPDK_BLOB_STATE_SYNCING); 138 139 if (blob->active.num_clusters) { 140 assert(blob->active.clusters); 141 clusters = calloc(blob->active.num_clusters, sizeof(*blob->active.clusters)); 142 if (!clusters) { 143 return -1; 144 } 145 memcpy(clusters, blob->active.clusters, blob->active.num_clusters * sizeof(*clusters)); 146 } 147 148 if (blob->active.num_pages) { 149 assert(blob->active.pages); 150 pages = calloc(blob->active.num_pages, sizeof(*blob->active.pages)); 151 if (!pages) { 152 free(clusters); 153 return -1; 154 } 155 memcpy(pages, blob->active.pages, blob->active.num_pages * sizeof(*pages)); 156 } 157 158 free(blob->clean.clusters); 159 free(blob->clean.pages); 160 161 blob->clean.num_clusters = blob->active.num_clusters; 162 blob->clean.clusters = blob->active.clusters; 163 blob->clean.num_pages = blob->active.num_pages; 164 blob->clean.pages = blob->active.pages; 165 166 blob->active.clusters = clusters; 167 blob->active.pages = pages; 168 169 blob->state = SPDK_BLOB_STATE_CLEAN; 170 171 return 0; 172 } 173 174 static void 175 _spdk_blob_parse_page(const struct spdk_blob_md_page *page, struct spdk_blob *blob) 176 { 177 struct spdk_blob_md_descriptor *desc; 178 size_t cur_desc = 0; 179 void *tmp; 180 181 desc = (struct spdk_blob_md_descriptor *)page->descriptors; 182 while (cur_desc < sizeof(page->descriptors)) { 183 if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_PADDING) { 184 if (desc->length == 0) { 185 /* If padding and length are 0, this terminates the page */ 186 break; 187 } 188 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_EXTENT) { 189 struct spdk_blob_md_descriptor_extent *desc_extent; 190 unsigned int i, j; 191 unsigned int cluster_count = blob->active.num_clusters; 192 193 desc_extent = (struct spdk_blob_md_descriptor_extent *)desc; 194 195 assert(desc_extent->length > 0); 196 assert(desc_extent->length % sizeof(desc_extent->extents[0]) == 0); 197 198 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 199 for (j = 0; j < desc_extent->extents[i].length; j++) { 200 assert(spdk_bit_array_get(blob->bs->used_clusters, desc_extent->extents[i].cluster_idx + j)); 201 cluster_count++; 202 } 203 } 204 205 assert(cluster_count > 0); 206 tmp = realloc(blob->active.clusters, cluster_count * sizeof(uint64_t)); 207 assert(tmp != NULL); 208 blob->active.clusters = tmp; 209 blob->active.cluster_array_size = cluster_count; 210 211 for (i = 0; i < desc_extent->length / sizeof(desc_extent->extents[0]); i++) { 212 for (j = 0; j < desc_extent->extents[i].length; j++) { 213 blob->active.clusters[blob->active.num_clusters++] = _spdk_bs_cluster_to_lba(blob->bs, 214 desc_extent->extents[i].cluster_idx + j); 215 } 216 } 217 218 } else if (desc->type == SPDK_MD_DESCRIPTOR_TYPE_XATTR) { 219 struct spdk_blob_md_descriptor_xattr *desc_xattr; 220 struct spdk_xattr *xattr; 221 222 desc_xattr = (struct spdk_blob_md_descriptor_xattr *)desc; 223 224 assert(desc_xattr->length == sizeof(desc_xattr->name_length) + 225 sizeof(desc_xattr->value_length) + 226 desc_xattr->name_length + desc_xattr->value_length); 227 228 xattr = calloc(1, sizeof(*xattr)); 229 assert(xattr != NULL); 230 231 xattr->name = malloc(desc_xattr->name_length + 1); 232 assert(xattr->name); 233 strncpy(xattr->name, desc_xattr->name, desc_xattr->name_length); 234 xattr->name[desc_xattr->name_length] = '\0'; 235 236 xattr->value = malloc(desc_xattr->value_length); 237 assert(xattr->value != NULL); 238 xattr->value_len = desc_xattr->value_length; 239 memcpy(xattr->value, 240 (void *)((uintptr_t)desc_xattr->name + desc_xattr->name_length), 241 desc_xattr->value_length); 242 243 TAILQ_INSERT_TAIL(&blob->xattrs, xattr, link); 244 } else { 245 /* Error */ 246 break; 247 } 248 249 /* Advance to the next descriptor */ 250 cur_desc += sizeof(*desc) + desc->length; 251 if (cur_desc + sizeof(*desc) > sizeof(page->descriptors)) { 252 break; 253 } 254 desc = (struct spdk_blob_md_descriptor *)((uintptr_t)page->descriptors + cur_desc); 255 } 256 } 257 258 static int 259 _spdk_blob_parse(const struct spdk_blob_md_page *pages, uint32_t page_count, 260 struct spdk_blob *blob) 261 { 262 const struct spdk_blob_md_page *page; 263 uint32_t i; 264 265 assert(page_count > 0); 266 assert(pages[0].sequence_num == 0); 267 assert(blob != NULL); 268 assert(blob->state == SPDK_BLOB_STATE_LOADING); 269 assert(blob->active.clusters == NULL); 270 assert(blob->id == pages[0].id); 271 assert(blob->state == SPDK_BLOB_STATE_LOADING); 272 273 for (i = 0; i < page_count; i++) { 274 page = &pages[i]; 275 276 assert(page->id == blob->id); 277 assert(page->sequence_num == i); 278 279 _spdk_blob_parse_page(page, blob); 280 } 281 282 return 0; 283 } 284 285 static int 286 _spdk_blob_serialize_add_page(const struct spdk_blob *blob, 287 struct spdk_blob_md_page **pages, 288 uint32_t *page_count, 289 struct spdk_blob_md_page **last_page) 290 { 291 struct spdk_blob_md_page *page; 292 293 assert(pages != NULL); 294 assert(page_count != NULL); 295 296 if (*page_count == 0) { 297 assert(*pages == NULL); 298 *page_count = 1; 299 *pages = spdk_malloc(sizeof(struct spdk_blob_md_page), 300 sizeof(struct spdk_blob_md_page), 301 NULL); 302 } else { 303 assert(*pages != NULL); 304 (*page_count)++; 305 *pages = spdk_realloc(*pages, 306 sizeof(struct spdk_blob_md_page) * (*page_count), 307 sizeof(struct spdk_blob_md_page), 308 NULL); 309 } 310 311 if (*pages == NULL) { 312 *page_count = 0; 313 *last_page = NULL; 314 return -ENOMEM; 315 } 316 317 page = &(*pages)[*page_count - 1]; 318 memset(page, 0, sizeof(*page)); 319 page->id = blob->id; 320 page->sequence_num = *page_count - 1; 321 page->next = SPDK_INVALID_MD_PAGE; 322 *last_page = page; 323 324 return 0; 325 } 326 327 /* Transform the in-memory representation 'xattr' into an on-disk xattr descriptor. 328 * Update required_sz on both success and failure. 329 * 330 */ 331 static int 332 _spdk_blob_serialize_xattr(const struct spdk_xattr *xattr, 333 uint8_t *buf, size_t buf_sz, 334 size_t *required_sz) 335 { 336 struct spdk_blob_md_descriptor_xattr *desc; 337 338 *required_sz = sizeof(struct spdk_blob_md_descriptor_xattr) + 339 strlen(xattr->name) + 340 xattr->value_len; 341 342 if (buf_sz < *required_sz) { 343 return -1; 344 } 345 346 desc = (struct spdk_blob_md_descriptor_xattr *)buf; 347 348 desc->type = SPDK_MD_DESCRIPTOR_TYPE_XATTR; 349 desc->length = sizeof(desc->name_length) + 350 sizeof(desc->value_length) + 351 strlen(xattr->name) + 352 xattr->value_len; 353 desc->name_length = strlen(xattr->name); 354 desc->value_length = xattr->value_len; 355 356 memcpy(desc->name, xattr->name, desc->name_length); 357 memcpy((void *)((uintptr_t)desc->name + desc->name_length), 358 xattr->value, 359 desc->value_length); 360 361 return 0; 362 } 363 364 static void 365 _spdk_blob_serialize_extent(const struct spdk_blob *blob, 366 uint64_t start_cluster, uint64_t *next_cluster, 367 uint8_t *buf, size_t buf_sz) 368 { 369 struct spdk_blob_md_descriptor_extent *desc; 370 size_t cur_sz; 371 uint64_t i, extent_idx; 372 uint32_t lba, lba_per_cluster, lba_count; 373 374 /* The buffer must have room for at least one extent */ 375 cur_sz = sizeof(struct spdk_blob_md_descriptor) + sizeof(desc->extents[0]); 376 if (buf_sz < cur_sz) { 377 *next_cluster = start_cluster; 378 return; 379 } 380 381 desc = (struct spdk_blob_md_descriptor_extent *)buf; 382 desc->type = SPDK_MD_DESCRIPTOR_TYPE_EXTENT; 383 384 lba_per_cluster = _spdk_bs_cluster_to_lba(blob->bs, 1); 385 386 lba = blob->active.clusters[start_cluster]; 387 lba_count = lba_per_cluster; 388 extent_idx = 0; 389 for (i = start_cluster + 1; i < blob->active.num_clusters; i++) { 390 if ((lba + lba_count) == blob->active.clusters[i]) { 391 lba_count += lba_per_cluster; 392 continue; 393 } 394 desc->extents[extent_idx].cluster_idx = lba / lba_per_cluster; 395 desc->extents[extent_idx].length = lba_count / lba_per_cluster; 396 extent_idx++; 397 398 cur_sz += sizeof(desc->extents[extent_idx]); 399 400 if (buf_sz < cur_sz) { 401 /* If we ran out of buffer space, return */ 402 desc->length = sizeof(desc->extents[0]) * extent_idx; 403 *next_cluster = i; 404 return; 405 } 406 407 lba = blob->active.clusters[i]; 408 lba_count = lba_per_cluster; 409 } 410 411 desc->extents[extent_idx].cluster_idx = lba / lba_per_cluster; 412 desc->extents[extent_idx].length = lba_count / lba_per_cluster; 413 extent_idx++; 414 415 desc->length = sizeof(desc->extents[0]) * extent_idx; 416 *next_cluster = blob->active.num_clusters; 417 418 return; 419 } 420 421 static int 422 _spdk_blob_serialize(const struct spdk_blob *blob, struct spdk_blob_md_page **pages, 423 uint32_t *page_count) 424 { 425 struct spdk_blob_md_page *cur_page; 426 const struct spdk_xattr *xattr; 427 int rc; 428 uint8_t *buf; 429 size_t remaining_sz; 430 431 assert(pages != NULL); 432 assert(page_count != NULL); 433 assert(blob != NULL); 434 assert(blob->state == SPDK_BLOB_STATE_SYNCING); 435 436 *pages = NULL; 437 *page_count = 0; 438 439 /* A blob always has at least 1 page, even if it has no descriptors */ 440 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, &cur_page); 441 if (rc < 0) { 442 return rc; 443 } 444 445 buf = (uint8_t *)cur_page->descriptors; 446 remaining_sz = sizeof(cur_page->descriptors); 447 448 /* Serialize xattrs */ 449 TAILQ_FOREACH(xattr, &blob->xattrs, link) { 450 size_t required_sz = 0; 451 rc = _spdk_blob_serialize_xattr(xattr, 452 buf, remaining_sz, 453 &required_sz); 454 if (rc < 0) { 455 /* Need to add a new page to the chain */ 456 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, 457 &cur_page); 458 if (rc < 0) { 459 spdk_free(*pages); 460 *pages = NULL; 461 *page_count = 0; 462 return rc; 463 } 464 465 buf = (uint8_t *)cur_page->descriptors; 466 remaining_sz = sizeof(cur_page->descriptors); 467 468 /* Try again */ 469 required_sz = 0; 470 rc = _spdk_blob_serialize_xattr(xattr, 471 buf, remaining_sz, 472 &required_sz); 473 474 if (rc < 0) { 475 spdk_free(*pages); 476 *pages = NULL; 477 *page_count = 0; 478 return -1; 479 } 480 } 481 482 remaining_sz -= required_sz; 483 buf += required_sz; 484 } 485 486 /* Serialize extents */ 487 uint64_t last_cluster = 0; 488 while (last_cluster < blob->active.num_clusters) { 489 _spdk_blob_serialize_extent(blob, last_cluster, &last_cluster, 490 buf, remaining_sz); 491 492 if (last_cluster == blob->active.num_clusters) { 493 break; 494 } 495 496 rc = _spdk_blob_serialize_add_page(blob, pages, page_count, 497 &cur_page); 498 if (rc < 0) { 499 return rc; 500 } 501 502 buf = (uint8_t *)cur_page->descriptors; 503 remaining_sz = sizeof(cur_page->descriptors); 504 } 505 506 return 0; 507 } 508 509 struct spdk_blob_load_ctx { 510 struct spdk_blob *blob; 511 512 struct spdk_blob_md_page *pages; 513 uint32_t num_pages; 514 515 spdk_bs_sequence_cpl cb_fn; 516 void *cb_arg; 517 }; 518 519 static void 520 _spdk_blob_load_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 521 { 522 struct spdk_blob_load_ctx *ctx = cb_arg; 523 struct spdk_blob *blob = ctx->blob; 524 struct spdk_blob_md_page *page; 525 int rc; 526 527 page = &ctx->pages[ctx->num_pages - 1]; 528 529 if (page->next != SPDK_INVALID_MD_PAGE) { 530 uint32_t next_page = page->next; 531 uint64_t next_lba = _spdk_bs_page_to_lba(blob->bs, blob->bs->md_start + next_page); 532 533 534 assert(next_lba < (blob->bs->md_start + blob->bs->md_len)); 535 536 /* Read the next page */ 537 ctx->num_pages++; 538 ctx->pages = spdk_realloc(ctx->pages, (sizeof(*page) * ctx->num_pages), 539 sizeof(*page), NULL); 540 if (ctx->pages == NULL) { 541 ctx->cb_fn(seq, ctx->cb_arg, -ENOMEM); 542 free(ctx); 543 return; 544 } 545 546 spdk_bs_sequence_read(seq, &ctx->pages[ctx->num_pages - 1], 547 next_lba, 548 _spdk_bs_byte_to_lba(blob->bs, sizeof(*page)), 549 _spdk_blob_load_cpl, ctx); 550 return; 551 } 552 553 /* Parse the pages */ 554 rc = _spdk_blob_parse(ctx->pages, ctx->num_pages, blob); 555 556 _spdk_blob_mark_clean(blob); 557 558 ctx->cb_fn(seq, ctx->cb_arg, rc); 559 560 /* Free the memory */ 561 spdk_free(ctx->pages); 562 free(ctx); 563 } 564 565 /* Load a blob from disk given a blobid */ 566 static void 567 _spdk_blob_load(spdk_bs_sequence_t *seq, struct spdk_blob *blob, 568 spdk_bs_sequence_cpl cb_fn, void *cb_arg) 569 { 570 struct spdk_blob_load_ctx *ctx; 571 struct spdk_blob_store *bs; 572 uint32_t page_num; 573 uint64_t lba; 574 575 assert(blob != NULL); 576 assert(blob->state == SPDK_BLOB_STATE_CLEAN || 577 blob->state == SPDK_BLOB_STATE_DIRTY); 578 579 bs = blob->bs; 580 581 ctx = calloc(1, sizeof(*ctx)); 582 if (!ctx) { 583 cb_fn(seq, cb_arg, -ENOMEM); 584 return; 585 } 586 587 ctx->blob = blob; 588 ctx->pages = spdk_realloc(ctx->pages, sizeof(struct spdk_blob_md_page), 589 sizeof(struct spdk_blob_md_page), NULL); 590 if (!ctx->pages) { 591 free(ctx); 592 cb_fn(seq, cb_arg, -ENOMEM); 593 return; 594 } 595 ctx->num_pages = 1; 596 ctx->cb_fn = cb_fn; 597 ctx->cb_arg = cb_arg; 598 599 page_num = _spdk_bs_blobid_to_page(blob->id); 600 lba = _spdk_bs_page_to_lba(blob->bs, bs->md_start + page_num); 601 602 blob->state = SPDK_BLOB_STATE_LOADING; 603 604 spdk_bs_sequence_read(seq, &ctx->pages[0], lba, 605 _spdk_bs_byte_to_lba(bs, sizeof(struct spdk_blob_md_page)), 606 _spdk_blob_load_cpl, ctx); 607 } 608 609 struct spdk_blob_persist_ctx { 610 struct spdk_blob *blob; 611 612 struct spdk_blob_md_page *pages; 613 614 uint64_t idx; 615 616 spdk_bs_sequence_cpl cb_fn; 617 void *cb_arg; 618 }; 619 620 static void 621 _spdk_blob_persist_complete(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 622 { 623 struct spdk_blob_persist_ctx *ctx = cb_arg; 624 struct spdk_blob *blob = ctx->blob; 625 626 if (bserrno == 0) { 627 _spdk_blob_mark_clean(blob); 628 } 629 630 /* Call user callback */ 631 ctx->cb_fn(seq, ctx->cb_arg, bserrno); 632 633 /* Free the memory */ 634 spdk_free(ctx->pages); 635 free(ctx); 636 } 637 638 static void 639 _spdk_blob_persist_unmap_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 640 { 641 struct spdk_blob_persist_ctx *ctx = cb_arg; 642 struct spdk_blob *blob = ctx->blob; 643 struct spdk_blob_store *bs = blob->bs; 644 void *tmp; 645 size_t i; 646 647 /* Release all clusters that were truncated */ 648 for (i = blob->active.num_clusters; i < blob->active.cluster_array_size; i++) { 649 uint32_t cluster_num = _spdk_bs_lba_to_cluster(bs, blob->active.clusters[i]); 650 651 _spdk_bs_release_cluster(bs, cluster_num); 652 } 653 654 if (blob->active.num_clusters == 0) { 655 free(blob->active.clusters); 656 blob->active.clusters = NULL; 657 blob->active.cluster_array_size = 0; 658 } else { 659 tmp = realloc(blob->active.clusters, sizeof(uint64_t) * blob->active.num_clusters); 660 assert(tmp != NULL); 661 blob->active.clusters = tmp; 662 blob->active.cluster_array_size = blob->active.num_clusters; 663 } 664 665 _spdk_blob_persist_complete(seq, ctx, bserrno); 666 } 667 668 static void 669 _spdk_blob_persist_unmap_clusters(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 670 { 671 struct spdk_blob_persist_ctx *ctx = cb_arg; 672 struct spdk_blob *blob = ctx->blob; 673 struct spdk_blob_store *bs = blob->bs; 674 spdk_bs_batch_t *batch; 675 size_t i; 676 677 /* Clusters don't move around in blobs. The list shrinks or grows 678 * at the end, but no changes ever occur in the middle of the list. 679 */ 680 681 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_unmap_clusters_cpl, ctx); 682 683 /* Unmap all clusters that were truncated */ 684 for (i = blob->active.num_clusters; i < blob->active.cluster_array_size; i++) { 685 uint64_t lba = blob->active.clusters[i]; 686 uint32_t lba_count = _spdk_bs_cluster_to_lba(bs, 1); 687 688 spdk_bs_batch_unmap(batch, lba, lba_count); 689 } 690 691 spdk_bs_batch_close(batch); 692 } 693 694 static void 695 _spdk_blob_persist_unmap_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 696 { 697 struct spdk_blob_persist_ctx *ctx = cb_arg; 698 struct spdk_blob *blob = ctx->blob; 699 struct spdk_blob_store *bs = blob->bs; 700 size_t i; 701 702 /* This loop starts at 1 because the first page is special and handled 703 * below. The pages (except the first) are never written in place, 704 * so any pages in the clean list must be unmapped. 705 */ 706 for (i = 1; i < blob->clean.num_pages; i++) { 707 spdk_bit_array_clear(bs->used_md_pages, blob->clean.pages[i]); 708 } 709 710 if (blob->active.num_pages == 0) { 711 uint32_t page_num; 712 713 page_num = _spdk_bs_blobid_to_page(blob->id); 714 spdk_bit_array_clear(bs->used_md_pages, page_num); 715 } 716 717 /* Move on to unmapping clusters */ 718 _spdk_blob_persist_unmap_clusters(seq, ctx, 0); 719 } 720 721 static void 722 _spdk_blob_persist_unmap_pages(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 723 { 724 struct spdk_blob_persist_ctx *ctx = cb_arg; 725 struct spdk_blob *blob = ctx->blob; 726 struct spdk_blob_store *bs = blob->bs; 727 uint64_t lba; 728 uint32_t lba_count; 729 spdk_bs_batch_t *batch; 730 size_t i; 731 732 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_unmap_pages_cpl, ctx); 733 734 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(struct spdk_blob_md_page)); 735 736 /* This loop starts at 1 because the first page is special and handled 737 * below. The pages (except the first) are never written in place, 738 * so any pages in the clean list must be unmapped. 739 */ 740 for (i = 1; i < blob->clean.num_pages; i++) { 741 lba = _spdk_bs_page_to_lba(bs, bs->md_start + blob->clean.pages[i]); 742 743 spdk_bs_batch_unmap(batch, lba, lba_count); 744 } 745 746 /* The first page will only be unmapped if this is a delete. */ 747 if (blob->active.num_pages == 0) { 748 uint32_t page_num; 749 750 /* The first page in the metadata goes where the blobid indicates */ 751 page_num = _spdk_bs_blobid_to_page(blob->id); 752 lba = _spdk_bs_page_to_lba(bs, bs->md_start + page_num); 753 754 spdk_bs_batch_unmap(batch, lba, lba_count); 755 } 756 757 spdk_bs_batch_close(batch); 758 } 759 760 static void 761 _spdk_blob_persist_write_page_root(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 762 { 763 struct spdk_blob_persist_ctx *ctx = cb_arg; 764 struct spdk_blob *blob = ctx->blob; 765 struct spdk_blob_store *bs = blob->bs; 766 uint64_t lba; 767 uint32_t lba_count; 768 struct spdk_blob_md_page *page; 769 770 if (blob->active.num_pages == 0) { 771 /* Move on to the next step */ 772 _spdk_blob_persist_unmap_pages(seq, ctx, 0); 773 return; 774 } 775 776 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(*page)); 777 778 page = &ctx->pages[0]; 779 /* The first page in the metadata goes where the blobid indicates */ 780 lba = _spdk_bs_page_to_lba(bs, bs->md_start + _spdk_bs_blobid_to_page(blob->id)); 781 782 spdk_bs_sequence_write(seq, page, lba, lba_count, 783 _spdk_blob_persist_unmap_pages, ctx); 784 } 785 786 static void 787 _spdk_blob_persist_write_page_chain(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 788 { 789 struct spdk_blob_persist_ctx *ctx = cb_arg; 790 struct spdk_blob *blob = ctx->blob; 791 struct spdk_blob_store *bs = blob->bs; 792 uint64_t lba; 793 uint32_t lba_count; 794 struct spdk_blob_md_page *page; 795 spdk_bs_batch_t *batch; 796 size_t i; 797 798 /* Clusters don't move around in blobs. The list shrinks or grows 799 * at the end, but no changes ever occur in the middle of the list. 800 */ 801 802 lba_count = _spdk_bs_byte_to_lba(bs, sizeof(*page)); 803 804 batch = spdk_bs_sequence_to_batch(seq, _spdk_blob_persist_write_page_root, ctx); 805 806 /* This starts at 1. The root page is not written until 807 * all of the others are finished 808 */ 809 for (i = 1; i < blob->active.num_pages; i++) { 810 page = &ctx->pages[i]; 811 assert(page->sequence_num == i); 812 813 lba = _spdk_bs_page_to_lba(bs, bs->md_start + blob->active.pages[i]); 814 815 spdk_bs_batch_write(batch, page, lba, lba_count); 816 } 817 818 spdk_bs_batch_close(batch); 819 } 820 821 static int 822 _spdk_resize_blob(struct spdk_blob *blob, uint64_t sz) 823 { 824 uint64_t i; 825 uint64_t *tmp; 826 uint64_t lfc; /* lowest free cluster */ 827 struct spdk_blob_store *bs; 828 829 bs = blob->bs; 830 831 assert(blob->state != SPDK_BLOB_STATE_LOADING && 832 blob->state != SPDK_BLOB_STATE_SYNCING); 833 834 if (blob->active.num_clusters == sz) { 835 return 0; 836 } 837 838 if (blob->active.num_clusters < blob->active.cluster_array_size) { 839 /* If this blob was resized to be larger, then smaller, then 840 * larger without syncing, then the cluster array already 841 * contains spare assigned clusters we can use. 842 */ 843 blob->active.num_clusters = spdk_min(blob->active.cluster_array_size, 844 sz); 845 } 846 847 blob->state = SPDK_BLOB_STATE_DIRTY; 848 849 /* Do two passes - one to verify that we can obtain enough clusters 850 * and another to actually claim them. 851 */ 852 853 lfc = 0; 854 for (i = blob->active.num_clusters; i < sz; i++) { 855 lfc = spdk_bit_array_find_first_clear(bs->used_clusters, lfc); 856 if (lfc >= bs->total_clusters) { 857 /* No more free clusters. Cannot satisfy the request */ 858 assert(false); 859 return -1; 860 } 861 lfc++; 862 } 863 864 if (sz > blob->active.num_clusters) { 865 /* Expand the cluster array if necessary. 866 * We only shrink the array when persisting. 867 */ 868 tmp = realloc(blob->active.clusters, sizeof(uint64_t) * sz); 869 if (sz > 0 && tmp == NULL) { 870 assert(false); 871 return -1; 872 } 873 blob->active.clusters = tmp; 874 blob->active.cluster_array_size = sz; 875 } 876 877 lfc = 0; 878 for (i = blob->active.num_clusters; i < sz; i++) { 879 lfc = spdk_bit_array_find_first_clear(bs->used_clusters, lfc); 880 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Claiming cluster %lu for blob %lu\n", lfc, blob->id); 881 _spdk_bs_claim_cluster(bs, lfc); 882 blob->active.clusters[i] = _spdk_bs_cluster_to_lba(bs, lfc); 883 lfc++; 884 } 885 886 blob->active.num_clusters = sz; 887 888 return 0; 889 } 890 891 /* Write a blob to disk */ 892 static void 893 _spdk_blob_persist(spdk_bs_sequence_t *seq, struct spdk_blob *blob, 894 spdk_bs_sequence_cpl cb_fn, void *cb_arg) 895 { 896 struct spdk_blob_persist_ctx *ctx; 897 int rc; 898 uint64_t i; 899 uint32_t page_num; 900 struct spdk_blob_store *bs; 901 902 assert(blob != NULL); 903 assert(blob->state == SPDK_BLOB_STATE_CLEAN || 904 blob->state == SPDK_BLOB_STATE_DIRTY); 905 906 if (blob->state == SPDK_BLOB_STATE_CLEAN) { 907 cb_fn(seq, cb_arg, 0); 908 return; 909 } 910 911 bs = blob->bs; 912 913 ctx = calloc(1, sizeof(*ctx)); 914 if (!ctx) { 915 cb_fn(seq, cb_arg, -ENOMEM); 916 return; 917 } 918 ctx->blob = blob; 919 ctx->cb_fn = cb_fn; 920 ctx->cb_arg = cb_arg; 921 922 blob->state = SPDK_BLOB_STATE_SYNCING; 923 924 if (blob->active.num_pages == 0) { 925 /* This is the signal that the blob should be deleted. 926 * Immediately jump to the clean up routine. */ 927 assert(blob->clean.num_pages > 0); 928 ctx->idx = blob->clean.num_pages - 1; 929 _spdk_blob_persist_unmap_pages(seq, ctx, 0); 930 return; 931 932 } 933 934 /* Generate the new metadata */ 935 rc = _spdk_blob_serialize(blob, &ctx->pages, &blob->active.num_pages); 936 if (rc < 0) { 937 free(ctx); 938 cb_fn(seq, cb_arg, rc); 939 return; 940 } 941 942 assert(blob->active.num_pages >= 1); 943 944 /* Resize the cache of page indices */ 945 blob->active.pages = realloc(blob->active.pages, 946 blob->active.num_pages * sizeof(*blob->active.pages)); 947 if (!blob->active.pages) { 948 free(ctx); 949 cb_fn(seq, cb_arg, -ENOMEM); 950 return; 951 } 952 953 /* Assign this metadata to pages. This requires two passes - 954 * one to verify that there are enough pages and a second 955 * to actually claim them. */ 956 page_num = 0; 957 /* Note that this loop starts at one. The first page location is fixed by the blobid. */ 958 for (i = 1; i < blob->active.num_pages; i++) { 959 page_num = spdk_bit_array_find_first_clear(bs->used_md_pages, page_num); 960 if (page_num >= spdk_bit_array_capacity(bs->used_md_pages)) { 961 spdk_free(ctx->pages); 962 free(ctx); 963 blob->state = SPDK_BLOB_STATE_DIRTY; 964 cb_fn(seq, cb_arg, -ENOMEM); 965 return; 966 } 967 page_num++; 968 } 969 970 page_num = 0; 971 blob->active.pages[0] = _spdk_bs_blobid_to_page(blob->id); 972 for (i = 1; i < blob->active.num_pages; i++) { 973 page_num = spdk_bit_array_find_first_clear(bs->used_md_pages, page_num); 974 ctx->pages[i - 1].next = page_num; 975 blob->active.pages[i] = page_num; 976 spdk_bit_array_set(bs->used_md_pages, page_num); 977 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Claiming page %u for blob %lu\n", page_num, blob->id); 978 page_num++; 979 } 980 981 /* Start writing the metadata from last page to first */ 982 ctx->idx = blob->active.num_pages - 1; 983 _spdk_blob_persist_write_page_chain(seq, ctx, 0); 984 } 985 986 static void 987 _spdk_blob_request_submit_rw(struct spdk_blob *blob, struct spdk_io_channel *_channel, 988 void *payload, uint64_t offset, uint64_t length, 989 spdk_blob_op_complete cb_fn, void *cb_arg, bool read) 990 { 991 spdk_bs_batch_t *batch; 992 struct spdk_bs_cpl cpl; 993 uint64_t lba; 994 uint32_t lba_count; 995 uint8_t *buf; 996 uint64_t page; 997 998 assert(blob != NULL); 999 1000 if (offset + length > blob->active.num_clusters * blob->bs->pages_per_cluster) { 1001 cb_fn(cb_arg, -EINVAL); 1002 return; 1003 } 1004 1005 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1006 cpl.u.blob_basic.cb_fn = cb_fn; 1007 cpl.u.blob_basic.cb_arg = cb_arg; 1008 1009 batch = spdk_bs_batch_open(_channel, &cpl); 1010 if (!batch) { 1011 cb_fn(cb_arg, -ENOMEM); 1012 return; 1013 } 1014 1015 length = _spdk_bs_page_to_lba(blob->bs, length); 1016 page = offset; 1017 buf = payload; 1018 while (length > 0) { 1019 lba = _spdk_bs_blob_page_to_lba(blob, page); 1020 lba_count = spdk_min(length, 1021 _spdk_bs_page_to_lba(blob->bs, 1022 _spdk_bs_num_pages_to_cluster_boundary(blob, page))); 1023 1024 if (read) { 1025 spdk_bs_batch_read(batch, buf, lba, lba_count); 1026 } else { 1027 spdk_bs_batch_write(batch, buf, lba, lba_count); 1028 } 1029 1030 length -= lba_count; 1031 buf += _spdk_bs_lba_to_byte(blob->bs, lba_count); 1032 page += _spdk_bs_lba_to_page(blob->bs, lba_count); 1033 } 1034 1035 spdk_bs_batch_close(batch); 1036 } 1037 1038 static struct spdk_blob * 1039 _spdk_blob_lookup(struct spdk_blob_store *bs, spdk_blob_id blobid) 1040 { 1041 struct spdk_blob *blob; 1042 1043 TAILQ_FOREACH(blob, &bs->blobs, link) { 1044 if (blob->id == blobid) { 1045 return blob; 1046 } 1047 } 1048 1049 return NULL; 1050 } 1051 1052 static int 1053 _spdk_bs_channel_create(void *io_device, uint32_t priority, void *ctx_buf, void *unique_ctx) 1054 { 1055 struct spdk_blob_store *bs = io_device; 1056 struct spdk_bs_dev *dev = bs->dev; 1057 struct spdk_bs_channel *channel = ctx_buf; 1058 uint32_t i; 1059 1060 channel->req_mem = calloc(spdk_max(bs->max_channel_ops, bs->max_md_ops), 1061 sizeof(struct spdk_bs_request_set)); 1062 if (!channel->req_mem) { 1063 return -1; 1064 } 1065 1066 TAILQ_INIT(&channel->reqs); 1067 1068 for (i = 0; i < spdk_max(bs->max_channel_ops, bs->max_md_ops); i++) { 1069 TAILQ_INSERT_TAIL(&channel->reqs, &channel->req_mem[i], link); 1070 } 1071 1072 channel->bs = bs; 1073 channel->dev = dev; 1074 channel->dev_channel = dev->create_channel(dev); 1075 1076 return 0; 1077 } 1078 1079 static void 1080 _spdk_bs_channel_destroy(void *io_device, void *ctx_buf) 1081 { 1082 struct spdk_bs_channel *channel = ctx_buf; 1083 1084 free(channel->req_mem); 1085 channel->dev->destroy_channel(channel->dev, channel->dev_channel); 1086 } 1087 1088 static void 1089 _spdk_bs_free(struct spdk_blob_store *bs) 1090 { 1091 struct spdk_blob *blob, *blob_tmp; 1092 1093 spdk_bs_unregister_md_thread(bs); 1094 spdk_io_device_unregister(bs); 1095 1096 TAILQ_FOREACH_SAFE(blob, &bs->blobs, link, blob_tmp) { 1097 TAILQ_REMOVE(&bs->blobs, blob, link); 1098 _spdk_blob_free(blob); 1099 } 1100 1101 spdk_bit_array_free(&bs->used_md_pages); 1102 spdk_bit_array_free(&bs->used_clusters); 1103 1104 bs->dev->destroy(bs->dev); 1105 free(bs); 1106 } 1107 1108 void 1109 spdk_bs_opts_init(struct spdk_bs_opts *opts) 1110 { 1111 opts->cluster_sz = SPDK_BLOB_OPTS_CLUSTER_SZ; 1112 opts->num_md_pages = SPDK_BLOB_OPTS_NUM_MD_PAGES; 1113 opts->max_md_ops = SPDK_BLOB_OPTS_MAX_MD_OPS; 1114 opts->max_channel_ops = SPDK_BLOB_OPTS_MAX_CHANNEL_OPS; 1115 } 1116 1117 static struct spdk_blob_store * 1118 _spdk_bs_alloc(struct spdk_bs_dev *dev, struct spdk_bs_opts *opts) 1119 { 1120 struct spdk_blob_store *bs; 1121 1122 bs = calloc(1, sizeof(struct spdk_blob_store)); 1123 if (!bs) { 1124 return NULL; 1125 } 1126 1127 TAILQ_INIT(&bs->blobs); 1128 bs->dev = dev; 1129 1130 /* 1131 * Do not use _spdk_bs_lba_to_cluster() here since blockcnt may not be an 1132 * even multiple of the cluster size. 1133 */ 1134 bs->cluster_sz = opts->cluster_sz; 1135 bs->total_clusters = dev->blockcnt / (bs->cluster_sz / dev->blocklen); 1136 bs->pages_per_cluster = bs->cluster_sz / sizeof(struct spdk_blob_md_page); 1137 bs->num_free_clusters = bs->total_clusters; 1138 bs->used_clusters = spdk_bit_array_create(bs->total_clusters); 1139 if (bs->used_clusters == NULL) { 1140 _spdk_bs_free(bs); 1141 return NULL; 1142 } 1143 1144 bs->max_md_ops = opts->max_md_ops; 1145 bs->max_channel_ops = opts->max_channel_ops; 1146 bs->super_blob = SPDK_BLOBID_INVALID; 1147 1148 /* The metadata is assumed to be at least 1 page */ 1149 bs->used_md_pages = spdk_bit_array_create(1); 1150 1151 spdk_io_device_register(bs, _spdk_bs_channel_create, _spdk_bs_channel_destroy, 1152 sizeof(struct spdk_bs_channel)); 1153 spdk_bs_register_md_thread(bs); 1154 1155 return bs; 1156 } 1157 1158 /* START spdk_bs_load */ 1159 1160 struct spdk_bs_load_ctx { 1161 struct spdk_blob_store *bs; 1162 struct spdk_bs_super_block *super; 1163 1164 struct spdk_bs_md_mask *mask; 1165 }; 1166 1167 static void 1168 _spdk_bs_load_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1169 { 1170 struct spdk_bs_load_ctx *ctx = cb_arg; 1171 uint32_t i, j; 1172 int rc; 1173 1174 /* The type must be correct */ 1175 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_CLUSTERS); 1176 /* The length of the mask (in bits) must not be greater than the length of the buffer (converted to bits) */ 1177 assert(ctx->mask->length <= (ctx->super->used_cluster_mask_len * sizeof( 1178 struct spdk_blob_md_page) * 8)); 1179 /* The length of the mask must be exactly equal to the total number of clusters */ 1180 assert(ctx->mask->length == ctx->bs->total_clusters); 1181 1182 rc = spdk_bit_array_resize(&ctx->bs->used_clusters, ctx->bs->total_clusters); 1183 if (rc < 0) { 1184 spdk_free(ctx->super); 1185 spdk_free(ctx->mask); 1186 _spdk_bs_free(ctx->bs); 1187 free(ctx); 1188 spdk_bs_sequence_finish(seq, -ENOMEM); 1189 return; 1190 } 1191 1192 ctx->bs->num_free_clusters = ctx->bs->total_clusters; 1193 for (i = 0; i < ctx->mask->length / 8; i++) { 1194 uint8_t segment = ctx->mask->mask[i]; 1195 for (j = 0; segment && (j < 8); j++) { 1196 if (segment & 1U) { 1197 spdk_bit_array_set(ctx->bs->used_clusters, (i * 8) + j); 1198 assert(ctx->bs->num_free_clusters > 0); 1199 ctx->bs->num_free_clusters--; 1200 } 1201 segment >>= 1U; 1202 } 1203 } 1204 1205 spdk_free(ctx->super); 1206 spdk_free(ctx->mask); 1207 free(ctx); 1208 1209 spdk_bs_sequence_finish(seq, bserrno); 1210 } 1211 1212 static void 1213 _spdk_bs_load_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1214 { 1215 struct spdk_bs_load_ctx *ctx = cb_arg; 1216 uint64_t lba, lba_count; 1217 uint32_t i, j; 1218 int rc; 1219 1220 /* The type must be correct */ 1221 assert(ctx->mask->type == SPDK_MD_MASK_TYPE_USED_PAGES); 1222 /* The length of the mask (in bits) must not be greater than the length of the buffer (converted to bits) */ 1223 assert(ctx->mask->length <= (ctx->super->used_page_mask_len * sizeof(struct spdk_blob_md_page) * 1224 8)); 1225 /* The length of the mask must be exactly equal to the size (in pages) of the metadata region */ 1226 assert(ctx->mask->length == ctx->super->md_len); 1227 1228 rc = spdk_bit_array_resize(&ctx->bs->used_md_pages, ctx->mask->length); 1229 if (rc < 0) { 1230 spdk_free(ctx->super); 1231 spdk_free(ctx->mask); 1232 _spdk_bs_free(ctx->bs); 1233 free(ctx); 1234 spdk_bs_sequence_finish(seq, -ENOMEM); 1235 return; 1236 } 1237 1238 for (i = 0; i < ctx->mask->length / 8; i++) { 1239 uint8_t segment = ctx->mask->mask[i]; 1240 for (j = 0; segment && (j < 8); j++) { 1241 if (segment & 1U) { 1242 spdk_bit_array_set(ctx->bs->used_md_pages, (i * 8) + j); 1243 } 1244 segment >>= 1U; 1245 } 1246 } 1247 spdk_free(ctx->mask); 1248 1249 /* Read the used clusters mask */ 1250 ctx->mask = spdk_zmalloc(ctx->super->used_cluster_mask_len * sizeof(struct spdk_blob_md_page), 1251 0x1000, NULL); 1252 if (!ctx->mask) { 1253 spdk_free(ctx->super); 1254 _spdk_bs_free(ctx->bs); 1255 free(ctx); 1256 spdk_bs_sequence_finish(seq, -ENOMEM); 1257 return; 1258 } 1259 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_start); 1260 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_len); 1261 spdk_bs_sequence_read(seq, ctx->mask, lba, lba_count, 1262 _spdk_bs_load_used_clusters_cpl, ctx); 1263 } 1264 1265 static void 1266 _spdk_bs_load_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1267 { 1268 struct spdk_bs_load_ctx *ctx = cb_arg; 1269 uint64_t lba, lba_count; 1270 1271 if (ctx->super->version != SPDK_BS_VERSION) { 1272 spdk_free(ctx->super); 1273 _spdk_bs_free(ctx->bs); 1274 free(ctx); 1275 spdk_bs_sequence_finish(seq, -EILSEQ); 1276 return; 1277 } 1278 1279 if (memcmp(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 1280 sizeof(ctx->super->signature)) != 0) { 1281 spdk_free(ctx->super); 1282 _spdk_bs_free(ctx->bs); 1283 free(ctx); 1284 spdk_bs_sequence_finish(seq, -EILSEQ); 1285 return; 1286 } 1287 1288 if (ctx->super->clean != 1) { 1289 /* TODO: ONLY CLEAN SHUTDOWN IS CURRENTLY SUPPORTED. 1290 * All of the necessary data to recover is available 1291 * on disk - the code just has not been written yet. 1292 */ 1293 assert(false); 1294 spdk_free(ctx->super); 1295 _spdk_bs_free(ctx->bs); 1296 free(ctx); 1297 spdk_bs_sequence_finish(seq, -EILSEQ); 1298 return; 1299 } 1300 ctx->super->clean = 0; 1301 1302 /* Parse the super block */ 1303 ctx->bs->cluster_sz = ctx->super->cluster_size; 1304 ctx->bs->total_clusters = ctx->bs->dev->blockcnt / (ctx->bs->cluster_sz / ctx->bs->dev->blocklen); 1305 ctx->bs->pages_per_cluster = ctx->bs->cluster_sz / sizeof(struct spdk_blob_md_page); 1306 ctx->bs->md_start = ctx->super->md_start; 1307 ctx->bs->md_len = ctx->super->md_len; 1308 1309 /* Read the used pages mask */ 1310 ctx->mask = spdk_zmalloc(ctx->super->used_page_mask_len * sizeof(struct spdk_blob_md_page), 0x1000, 1311 NULL); 1312 if (!ctx->mask) { 1313 spdk_free(ctx->super); 1314 _spdk_bs_free(ctx->bs); 1315 free(ctx); 1316 spdk_bs_sequence_finish(seq, -ENOMEM); 1317 return; 1318 } 1319 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_start); 1320 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_len); 1321 spdk_bs_sequence_read(seq, ctx->mask, lba, lba_count, 1322 _spdk_bs_load_used_pages_cpl, ctx); 1323 } 1324 1325 void 1326 spdk_bs_load(struct spdk_bs_dev *dev, 1327 spdk_bs_op_with_handle_complete cb_fn, void *cb_arg) 1328 { 1329 struct spdk_blob_store *bs; 1330 struct spdk_bs_cpl cpl; 1331 spdk_bs_sequence_t *seq; 1332 struct spdk_bs_load_ctx *ctx; 1333 struct spdk_bs_opts opts = {}; 1334 1335 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Loading blobstore from dev %p\n", dev); 1336 1337 spdk_bs_opts_init(&opts); 1338 1339 bs = _spdk_bs_alloc(dev, &opts); 1340 if (!bs) { 1341 cb_fn(cb_arg, NULL, -ENOMEM); 1342 return; 1343 } 1344 1345 ctx = calloc(1, sizeof(*ctx)); 1346 if (!ctx) { 1347 _spdk_bs_free(bs); 1348 cb_fn(cb_arg, NULL, -ENOMEM); 1349 return; 1350 } 1351 1352 ctx->bs = bs; 1353 1354 /* Allocate memory for the super block */ 1355 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL); 1356 if (!ctx->super) { 1357 free(ctx); 1358 _spdk_bs_free(bs); 1359 return; 1360 } 1361 1362 cpl.type = SPDK_BS_CPL_TYPE_BS_HANDLE; 1363 cpl.u.bs_handle.cb_fn = cb_fn; 1364 cpl.u.bs_handle.cb_arg = cb_arg; 1365 cpl.u.bs_handle.bs = bs; 1366 1367 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 1368 if (!seq) { 1369 spdk_free(ctx->super); 1370 free(ctx); 1371 _spdk_bs_free(bs); 1372 cb_fn(cb_arg, NULL, -ENOMEM); 1373 return; 1374 } 1375 1376 /* Read the super block */ 1377 spdk_bs_sequence_read(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 1378 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 1379 _spdk_bs_load_super_cpl, ctx); 1380 } 1381 1382 /* END spdk_bs_load */ 1383 1384 /* START spdk_bs_init */ 1385 1386 struct spdk_bs_init_ctx { 1387 struct spdk_blob_store *bs; 1388 struct spdk_bs_super_block *super; 1389 }; 1390 1391 static void 1392 _spdk_bs_init_persist_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1393 { 1394 struct spdk_bs_init_ctx *ctx = cb_arg; 1395 1396 spdk_free(ctx->super); 1397 free(ctx); 1398 1399 spdk_bs_sequence_finish(seq, bserrno); 1400 } 1401 1402 static void 1403 _spdk_bs_init_trim_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1404 { 1405 struct spdk_bs_init_ctx *ctx = cb_arg; 1406 1407 /* Write super block */ 1408 spdk_bs_sequence_write(seq, ctx->super, _spdk_bs_page_to_lba(ctx->bs, 0), 1409 _spdk_bs_byte_to_lba(ctx->bs, sizeof(*ctx->super)), 1410 _spdk_bs_init_persist_super_cpl, ctx); 1411 } 1412 1413 void 1414 spdk_bs_init(struct spdk_bs_dev *dev, struct spdk_bs_opts *o, 1415 spdk_bs_op_with_handle_complete cb_fn, void *cb_arg) 1416 { 1417 struct spdk_bs_init_ctx *ctx; 1418 struct spdk_blob_store *bs; 1419 struct spdk_bs_cpl cpl; 1420 spdk_bs_sequence_t *seq; 1421 uint64_t num_md_pages; 1422 uint32_t i; 1423 struct spdk_bs_opts opts = {}; 1424 int rc; 1425 1426 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Initializing blobstore on dev %p\n", dev); 1427 1428 if (o) { 1429 opts = *o; 1430 } else { 1431 spdk_bs_opts_init(&opts); 1432 } 1433 1434 bs = _spdk_bs_alloc(dev, &opts); 1435 if (!bs) { 1436 cb_fn(cb_arg, NULL, -ENOMEM); 1437 return; 1438 } 1439 1440 if (opts.num_md_pages == UINT32_MAX) { 1441 /* By default, allocate 1 page per cluster. 1442 * Technically, this over-allocates metadata 1443 * because more metadata will reduce the number 1444 * of usable clusters. This can be addressed with 1445 * more complex math in the future. 1446 */ 1447 bs->md_len = bs->total_clusters; 1448 } else { 1449 bs->md_len = opts.num_md_pages; 1450 } 1451 1452 rc = spdk_bit_array_resize(&bs->used_md_pages, bs->md_len); 1453 if (rc < 0) { 1454 _spdk_bs_free(bs); 1455 cb_fn(cb_arg, NULL, -ENOMEM); 1456 return; 1457 } 1458 1459 ctx = calloc(1, sizeof(*ctx)); 1460 if (!ctx) { 1461 _spdk_bs_free(bs); 1462 cb_fn(cb_arg, NULL, -ENOMEM); 1463 return; 1464 } 1465 1466 ctx->bs = bs; 1467 1468 /* Allocate memory for the super block */ 1469 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL); 1470 if (!ctx->super) { 1471 free(ctx); 1472 _spdk_bs_free(bs); 1473 return; 1474 } 1475 memcpy(ctx->super->signature, SPDK_BS_SUPER_BLOCK_SIG, 1476 sizeof(ctx->super->signature)); 1477 ctx->super->version = SPDK_BS_VERSION; 1478 ctx->super->length = sizeof(*ctx->super); 1479 ctx->super->super_blob = bs->super_blob; 1480 ctx->super->clean = 0; 1481 ctx->super->cluster_size = bs->cluster_sz; 1482 1483 /* Calculate how many pages the metadata consumes at the front 1484 * of the disk. 1485 */ 1486 1487 /* The super block uses 1 page */ 1488 num_md_pages = 1; 1489 1490 /* The used_md_pages mask requires 1 bit per metadata page, rounded 1491 * up to the nearest page, plus a header. 1492 */ 1493 ctx->super->used_page_mask_start = num_md_pages; 1494 ctx->super->used_page_mask_len = divide_round_up(sizeof(struct spdk_bs_md_mask) + 1495 divide_round_up(bs->md_len, 8), 1496 sizeof(struct spdk_blob_md_page)); 1497 num_md_pages += ctx->super->used_page_mask_len; 1498 1499 /* The used_clusters mask requires 1 bit per cluster, rounded 1500 * up to the nearest page, plus a header. 1501 */ 1502 ctx->super->used_cluster_mask_start = num_md_pages; 1503 ctx->super->used_cluster_mask_len = divide_round_up(sizeof(struct spdk_bs_md_mask) + 1504 divide_round_up(bs->total_clusters, 8), 1505 sizeof(struct spdk_blob_md_page)); 1506 num_md_pages += ctx->super->used_cluster_mask_len; 1507 1508 /* The metadata region size was chosen above */ 1509 ctx->super->md_start = bs->md_start = num_md_pages; 1510 ctx->super->md_len = bs->md_len; 1511 num_md_pages += bs->md_len; 1512 1513 /* Claim all of the clusters used by the metadata */ 1514 for (i = 0; i < divide_round_up(num_md_pages, bs->pages_per_cluster); i++) { 1515 _spdk_bs_claim_cluster(bs, i); 1516 } 1517 1518 cpl.type = SPDK_BS_CPL_TYPE_BS_HANDLE; 1519 cpl.u.bs_handle.cb_fn = cb_fn; 1520 cpl.u.bs_handle.cb_arg = cb_arg; 1521 cpl.u.bs_handle.bs = bs; 1522 1523 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 1524 if (!seq) { 1525 spdk_free(ctx->super); 1526 free(ctx); 1527 _spdk_bs_free(bs); 1528 cb_fn(cb_arg, NULL, -ENOMEM); 1529 return; 1530 } 1531 1532 /* TRIM the entire device */ 1533 spdk_bs_sequence_unmap(seq, 0, bs->dev->blockcnt, _spdk_bs_init_trim_cpl, ctx); 1534 } 1535 1536 /* END spdk_bs_init */ 1537 1538 /* START spdk_bs_unload */ 1539 1540 struct spdk_bs_unload_ctx { 1541 struct spdk_blob_store *bs; 1542 struct spdk_bs_super_block *super; 1543 1544 struct spdk_bs_md_mask *mask; 1545 }; 1546 1547 static void 1548 _spdk_bs_unload_write_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1549 { 1550 struct spdk_bs_unload_ctx *ctx = cb_arg; 1551 1552 spdk_free(ctx->super); 1553 1554 spdk_bs_sequence_finish(seq, bserrno); 1555 1556 _spdk_bs_free(ctx->bs); 1557 free(ctx); 1558 } 1559 1560 static void 1561 _spdk_bs_unload_write_used_clusters_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1562 { 1563 struct spdk_bs_unload_ctx *ctx = cb_arg; 1564 1565 spdk_free(ctx->mask); 1566 1567 /* Update the values in the super block */ 1568 ctx->super->super_blob = ctx->bs->super_blob; 1569 ctx->super->clean = 1; 1570 1571 spdk_bs_sequence_write(seq, ctx->super, _spdk_bs_page_to_lba(ctx->bs, 0), 1572 _spdk_bs_byte_to_lba(ctx->bs, sizeof(*ctx->super)), 1573 _spdk_bs_unload_write_super_cpl, ctx); 1574 } 1575 1576 static void 1577 _spdk_bs_unload_write_used_pages_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1578 { 1579 struct spdk_bs_unload_ctx *ctx = cb_arg; 1580 uint32_t i; 1581 uint64_t lba, lba_count; 1582 1583 spdk_free(ctx->mask); 1584 1585 /* Write out the used clusters mask */ 1586 ctx->mask = spdk_zmalloc(ctx->super->used_cluster_mask_len * sizeof(struct spdk_blob_md_page), 1587 0x1000, NULL); 1588 if (!ctx->mask) { 1589 spdk_free(ctx->super); 1590 free(ctx); 1591 spdk_bs_sequence_finish(seq, -ENOMEM); 1592 return; 1593 } 1594 1595 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_CLUSTERS; 1596 ctx->mask->length = ctx->bs->total_clusters; 1597 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_clusters)); 1598 1599 i = 0; 1600 while (true) { 1601 i = spdk_bit_array_find_first_set(ctx->bs->used_clusters, i); 1602 if (i > ctx->mask->length) { 1603 break; 1604 } 1605 ctx->mask->mask[i / 8] |= 1U << (i % 8); 1606 i++; 1607 } 1608 1609 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_start); 1610 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_cluster_mask_len); 1611 spdk_bs_sequence_write(seq, ctx->mask, lba, lba_count, 1612 _spdk_bs_unload_write_used_clusters_cpl, ctx); 1613 } 1614 1615 static void 1616 _spdk_bs_unload_read_super_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1617 { 1618 struct spdk_bs_unload_ctx *ctx = cb_arg; 1619 uint32_t i; 1620 uint64_t lba, lba_count; 1621 1622 /* Write out the used page mask */ 1623 ctx->mask = spdk_zmalloc(ctx->super->used_page_mask_len * sizeof(struct spdk_blob_md_page), 1624 0x1000, NULL); 1625 if (!ctx->mask) { 1626 spdk_free(ctx->super); 1627 free(ctx); 1628 spdk_bs_sequence_finish(seq, -ENOMEM); 1629 return; 1630 } 1631 1632 ctx->mask->type = SPDK_MD_MASK_TYPE_USED_PAGES; 1633 ctx->mask->length = ctx->super->md_len; 1634 assert(ctx->mask->length == spdk_bit_array_capacity(ctx->bs->used_md_pages)); 1635 1636 i = 0; 1637 while (true) { 1638 i = spdk_bit_array_find_first_set(ctx->bs->used_md_pages, i); 1639 if (i > ctx->mask->length) { 1640 break; 1641 } 1642 ctx->mask->mask[i / 8] |= 1U << (i % 8); 1643 i++; 1644 } 1645 1646 lba = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_start); 1647 lba_count = _spdk_bs_page_to_lba(ctx->bs, ctx->super->used_page_mask_len); 1648 spdk_bs_sequence_write(seq, ctx->mask, lba, lba_count, 1649 _spdk_bs_unload_write_used_pages_cpl, ctx); 1650 } 1651 1652 void 1653 spdk_bs_unload(struct spdk_blob_store *bs, spdk_bs_op_complete cb_fn, void *cb_arg) 1654 { 1655 struct spdk_bs_cpl cpl; 1656 spdk_bs_sequence_t *seq; 1657 struct spdk_bs_unload_ctx *ctx; 1658 1659 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Syncing blobstore\n"); 1660 1661 ctx = calloc(1, sizeof(*ctx)); 1662 if (!ctx) { 1663 cb_fn(cb_arg, -ENOMEM); 1664 return; 1665 } 1666 1667 ctx->bs = bs; 1668 1669 ctx->super = spdk_zmalloc(sizeof(*ctx->super), 0x1000, NULL); 1670 if (!ctx->super) { 1671 free(ctx); 1672 cb_fn(cb_arg, -ENOMEM); 1673 return; 1674 } 1675 1676 cpl.type = SPDK_BS_CPL_TYPE_BS_BASIC; 1677 cpl.u.bs_basic.cb_fn = cb_fn; 1678 cpl.u.bs_basic.cb_arg = cb_arg; 1679 1680 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 1681 if (!seq) { 1682 spdk_free(ctx->super); 1683 free(ctx); 1684 cb_fn(cb_arg, -ENOMEM); 1685 return; 1686 } 1687 1688 assert(TAILQ_EMPTY(&bs->blobs)); 1689 1690 /* Read super block */ 1691 spdk_bs_sequence_read(seq, ctx->super, _spdk_bs_page_to_lba(bs, 0), 1692 _spdk_bs_byte_to_lba(bs, sizeof(*ctx->super)), 1693 _spdk_bs_unload_read_super_cpl, ctx); 1694 } 1695 1696 /* END spdk_bs_unload */ 1697 1698 void 1699 spdk_bs_set_super(struct spdk_blob_store *bs, spdk_blob_id blobid, 1700 spdk_bs_op_complete cb_fn, void *cb_arg) 1701 { 1702 bs->super_blob = blobid; 1703 cb_fn(cb_arg, 0); 1704 } 1705 1706 void 1707 spdk_bs_get_super(struct spdk_blob_store *bs, 1708 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 1709 { 1710 if (bs->super_blob == SPDK_BLOBID_INVALID) { 1711 cb_fn(cb_arg, SPDK_BLOBID_INVALID, -ENOENT); 1712 } else { 1713 cb_fn(cb_arg, bs->super_blob, 0); 1714 } 1715 } 1716 1717 uint64_t 1718 spdk_bs_get_cluster_size(struct spdk_blob_store *bs) 1719 { 1720 return bs->cluster_sz; 1721 } 1722 1723 uint64_t 1724 spdk_bs_get_page_size(struct spdk_blob_store *bs) 1725 { 1726 return sizeof(struct spdk_blob_md_page); 1727 } 1728 1729 uint64_t 1730 spdk_bs_free_cluster_count(struct spdk_blob_store *bs) 1731 { 1732 return bs->num_free_clusters; 1733 } 1734 1735 int spdk_bs_register_md_thread(struct spdk_blob_store *bs) 1736 { 1737 bs->md_channel = spdk_get_io_channel(bs, SPDK_IO_PRIORITY_DEFAULT, true, 1738 NULL); 1739 1740 return 0; 1741 } 1742 1743 int spdk_bs_unregister_md_thread(struct spdk_blob_store *bs) 1744 { 1745 spdk_put_io_channel(bs->md_channel); 1746 1747 return 0; 1748 } 1749 1750 spdk_blob_id spdk_blob_get_id(struct spdk_blob *blob) 1751 { 1752 assert(blob != NULL); 1753 1754 return blob->id; 1755 } 1756 1757 uint64_t spdk_blob_get_num_pages(struct spdk_blob *blob) 1758 { 1759 assert(blob != NULL); 1760 1761 return _spdk_bs_cluster_to_page(blob->bs, blob->active.num_clusters); 1762 } 1763 1764 uint64_t spdk_blob_get_num_clusters(struct spdk_blob *blob) 1765 { 1766 assert(blob != NULL); 1767 1768 return blob->active.num_clusters; 1769 } 1770 1771 /* START spdk_bs_md_create_blob */ 1772 1773 static void 1774 _spdk_bs_md_create_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1775 { 1776 struct spdk_blob *blob = cb_arg; 1777 1778 _spdk_blob_free(blob); 1779 1780 spdk_bs_sequence_finish(seq, bserrno); 1781 } 1782 1783 void spdk_bs_md_create_blob(struct spdk_blob_store *bs, 1784 spdk_blob_op_with_id_complete cb_fn, void *cb_arg) 1785 { 1786 struct spdk_blob *blob; 1787 uint32_t page_idx; 1788 struct spdk_bs_cpl cpl; 1789 spdk_bs_sequence_t *seq; 1790 spdk_blob_id id; 1791 1792 page_idx = spdk_bit_array_find_first_clear(bs->used_md_pages, 0); 1793 if (page_idx >= spdk_bit_array_capacity(bs->used_md_pages)) { 1794 cb_fn(cb_arg, 0, -ENOMEM); 1795 return; 1796 } 1797 spdk_bit_array_set(bs->used_md_pages, page_idx); 1798 1799 /* The blob id is a 64 bit number. The lower 32 bits are the page_idx. The upper 1800 * 32 bits are not currently used. Stick a 1 there just to catch bugs where the 1801 * code assumes blob id == page_idx. 1802 */ 1803 id = (1ULL << 32) | page_idx; 1804 1805 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Creating blob with id %lu at page %u\n", id, page_idx); 1806 1807 blob = _spdk_blob_alloc(bs, id); 1808 if (!blob) { 1809 cb_fn(cb_arg, 0, -ENOMEM); 1810 return; 1811 } 1812 1813 cpl.type = SPDK_BS_CPL_TYPE_BLOBID; 1814 cpl.u.blobid.cb_fn = cb_fn; 1815 cpl.u.blobid.cb_arg = cb_arg; 1816 cpl.u.blobid.blobid = blob->id; 1817 1818 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 1819 if (!seq) { 1820 _spdk_blob_free(blob); 1821 cb_fn(cb_arg, 0, -ENOMEM); 1822 return; 1823 } 1824 1825 _spdk_blob_persist(seq, blob, _spdk_bs_md_create_blob_cpl, blob); 1826 } 1827 1828 /* END spdk_bs_md_create_blob */ 1829 1830 /* START spdk_bs_md_resize_blob */ 1831 int 1832 spdk_bs_md_resize_blob(struct spdk_blob *blob, uint64_t sz) 1833 { 1834 int rc; 1835 1836 assert(blob != NULL); 1837 1838 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Resizing blob %lu to %lu clusters\n", blob->id, sz); 1839 1840 if (sz == blob->active.num_clusters) { 1841 return 0; 1842 } 1843 1844 rc = _spdk_resize_blob(blob, sz); 1845 if (rc < 0) { 1846 return rc; 1847 } 1848 1849 return 0; 1850 } 1851 1852 /* END spdk_bs_md_resize_blob */ 1853 1854 1855 /* START spdk_bs_md_delete_blob */ 1856 1857 static void 1858 _spdk_bs_md_delete_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1859 { 1860 struct spdk_blob *blob = cb_arg; 1861 1862 _spdk_blob_free(blob); 1863 1864 spdk_bs_sequence_finish(seq, bserrno); 1865 } 1866 1867 static void 1868 _spdk_bs_md_delete_open_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1869 { 1870 struct spdk_blob *blob = cb_arg; 1871 1872 blob->state = SPDK_BLOB_STATE_DIRTY; 1873 blob->active.num_pages = 0; 1874 _spdk_resize_blob(blob, 0); 1875 1876 _spdk_blob_persist(seq, blob, _spdk_bs_md_delete_blob_cpl, blob); 1877 } 1878 1879 void 1880 spdk_bs_md_delete_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 1881 spdk_blob_op_complete cb_fn, void *cb_arg) 1882 { 1883 struct spdk_blob *blob; 1884 struct spdk_bs_cpl cpl; 1885 spdk_bs_sequence_t *seq; 1886 1887 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Deleting blob %lu\n", blobid); 1888 1889 blob = _spdk_blob_lookup(bs, blobid); 1890 if (blob) { 1891 assert(blob->open_ref > 0); 1892 cb_fn(cb_arg, -EINVAL); 1893 return; 1894 } 1895 1896 blob = _spdk_blob_alloc(bs, blobid); 1897 if (!blob) { 1898 cb_fn(cb_arg, -ENOMEM); 1899 return; 1900 } 1901 1902 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 1903 cpl.u.blob_basic.cb_fn = cb_fn; 1904 cpl.u.blob_basic.cb_arg = cb_arg; 1905 1906 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 1907 if (!seq) { 1908 _spdk_blob_free(blob); 1909 cb_fn(cb_arg, -ENOMEM); 1910 return; 1911 } 1912 1913 _spdk_blob_load(seq, blob, _spdk_bs_md_delete_open_cpl, blob); 1914 } 1915 1916 /* END spdk_bs_md_delete_blob */ 1917 1918 /* START spdk_bs_md_open_blob */ 1919 1920 static void 1921 _spdk_bs_md_open_blob_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1922 { 1923 struct spdk_blob *blob = cb_arg; 1924 1925 blob->open_ref++; 1926 1927 TAILQ_INSERT_HEAD(&blob->bs->blobs, blob, link); 1928 1929 spdk_bs_sequence_finish(seq, bserrno); 1930 } 1931 1932 void spdk_bs_md_open_blob(struct spdk_blob_store *bs, spdk_blob_id blobid, 1933 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 1934 { 1935 struct spdk_blob *blob; 1936 struct spdk_bs_cpl cpl; 1937 spdk_bs_sequence_t *seq; 1938 uint32_t page_num; 1939 1940 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Opening blob %lu\n", blobid); 1941 1942 blob = _spdk_blob_lookup(bs, blobid); 1943 if (blob) { 1944 blob->open_ref++; 1945 cb_fn(cb_arg, blob, 0); 1946 return; 1947 } 1948 1949 page_num = _spdk_bs_blobid_to_page(blobid); 1950 if (spdk_bit_array_get(bs->used_md_pages, page_num) == false) { 1951 /* Invalid blobid */ 1952 cb_fn(cb_arg, NULL, -ENOENT); 1953 return; 1954 } 1955 1956 blob = _spdk_blob_alloc(bs, blobid); 1957 if (!blob) { 1958 cb_fn(cb_arg, NULL, -ENOMEM); 1959 return; 1960 } 1961 1962 cpl.type = SPDK_BS_CPL_TYPE_BLOB_HANDLE; 1963 cpl.u.blob_handle.cb_fn = cb_fn; 1964 cpl.u.blob_handle.cb_arg = cb_arg; 1965 cpl.u.blob_handle.blob = blob; 1966 1967 seq = spdk_bs_sequence_start(bs->md_channel, &cpl); 1968 if (!seq) { 1969 _spdk_blob_free(blob); 1970 cb_fn(cb_arg, NULL, -ENOMEM); 1971 return; 1972 } 1973 1974 _spdk_blob_load(seq, blob, _spdk_bs_md_open_blob_cpl, blob); 1975 } 1976 1977 /* START spdk_bs_md_sync_blob */ 1978 static void 1979 _spdk_blob_sync_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 1980 { 1981 spdk_bs_sequence_finish(seq, bserrno); 1982 } 1983 1984 void spdk_bs_md_sync_blob(struct spdk_blob *blob, 1985 spdk_blob_op_complete cb_fn, void *cb_arg) 1986 { 1987 struct spdk_bs_cpl cpl; 1988 spdk_bs_sequence_t *seq; 1989 1990 assert(blob != NULL); 1991 1992 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Syncing blob %lu\n", blob->id); 1993 1994 assert(blob->state != SPDK_BLOB_STATE_LOADING && 1995 blob->state != SPDK_BLOB_STATE_SYNCING); 1996 1997 if (blob->state == SPDK_BLOB_STATE_CLEAN) { 1998 cb_fn(cb_arg, 0); 1999 return; 2000 } 2001 2002 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 2003 cpl.u.blob_basic.cb_fn = cb_fn; 2004 cpl.u.blob_basic.cb_arg = cb_arg; 2005 2006 seq = spdk_bs_sequence_start(blob->bs->md_channel, &cpl); 2007 if (!seq) { 2008 cb_fn(cb_arg, -ENOMEM); 2009 return; 2010 } 2011 2012 _spdk_blob_persist(seq, blob, _spdk_blob_sync_cpl, blob); 2013 } 2014 2015 /* END spdk_bs_md_sync_blob */ 2016 2017 /* START spdk_bs_md_close_blob */ 2018 2019 static void 2020 _spdk_blob_close_cpl(spdk_bs_sequence_t *seq, void *cb_arg, int bserrno) 2021 { 2022 struct spdk_blob **blob = cb_arg; 2023 2024 if ((*blob)->open_ref == 0) { 2025 TAILQ_REMOVE(&(*blob)->bs->blobs, (*blob), link); 2026 _spdk_blob_free((*blob)); 2027 } 2028 2029 *blob = NULL; 2030 2031 spdk_bs_sequence_finish(seq, bserrno); 2032 } 2033 2034 void spdk_bs_md_close_blob(struct spdk_blob **b, 2035 spdk_blob_op_complete cb_fn, void *cb_arg) 2036 { 2037 struct spdk_bs_cpl cpl; 2038 struct spdk_blob *blob; 2039 spdk_bs_sequence_t *seq; 2040 2041 assert(b != NULL); 2042 blob = *b; 2043 assert(blob != NULL); 2044 2045 SPDK_TRACELOG(SPDK_TRACE_BLOB, "Closing blob %lu\n", blob->id); 2046 2047 assert(blob->state != SPDK_BLOB_STATE_LOADING && 2048 blob->state != SPDK_BLOB_STATE_SYNCING); 2049 2050 if (blob->open_ref == 0) { 2051 cb_fn(cb_arg, -EBADF); 2052 return; 2053 } 2054 2055 blob->open_ref--; 2056 2057 cpl.type = SPDK_BS_CPL_TYPE_BLOB_BASIC; 2058 cpl.u.blob_basic.cb_fn = cb_fn; 2059 cpl.u.blob_basic.cb_arg = cb_arg; 2060 2061 seq = spdk_bs_sequence_start(blob->bs->md_channel, &cpl); 2062 if (!seq) { 2063 cb_fn(cb_arg, -ENOMEM); 2064 return; 2065 } 2066 2067 if (blob->state == SPDK_BLOB_STATE_CLEAN) { 2068 _spdk_blob_close_cpl(seq, b, 0); 2069 return; 2070 } 2071 2072 /* Sync metadata */ 2073 _spdk_blob_persist(seq, blob, _spdk_blob_close_cpl, b); 2074 } 2075 2076 /* END spdk_bs_md_close_blob */ 2077 2078 struct spdk_io_channel *spdk_bs_alloc_io_channel(struct spdk_blob_store *bs, 2079 uint32_t priority) 2080 { 2081 return spdk_get_io_channel(bs, priority, true, NULL); 2082 } 2083 2084 void spdk_bs_free_io_channel(struct spdk_io_channel *channel) 2085 { 2086 spdk_put_io_channel(channel); 2087 } 2088 2089 void spdk_bs_io_flush_channel(struct spdk_io_channel *channel, 2090 spdk_blob_op_complete cb_fn, void *cb_arg) 2091 { 2092 /* Flush is synchronous right now */ 2093 cb_fn(cb_arg, 0); 2094 } 2095 2096 void spdk_bs_io_write_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 2097 void *payload, uint64_t offset, uint64_t length, 2098 spdk_blob_op_complete cb_fn, void *cb_arg) 2099 { 2100 _spdk_blob_request_submit_rw(blob, channel, payload, offset, length, cb_fn, cb_arg, false); 2101 } 2102 2103 void spdk_bs_io_read_blob(struct spdk_blob *blob, struct spdk_io_channel *channel, 2104 void *payload, uint64_t offset, uint64_t length, 2105 spdk_blob_op_complete cb_fn, void *cb_arg) 2106 { 2107 _spdk_blob_request_submit_rw(blob, channel, payload, offset, length, cb_fn, cb_arg, true); 2108 } 2109 2110 struct spdk_bs_iter_ctx { 2111 int64_t page_num; 2112 struct spdk_blob_store *bs; 2113 2114 spdk_blob_op_with_handle_complete cb_fn; 2115 void *cb_arg; 2116 }; 2117 2118 static void 2119 _spdk_bs_iter_cpl(void *cb_arg, struct spdk_blob *blob, int bserrno) 2120 { 2121 struct spdk_bs_iter_ctx *ctx = cb_arg; 2122 struct spdk_blob_store *bs = ctx->bs; 2123 spdk_blob_id id; 2124 2125 if (bserrno == 0) { 2126 ctx->cb_fn(ctx->cb_arg, blob, bserrno); 2127 free(ctx); 2128 return; 2129 } 2130 2131 ctx->page_num++; 2132 ctx->page_num = spdk_bit_array_find_first_set(bs->used_md_pages, ctx->page_num); 2133 if (ctx->page_num >= spdk_bit_array_capacity(bs->used_md_pages)) { 2134 ctx->cb_fn(ctx->cb_arg, NULL, -ENOENT); 2135 free(ctx); 2136 return; 2137 } 2138 2139 id = (1ULL << 32) | ctx->page_num; 2140 2141 blob = _spdk_blob_lookup(bs, id); 2142 if (blob) { 2143 blob->open_ref++; 2144 ctx->cb_fn(ctx->cb_arg, blob, 0); 2145 free(ctx); 2146 return; 2147 } 2148 2149 spdk_bs_md_open_blob(bs, id, _spdk_bs_iter_cpl, ctx); 2150 } 2151 2152 void 2153 spdk_bs_md_iter_first(struct spdk_blob_store *bs, 2154 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 2155 { 2156 struct spdk_bs_iter_ctx *ctx; 2157 2158 ctx = calloc(1, sizeof(*ctx)); 2159 if (!ctx) { 2160 cb_fn(cb_arg, NULL, -ENOMEM); 2161 return; 2162 } 2163 2164 ctx->page_num = -1; 2165 ctx->bs = bs; 2166 ctx->cb_fn = cb_fn; 2167 ctx->cb_arg = cb_arg; 2168 2169 _spdk_bs_iter_cpl(ctx, NULL, -1); 2170 } 2171 2172 static void 2173 _spdk_bs_iter_close_cpl(void *cb_arg, int bserrno) 2174 { 2175 struct spdk_bs_iter_ctx *ctx = cb_arg; 2176 2177 _spdk_bs_iter_cpl(ctx, NULL, -1); 2178 } 2179 2180 void 2181 spdk_bs_md_iter_next(struct spdk_blob_store *bs, struct spdk_blob **b, 2182 spdk_blob_op_with_handle_complete cb_fn, void *cb_arg) 2183 { 2184 struct spdk_bs_iter_ctx *ctx; 2185 struct spdk_blob *blob; 2186 2187 assert(b != NULL); 2188 blob = *b; 2189 assert(blob != NULL); 2190 2191 ctx = calloc(1, sizeof(*ctx)); 2192 if (!ctx) { 2193 cb_fn(cb_arg, NULL, -ENOMEM); 2194 return; 2195 } 2196 2197 ctx->page_num = _spdk_bs_blobid_to_page(blob->id); 2198 ctx->bs = bs; 2199 ctx->cb_fn = cb_fn; 2200 ctx->cb_arg = cb_arg; 2201 2202 /* Close the existing blob */ 2203 spdk_bs_md_close_blob(b, _spdk_bs_iter_close_cpl, ctx); 2204 } 2205 2206 int 2207 spdk_blob_md_set_xattr(struct spdk_blob *blob, const char *name, const void *value, 2208 uint16_t value_len) 2209 { 2210 struct spdk_xattr *xattr; 2211 2212 assert(blob != NULL); 2213 2214 assert(blob->state != SPDK_BLOB_STATE_LOADING && 2215 blob->state != SPDK_BLOB_STATE_SYNCING); 2216 2217 TAILQ_FOREACH(xattr, &blob->xattrs, link) { 2218 if (!strcmp(name, xattr->name)) { 2219 free(xattr->value); 2220 xattr->value_len = value_len; 2221 xattr->value = malloc(value_len); 2222 memcpy(xattr->value, value, value_len); 2223 2224 blob->state = SPDK_BLOB_STATE_DIRTY; 2225 2226 return 0; 2227 } 2228 } 2229 2230 xattr = calloc(1, sizeof(*xattr)); 2231 if (!xattr) { 2232 return -1; 2233 } 2234 xattr->name = strdup(name); 2235 xattr->value_len = value_len; 2236 xattr->value = malloc(value_len); 2237 memcpy(xattr->value, value, value_len); 2238 TAILQ_INSERT_TAIL(&blob->xattrs, xattr, link); 2239 2240 blob->state = SPDK_BLOB_STATE_DIRTY; 2241 2242 return 0; 2243 } 2244 2245 int 2246 spdk_blob_md_remove_xattr(struct spdk_blob *blob, const char *name) 2247 { 2248 struct spdk_xattr *xattr; 2249 2250 assert(blob != NULL); 2251 2252 assert(blob->state != SPDK_BLOB_STATE_LOADING && 2253 blob->state != SPDK_BLOB_STATE_SYNCING); 2254 2255 TAILQ_FOREACH(xattr, &blob->xattrs, link) { 2256 if (!strcmp(name, xattr->name)) { 2257 TAILQ_REMOVE(&blob->xattrs, xattr, link); 2258 free(xattr->value); 2259 free(xattr->name); 2260 free(xattr); 2261 2262 blob->state = SPDK_BLOB_STATE_DIRTY; 2263 2264 return 0; 2265 } 2266 } 2267 2268 return -ENOENT; 2269 } 2270 2271 int 2272 spdk_bs_md_get_xattr_value(struct spdk_blob *blob, const char *name, 2273 const void **value, size_t *value_len) 2274 { 2275 struct spdk_xattr *xattr; 2276 2277 TAILQ_FOREACH(xattr, &blob->xattrs, link) { 2278 if (!strcmp(name, xattr->name)) { 2279 *value = xattr->value; 2280 *value_len = xattr->value_len; 2281 return 0; 2282 } 2283 } 2284 2285 return -ENOENT; 2286 } 2287 2288 struct spdk_xattr_names { 2289 uint32_t count; 2290 const char *names[0]; 2291 }; 2292 2293 int 2294 spdk_bs_md_get_xattr_names(struct spdk_blob *blob, 2295 struct spdk_xattr_names **names) 2296 { 2297 struct spdk_xattr *xattr; 2298 int count = 0; 2299 2300 TAILQ_FOREACH(xattr, &blob->xattrs, link) { 2301 count++; 2302 } 2303 2304 *names = calloc(1, sizeof(struct spdk_xattr_names) + count * sizeof(char *)); 2305 if (*names == NULL) { 2306 return -ENOMEM; 2307 } 2308 2309 TAILQ_FOREACH(xattr, &blob->xattrs, link) { 2310 (*names)->names[(*names)->count++] = xattr->name; 2311 } 2312 2313 return 0; 2314 } 2315 2316 uint32_t 2317 spdk_xattr_names_get_count(struct spdk_xattr_names *names) 2318 { 2319 assert(names != NULL); 2320 2321 return names->count; 2322 } 2323 2324 const char * 2325 spdk_xattr_names_get_name(struct spdk_xattr_names *names, uint32_t index) 2326 { 2327 if (index >= names->count) { 2328 return NULL; 2329 } 2330 2331 return names->names[index]; 2332 } 2333 2334 void 2335 spdk_xattr_names_free(struct spdk_xattr_names *names) 2336 { 2337 free(names); 2338 } 2339 2340 SPDK_LOG_REGISTER_TRACE_FLAG("blob", SPDK_TRACE_BLOB); 2341