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/blobfs.h" 37 #include "spdk/conf.h" 38 #include "blobfs_internal.h" 39 40 #include "spdk/queue.h" 41 #include "spdk/io_channel.h" 42 #include "spdk/assert.h" 43 #include "spdk/env.h" 44 #include "spdk/util.h" 45 #include "spdk_internal/log.h" 46 47 #define BLOBFS_TRACE(file, str, args...) \ 48 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "file=%s " str, file->name, ##args) 49 50 #define BLOBFS_TRACE_RW(file, str, args...) \ 51 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS_RW, "file=%s " str, file->name, ##args) 52 53 #define BLOBFS_CACHE_SIZE (4ULL * 1024 * 1024 * 1024) 54 #define SPDK_BLOBFS_OPTS_CLUSTER_SZ (1024 * 1024) 55 56 static uint64_t g_fs_cache_size = BLOBFS_CACHE_SIZE; 57 static struct spdk_mempool *g_cache_pool; 58 static TAILQ_HEAD(, spdk_file) g_caches; 59 static int g_fs_count = 0; 60 static pthread_mutex_t g_cache_init_lock = PTHREAD_MUTEX_INITIALIZER; 61 static pthread_spinlock_t g_caches_lock; 62 63 static void 64 __sem_post(void *arg, int bserrno) 65 { 66 sem_t *sem = arg; 67 68 sem_post(sem); 69 } 70 71 void 72 spdk_cache_buffer_free(struct cache_buffer *cache_buffer) 73 { 74 spdk_mempool_put(g_cache_pool, cache_buffer->buf); 75 free(cache_buffer); 76 } 77 78 #define CACHE_READAHEAD_THRESHOLD (128 * 1024) 79 80 struct spdk_file { 81 struct spdk_filesystem *fs; 82 struct spdk_blob *blob; 83 char *name; 84 uint64_t length; 85 bool is_deleted; 86 bool open_for_writing; 87 uint64_t length_flushed; 88 uint64_t append_pos; 89 uint64_t seq_byte_count; 90 uint64_t next_seq_offset; 91 uint32_t priority; 92 TAILQ_ENTRY(spdk_file) tailq; 93 spdk_blob_id blobid; 94 uint32_t ref_count; 95 pthread_spinlock_t lock; 96 struct cache_buffer *last; 97 struct cache_tree *tree; 98 TAILQ_HEAD(open_requests_head, spdk_fs_request) open_requests; 99 TAILQ_HEAD(sync_requests_head, spdk_fs_request) sync_requests; 100 TAILQ_ENTRY(spdk_file) cache_tailq; 101 }; 102 103 struct spdk_deleted_file { 104 spdk_blob_id id; 105 TAILQ_ENTRY(spdk_deleted_file) tailq; 106 }; 107 108 struct spdk_filesystem { 109 struct spdk_blob_store *bs; 110 TAILQ_HEAD(, spdk_file) files; 111 struct spdk_bs_opts bs_opts; 112 struct spdk_bs_dev *bdev; 113 fs_send_request_fn send_request; 114 115 struct { 116 uint32_t max_ops; 117 struct spdk_io_channel *sync_io_channel; 118 struct spdk_fs_channel *sync_fs_channel; 119 } sync_target; 120 121 struct { 122 uint32_t max_ops; 123 struct spdk_io_channel *md_io_channel; 124 struct spdk_fs_channel *md_fs_channel; 125 } md_target; 126 127 struct { 128 uint32_t max_ops; 129 } io_target; 130 }; 131 132 struct spdk_fs_cb_args { 133 union { 134 spdk_fs_op_with_handle_complete fs_op_with_handle; 135 spdk_fs_op_complete fs_op; 136 spdk_file_op_with_handle_complete file_op_with_handle; 137 spdk_file_op_complete file_op; 138 spdk_file_stat_op_complete stat_op; 139 } fn; 140 void *arg; 141 sem_t *sem; 142 struct spdk_filesystem *fs; 143 struct spdk_file *file; 144 int rc; 145 bool from_request; 146 union { 147 struct { 148 TAILQ_HEAD(, spdk_deleted_file) deleted_files; 149 } fs_load; 150 struct { 151 uint64_t length; 152 } truncate; 153 struct { 154 struct spdk_io_channel *channel; 155 void *user_buf; 156 void *pin_buf; 157 int is_read; 158 off_t offset; 159 size_t length; 160 uint64_t start_page; 161 uint64_t num_pages; 162 uint32_t blocklen; 163 } rw; 164 struct { 165 const char *old_name; 166 const char *new_name; 167 } rename; 168 struct { 169 struct cache_buffer *cache_buffer; 170 uint64_t length; 171 } flush; 172 struct { 173 struct cache_buffer *cache_buffer; 174 uint64_t length; 175 uint64_t offset; 176 } readahead; 177 struct { 178 uint64_t offset; 179 TAILQ_ENTRY(spdk_fs_request) tailq; 180 bool xattr_in_progress; 181 } sync; 182 struct { 183 uint32_t num_clusters; 184 } resize; 185 struct { 186 const char *name; 187 uint32_t flags; 188 TAILQ_ENTRY(spdk_fs_request) tailq; 189 } open; 190 struct { 191 const char *name; 192 } create; 193 struct { 194 const char *name; 195 } delete; 196 struct { 197 const char *name; 198 } stat; 199 } op; 200 }; 201 202 static void cache_free_buffers(struct spdk_file *file); 203 204 void 205 spdk_fs_opts_init(struct spdk_blobfs_opts *opts) 206 { 207 opts->cluster_sz = SPDK_BLOBFS_OPTS_CLUSTER_SZ; 208 } 209 210 static void 211 __initialize_cache(void) 212 { 213 assert(g_cache_pool == NULL); 214 215 g_cache_pool = spdk_mempool_create("spdk_fs_cache", 216 g_fs_cache_size / CACHE_BUFFER_SIZE, 217 CACHE_BUFFER_SIZE, 218 SPDK_MEMPOOL_DEFAULT_CACHE_SIZE, 219 SPDK_ENV_SOCKET_ID_ANY); 220 TAILQ_INIT(&g_caches); 221 pthread_spin_init(&g_caches_lock, 0); 222 } 223 224 static void 225 __free_cache(void) 226 { 227 assert(g_cache_pool != NULL); 228 229 spdk_mempool_free(g_cache_pool); 230 g_cache_pool = NULL; 231 } 232 233 static uint64_t 234 __file_get_blob_size(struct spdk_file *file) 235 { 236 uint64_t cluster_sz; 237 238 cluster_sz = file->fs->bs_opts.cluster_sz; 239 return cluster_sz * spdk_blob_get_num_clusters(file->blob); 240 } 241 242 struct spdk_fs_request { 243 struct spdk_fs_cb_args args; 244 TAILQ_ENTRY(spdk_fs_request) link; 245 struct spdk_fs_channel *channel; 246 }; 247 248 struct spdk_fs_channel { 249 struct spdk_fs_request *req_mem; 250 TAILQ_HEAD(, spdk_fs_request) reqs; 251 sem_t sem; 252 struct spdk_filesystem *fs; 253 struct spdk_io_channel *bs_channel; 254 fs_send_request_fn send_request; 255 bool sync; 256 pthread_spinlock_t lock; 257 }; 258 259 static struct spdk_fs_request * 260 alloc_fs_request(struct spdk_fs_channel *channel) 261 { 262 struct spdk_fs_request *req; 263 264 if (channel->sync) { 265 pthread_spin_lock(&channel->lock); 266 } 267 268 req = TAILQ_FIRST(&channel->reqs); 269 if (req) { 270 TAILQ_REMOVE(&channel->reqs, req, link); 271 } 272 273 if (channel->sync) { 274 pthread_spin_unlock(&channel->lock); 275 } 276 277 if (req == NULL) { 278 return NULL; 279 } 280 memset(req, 0, sizeof(*req)); 281 req->channel = channel; 282 req->args.from_request = true; 283 284 return req; 285 } 286 287 static void 288 free_fs_request(struct spdk_fs_request *req) 289 { 290 struct spdk_fs_channel *channel = req->channel; 291 292 if (channel->sync) { 293 pthread_spin_lock(&channel->lock); 294 } 295 296 TAILQ_INSERT_HEAD(&req->channel->reqs, req, link); 297 298 if (channel->sync) { 299 pthread_spin_unlock(&channel->lock); 300 } 301 } 302 303 static int 304 _spdk_fs_channel_create(struct spdk_filesystem *fs, struct spdk_fs_channel *channel, 305 uint32_t max_ops) 306 { 307 uint32_t i; 308 309 channel->req_mem = calloc(max_ops, sizeof(struct spdk_fs_request)); 310 if (!channel->req_mem) { 311 return -1; 312 } 313 314 TAILQ_INIT(&channel->reqs); 315 sem_init(&channel->sem, 0, 0); 316 317 for (i = 0; i < max_ops; i++) { 318 TAILQ_INSERT_TAIL(&channel->reqs, &channel->req_mem[i], link); 319 } 320 321 channel->fs = fs; 322 323 return 0; 324 } 325 326 static int 327 _spdk_fs_md_channel_create(void *io_device, void *ctx_buf) 328 { 329 struct spdk_filesystem *fs; 330 struct spdk_fs_channel *channel = ctx_buf; 331 332 fs = SPDK_CONTAINEROF(io_device, struct spdk_filesystem, md_target); 333 334 return _spdk_fs_channel_create(fs, channel, fs->md_target.max_ops); 335 } 336 337 static int 338 _spdk_fs_sync_channel_create(void *io_device, void *ctx_buf) 339 { 340 struct spdk_filesystem *fs; 341 struct spdk_fs_channel *channel = ctx_buf; 342 343 fs = SPDK_CONTAINEROF(io_device, struct spdk_filesystem, sync_target); 344 345 return _spdk_fs_channel_create(fs, channel, fs->sync_target.max_ops); 346 } 347 348 static int 349 _spdk_fs_io_channel_create(void *io_device, void *ctx_buf) 350 { 351 struct spdk_filesystem *fs; 352 struct spdk_fs_channel *channel = ctx_buf; 353 354 fs = SPDK_CONTAINEROF(io_device, struct spdk_filesystem, io_target); 355 356 return _spdk_fs_channel_create(fs, channel, fs->io_target.max_ops); 357 } 358 359 static void 360 _spdk_fs_channel_destroy(void *io_device, void *ctx_buf) 361 { 362 struct spdk_fs_channel *channel = ctx_buf; 363 364 free(channel->req_mem); 365 if (channel->bs_channel != NULL) { 366 spdk_bs_free_io_channel(channel->bs_channel); 367 } 368 } 369 370 static void 371 __send_request_direct(fs_request_fn fn, void *arg) 372 { 373 fn(arg); 374 } 375 376 static void 377 common_fs_bs_init(struct spdk_filesystem *fs, struct spdk_blob_store *bs) 378 { 379 fs->bs = bs; 380 fs->bs_opts.cluster_sz = spdk_bs_get_cluster_size(bs); 381 fs->md_target.md_fs_channel->bs_channel = spdk_bs_alloc_io_channel(fs->bs); 382 fs->md_target.md_fs_channel->send_request = __send_request_direct; 383 fs->sync_target.sync_fs_channel->bs_channel = spdk_bs_alloc_io_channel(fs->bs); 384 fs->sync_target.sync_fs_channel->send_request = __send_request_direct; 385 386 pthread_mutex_lock(&g_cache_init_lock); 387 if (g_fs_count == 0) { 388 __initialize_cache(); 389 } 390 g_fs_count++; 391 pthread_mutex_unlock(&g_cache_init_lock); 392 } 393 394 static void 395 init_cb(void *ctx, struct spdk_blob_store *bs, int bserrno) 396 { 397 struct spdk_fs_request *req = ctx; 398 struct spdk_fs_cb_args *args = &req->args; 399 struct spdk_filesystem *fs = args->fs; 400 401 if (bserrno == 0) { 402 common_fs_bs_init(fs, bs); 403 } else { 404 free(fs); 405 fs = NULL; 406 } 407 408 args->fn.fs_op_with_handle(args->arg, fs, bserrno); 409 free_fs_request(req); 410 } 411 412 static void 413 fs_conf_parse(void) 414 { 415 struct spdk_conf_section *sp; 416 417 sp = spdk_conf_find_section(NULL, "Blobfs"); 418 if (sp == NULL) { 419 g_fs_cache_buffer_shift = CACHE_BUFFER_SHIFT_DEFAULT; 420 return; 421 } 422 423 g_fs_cache_buffer_shift = spdk_conf_section_get_intval(sp, "CacheBufferShift"); 424 if (g_fs_cache_buffer_shift <= 0) { 425 g_fs_cache_buffer_shift = CACHE_BUFFER_SHIFT_DEFAULT; 426 } 427 } 428 429 static struct spdk_filesystem * 430 fs_alloc(struct spdk_bs_dev *dev, fs_send_request_fn send_request_fn) 431 { 432 struct spdk_filesystem *fs; 433 434 fs = calloc(1, sizeof(*fs)); 435 if (fs == NULL) { 436 return NULL; 437 } 438 439 fs->bdev = dev; 440 fs->send_request = send_request_fn; 441 TAILQ_INIT(&fs->files); 442 443 fs->md_target.max_ops = 512; 444 spdk_io_device_register(&fs->md_target, _spdk_fs_md_channel_create, _spdk_fs_channel_destroy, 445 sizeof(struct spdk_fs_channel)); 446 fs->md_target.md_io_channel = spdk_get_io_channel(&fs->md_target); 447 fs->md_target.md_fs_channel = spdk_io_channel_get_ctx(fs->md_target.md_io_channel); 448 449 fs->sync_target.max_ops = 512; 450 spdk_io_device_register(&fs->sync_target, _spdk_fs_sync_channel_create, _spdk_fs_channel_destroy, 451 sizeof(struct spdk_fs_channel)); 452 fs->sync_target.sync_io_channel = spdk_get_io_channel(&fs->sync_target); 453 fs->sync_target.sync_fs_channel = spdk_io_channel_get_ctx(fs->sync_target.sync_io_channel); 454 455 fs->io_target.max_ops = 512; 456 spdk_io_device_register(&fs->io_target, _spdk_fs_io_channel_create, _spdk_fs_channel_destroy, 457 sizeof(struct spdk_fs_channel)); 458 459 return fs; 460 } 461 462 void 463 spdk_fs_init(struct spdk_bs_dev *dev, struct spdk_blobfs_opts *opt, 464 fs_send_request_fn send_request_fn, 465 spdk_fs_op_with_handle_complete cb_fn, void *cb_arg) 466 { 467 struct spdk_filesystem *fs; 468 struct spdk_fs_request *req; 469 struct spdk_fs_cb_args *args; 470 struct spdk_bs_opts opts = {}; 471 472 fs = fs_alloc(dev, send_request_fn); 473 if (fs == NULL) { 474 cb_fn(cb_arg, NULL, -ENOMEM); 475 return; 476 } 477 478 fs_conf_parse(); 479 480 req = alloc_fs_request(fs->md_target.md_fs_channel); 481 if (req == NULL) { 482 spdk_put_io_channel(fs->md_target.md_io_channel); 483 spdk_io_device_unregister(&fs->md_target, NULL); 484 spdk_put_io_channel(fs->sync_target.sync_io_channel); 485 spdk_io_device_unregister(&fs->sync_target, NULL); 486 spdk_io_device_unregister(&fs->io_target, NULL); 487 free(fs); 488 cb_fn(cb_arg, NULL, -ENOMEM); 489 return; 490 } 491 492 args = &req->args; 493 args->fn.fs_op_with_handle = cb_fn; 494 args->arg = cb_arg; 495 args->fs = fs; 496 497 spdk_bs_opts_init(&opts); 498 strncpy(opts.bstype.bstype, "BLOBFS", SPDK_BLOBSTORE_TYPE_LENGTH); 499 if (opt) { 500 opts.cluster_sz = opt->cluster_sz; 501 } 502 spdk_bs_init(dev, &opts, init_cb, req); 503 } 504 505 static struct spdk_file * 506 file_alloc(struct spdk_filesystem *fs) 507 { 508 struct spdk_file *file; 509 510 file = calloc(1, sizeof(*file)); 511 if (file == NULL) { 512 return NULL; 513 } 514 515 file->tree = calloc(1, sizeof(*file->tree)); 516 if (file->tree == NULL) { 517 free(file); 518 return NULL; 519 } 520 521 file->fs = fs; 522 TAILQ_INIT(&file->open_requests); 523 TAILQ_INIT(&file->sync_requests); 524 pthread_spin_init(&file->lock, 0); 525 TAILQ_INSERT_TAIL(&fs->files, file, tailq); 526 file->priority = SPDK_FILE_PRIORITY_LOW; 527 return file; 528 } 529 530 static void iter_delete_cb(void *ctx, int bserrno); 531 532 static int 533 _handle_deleted_files(struct spdk_fs_request *req) 534 { 535 struct spdk_fs_cb_args *args = &req->args; 536 struct spdk_filesystem *fs = args->fs; 537 538 if (!TAILQ_EMPTY(&args->op.fs_load.deleted_files)) { 539 struct spdk_deleted_file *deleted_file; 540 541 deleted_file = TAILQ_FIRST(&args->op.fs_load.deleted_files); 542 TAILQ_REMOVE(&args->op.fs_load.deleted_files, deleted_file, tailq); 543 spdk_bs_delete_blob(fs->bs, deleted_file->id, iter_delete_cb, req); 544 free(deleted_file); 545 return 0; 546 } 547 548 return 1; 549 } 550 551 static void 552 iter_delete_cb(void *ctx, int bserrno) 553 { 554 struct spdk_fs_request *req = ctx; 555 struct spdk_fs_cb_args *args = &req->args; 556 struct spdk_filesystem *fs = args->fs; 557 558 if (_handle_deleted_files(req) == 0) { 559 return; 560 } 561 562 args->fn.fs_op_with_handle(args->arg, fs, 0); 563 free_fs_request(req); 564 565 } 566 567 static void 568 iter_cb(void *ctx, struct spdk_blob *blob, int rc) 569 { 570 struct spdk_fs_request *req = ctx; 571 struct spdk_fs_cb_args *args = &req->args; 572 struct spdk_filesystem *fs = args->fs; 573 uint64_t *length; 574 const char *name; 575 uint32_t *is_deleted; 576 size_t value_len; 577 578 if (rc == -ENOENT) { 579 /* Finished iterating */ 580 if (_handle_deleted_files(req) == 0) { 581 return; 582 } 583 args->fn.fs_op_with_handle(args->arg, fs, 0); 584 free_fs_request(req); 585 return; 586 } else if (rc < 0) { 587 args->fn.fs_op_with_handle(args->arg, fs, rc); 588 free_fs_request(req); 589 return; 590 } 591 592 rc = spdk_blob_get_xattr_value(blob, "name", (const void **)&name, &value_len); 593 if (rc < 0) { 594 args->fn.fs_op_with_handle(args->arg, fs, rc); 595 free_fs_request(req); 596 return; 597 } 598 599 rc = spdk_blob_get_xattr_value(blob, "length", (const void **)&length, &value_len); 600 if (rc < 0) { 601 args->fn.fs_op_with_handle(args->arg, fs, rc); 602 free_fs_request(req); 603 return; 604 } 605 606 assert(value_len == 8); 607 608 /* This file could be deleted last time without close it, then app crashed, so we delete it now */ 609 rc = spdk_blob_get_xattr_value(blob, "is_deleted", (const void **)&is_deleted, &value_len); 610 if (rc < 0) { 611 struct spdk_file *f; 612 613 f = file_alloc(fs); 614 if (f == NULL) { 615 args->fn.fs_op_with_handle(args->arg, fs, -ENOMEM); 616 free_fs_request(req); 617 return; 618 } 619 620 f->name = strdup(name); 621 f->blobid = spdk_blob_get_id(blob); 622 f->length = *length; 623 f->length_flushed = *length; 624 f->append_pos = *length; 625 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "added file %s length=%ju\n", f->name, f->length); 626 } else { 627 struct spdk_deleted_file *deleted_file; 628 629 deleted_file = calloc(1, sizeof(*deleted_file)); 630 if (deleted_file == NULL) { 631 args->fn.fs_op_with_handle(args->arg, fs, -ENOMEM); 632 free_fs_request(req); 633 return; 634 } 635 deleted_file->id = spdk_blob_get_id(blob); 636 TAILQ_INSERT_TAIL(&args->op.fs_load.deleted_files, deleted_file, tailq); 637 } 638 639 spdk_bs_iter_next(fs->bs, blob, iter_cb, req); 640 } 641 642 static void 643 load_cb(void *ctx, struct spdk_blob_store *bs, int bserrno) 644 { 645 struct spdk_fs_request *req = ctx; 646 struct spdk_fs_cb_args *args = &req->args; 647 struct spdk_filesystem *fs = args->fs; 648 struct spdk_bs_type bstype; 649 static const struct spdk_bs_type blobfs_type = {"BLOBFS"}; 650 static const struct spdk_bs_type zeros; 651 652 if (bserrno != 0) { 653 args->fn.fs_op_with_handle(args->arg, NULL, bserrno); 654 free_fs_request(req); 655 free(fs); 656 return; 657 } 658 659 bstype = spdk_bs_get_bstype(bs); 660 661 if (!memcmp(&bstype, &zeros, sizeof(bstype))) { 662 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "assigning bstype\n"); 663 spdk_bs_set_bstype(bs, blobfs_type); 664 } else if (memcmp(&bstype, &blobfs_type, sizeof(bstype))) { 665 SPDK_DEBUGLOG(SPDK_LOG_BLOB, "not blobfs\n"); 666 SPDK_TRACEDUMP(SPDK_LOG_BLOB, "bstype", &bstype, sizeof(bstype)); 667 args->fn.fs_op_with_handle(args->arg, NULL, bserrno); 668 free_fs_request(req); 669 free(fs); 670 return; 671 } 672 673 common_fs_bs_init(fs, bs); 674 spdk_bs_iter_first(fs->bs, iter_cb, req); 675 } 676 677 void 678 spdk_fs_load(struct spdk_bs_dev *dev, fs_send_request_fn send_request_fn, 679 spdk_fs_op_with_handle_complete cb_fn, void *cb_arg) 680 { 681 struct spdk_filesystem *fs; 682 struct spdk_fs_cb_args *args; 683 struct spdk_fs_request *req; 684 685 fs = fs_alloc(dev, send_request_fn); 686 if (fs == NULL) { 687 cb_fn(cb_arg, NULL, -ENOMEM); 688 return; 689 } 690 691 fs_conf_parse(); 692 693 req = alloc_fs_request(fs->md_target.md_fs_channel); 694 if (req == NULL) { 695 spdk_put_io_channel(fs->md_target.md_io_channel); 696 spdk_io_device_unregister(&fs->md_target, NULL); 697 spdk_put_io_channel(fs->sync_target.sync_io_channel); 698 spdk_io_device_unregister(&fs->sync_target, NULL); 699 spdk_io_device_unregister(&fs->io_target, NULL); 700 free(fs); 701 cb_fn(cb_arg, NULL, -ENOMEM); 702 return; 703 } 704 705 args = &req->args; 706 args->fn.fs_op_with_handle = cb_fn; 707 args->arg = cb_arg; 708 args->fs = fs; 709 TAILQ_INIT(&args->op.fs_load.deleted_files); 710 spdk_bs_load(dev, NULL, load_cb, req); 711 } 712 713 static void 714 unload_cb(void *ctx, int bserrno) 715 { 716 struct spdk_fs_request *req = ctx; 717 struct spdk_fs_cb_args *args = &req->args; 718 struct spdk_filesystem *fs = args->fs; 719 720 pthread_mutex_lock(&g_cache_init_lock); 721 g_fs_count--; 722 if (g_fs_count == 0) { 723 __free_cache(); 724 } 725 pthread_mutex_unlock(&g_cache_init_lock); 726 727 args->fn.fs_op(args->arg, bserrno); 728 free(req); 729 730 spdk_io_device_unregister(&fs->io_target, NULL); 731 spdk_io_device_unregister(&fs->sync_target, NULL); 732 spdk_io_device_unregister(&fs->md_target, NULL); 733 734 free(fs); 735 } 736 737 void 738 spdk_fs_unload(struct spdk_filesystem *fs, spdk_fs_op_complete cb_fn, void *cb_arg) 739 { 740 struct spdk_fs_request *req; 741 struct spdk_fs_cb_args *args; 742 743 /* 744 * We must free the md_channel before unloading the blobstore, so just 745 * allocate this request from the general heap. 746 */ 747 req = calloc(1, sizeof(*req)); 748 if (req == NULL) { 749 cb_fn(cb_arg, -ENOMEM); 750 return; 751 } 752 753 args = &req->args; 754 args->fn.fs_op = cb_fn; 755 args->arg = cb_arg; 756 args->fs = fs; 757 758 spdk_fs_free_io_channel(fs->md_target.md_io_channel); 759 spdk_fs_free_io_channel(fs->sync_target.sync_io_channel); 760 spdk_bs_unload(fs->bs, unload_cb, req); 761 } 762 763 static struct spdk_file * 764 fs_find_file(struct spdk_filesystem *fs, const char *name) 765 { 766 struct spdk_file *file; 767 768 TAILQ_FOREACH(file, &fs->files, tailq) { 769 if (!strncmp(name, file->name, SPDK_FILE_NAME_MAX)) { 770 return file; 771 } 772 } 773 774 return NULL; 775 } 776 777 void 778 spdk_fs_file_stat_async(struct spdk_filesystem *fs, const char *name, 779 spdk_file_stat_op_complete cb_fn, void *cb_arg) 780 { 781 struct spdk_file_stat stat; 782 struct spdk_file *f = NULL; 783 784 if (strnlen(name, SPDK_FILE_NAME_MAX + 1) == SPDK_FILE_NAME_MAX + 1) { 785 cb_fn(cb_arg, NULL, -ENAMETOOLONG); 786 return; 787 } 788 789 f = fs_find_file(fs, name); 790 if (f != NULL) { 791 stat.blobid = f->blobid; 792 stat.size = f->append_pos >= f->length ? f->append_pos : f->length; 793 cb_fn(cb_arg, &stat, 0); 794 return; 795 } 796 797 cb_fn(cb_arg, NULL, -ENOENT); 798 } 799 800 static void 801 __copy_stat(void *arg, struct spdk_file_stat *stat, int fserrno) 802 { 803 struct spdk_fs_request *req = arg; 804 struct spdk_fs_cb_args *args = &req->args; 805 806 args->rc = fserrno; 807 if (fserrno == 0) { 808 memcpy(args->arg, stat, sizeof(*stat)); 809 } 810 sem_post(args->sem); 811 } 812 813 static void 814 __file_stat(void *arg) 815 { 816 struct spdk_fs_request *req = arg; 817 struct spdk_fs_cb_args *args = &req->args; 818 819 spdk_fs_file_stat_async(args->fs, args->op.stat.name, 820 args->fn.stat_op, req); 821 } 822 823 int 824 spdk_fs_file_stat(struct spdk_filesystem *fs, struct spdk_io_channel *_channel, 825 const char *name, struct spdk_file_stat *stat) 826 { 827 struct spdk_fs_channel *channel = spdk_io_channel_get_ctx(_channel); 828 struct spdk_fs_request *req; 829 int rc; 830 831 req = alloc_fs_request(channel); 832 assert(req != NULL); 833 834 req->args.fs = fs; 835 req->args.op.stat.name = name; 836 req->args.fn.stat_op = __copy_stat; 837 req->args.arg = stat; 838 req->args.sem = &channel->sem; 839 channel->send_request(__file_stat, req); 840 sem_wait(&channel->sem); 841 842 rc = req->args.rc; 843 free_fs_request(req); 844 845 return rc; 846 } 847 848 static void 849 fs_create_blob_close_cb(void *ctx, int bserrno) 850 { 851 struct spdk_fs_request *req = ctx; 852 struct spdk_fs_cb_args *args = &req->args; 853 854 args->fn.file_op(args->arg, bserrno); 855 free_fs_request(req); 856 } 857 858 static void 859 fs_create_blob_open_cb(void *ctx, struct spdk_blob *blob, int bserrno) 860 { 861 struct spdk_fs_request *req = ctx; 862 struct spdk_fs_cb_args *args = &req->args; 863 struct spdk_file *f = args->file; 864 uint64_t length = 0; 865 866 spdk_blob_resize(blob, 1); 867 spdk_blob_set_xattr(blob, "name", f->name, strlen(f->name) + 1); 868 spdk_blob_set_xattr(blob, "length", &length, sizeof(length)); 869 870 spdk_blob_close(blob, fs_create_blob_close_cb, args); 871 } 872 873 static void 874 fs_create_blob_create_cb(void *ctx, spdk_blob_id blobid, int bserrno) 875 { 876 struct spdk_fs_request *req = ctx; 877 struct spdk_fs_cb_args *args = &req->args; 878 struct spdk_file *f = args->file; 879 880 f->blobid = blobid; 881 spdk_bs_open_blob(f->fs->bs, blobid, fs_create_blob_open_cb, req); 882 } 883 884 void 885 spdk_fs_create_file_async(struct spdk_filesystem *fs, const char *name, 886 spdk_file_op_complete cb_fn, void *cb_arg) 887 { 888 struct spdk_file *file; 889 struct spdk_fs_request *req; 890 struct spdk_fs_cb_args *args; 891 892 if (strnlen(name, SPDK_FILE_NAME_MAX + 1) == SPDK_FILE_NAME_MAX + 1) { 893 cb_fn(cb_arg, -ENAMETOOLONG); 894 return; 895 } 896 897 file = fs_find_file(fs, name); 898 if (file != NULL) { 899 cb_fn(cb_arg, -EEXIST); 900 return; 901 } 902 903 file = file_alloc(fs); 904 if (file == NULL) { 905 cb_fn(cb_arg, -ENOMEM); 906 return; 907 } 908 909 req = alloc_fs_request(fs->md_target.md_fs_channel); 910 if (req == NULL) { 911 cb_fn(cb_arg, -ENOMEM); 912 return; 913 } 914 915 args = &req->args; 916 args->file = file; 917 args->fn.file_op = cb_fn; 918 args->arg = cb_arg; 919 920 file->name = strdup(name); 921 spdk_bs_create_blob(fs->bs, fs_create_blob_create_cb, args); 922 } 923 924 static void 925 __fs_create_file_done(void *arg, int fserrno) 926 { 927 struct spdk_fs_request *req = arg; 928 struct spdk_fs_cb_args *args = &req->args; 929 930 args->rc = fserrno; 931 sem_post(args->sem); 932 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "file=%s\n", args->op.create.name); 933 } 934 935 static void 936 __fs_create_file(void *arg) 937 { 938 struct spdk_fs_request *req = arg; 939 struct spdk_fs_cb_args *args = &req->args; 940 941 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "file=%s\n", args->op.create.name); 942 spdk_fs_create_file_async(args->fs, args->op.create.name, __fs_create_file_done, req); 943 } 944 945 int 946 spdk_fs_create_file(struct spdk_filesystem *fs, struct spdk_io_channel *_channel, const char *name) 947 { 948 struct spdk_fs_channel *channel = spdk_io_channel_get_ctx(_channel); 949 struct spdk_fs_request *req; 950 struct spdk_fs_cb_args *args; 951 int rc; 952 953 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "file=%s\n", name); 954 955 req = alloc_fs_request(channel); 956 assert(req != NULL); 957 958 args = &req->args; 959 args->fs = fs; 960 args->op.create.name = name; 961 args->sem = &channel->sem; 962 fs->send_request(__fs_create_file, req); 963 sem_wait(&channel->sem); 964 rc = args->rc; 965 free_fs_request(req); 966 967 return rc; 968 } 969 970 static void 971 fs_open_blob_done(void *ctx, struct spdk_blob *blob, int bserrno) 972 { 973 struct spdk_fs_request *req = ctx; 974 struct spdk_fs_cb_args *args = &req->args; 975 struct spdk_file *f = args->file; 976 977 f->blob = blob; 978 while (!TAILQ_EMPTY(&f->open_requests)) { 979 req = TAILQ_FIRST(&f->open_requests); 980 args = &req->args; 981 TAILQ_REMOVE(&f->open_requests, req, args.op.open.tailq); 982 args->fn.file_op_with_handle(args->arg, f, bserrno); 983 free_fs_request(req); 984 } 985 } 986 987 static void 988 fs_open_blob_create_cb(void *ctx, int bserrno) 989 { 990 struct spdk_fs_request *req = ctx; 991 struct spdk_fs_cb_args *args = &req->args; 992 struct spdk_file *file = args->file; 993 struct spdk_filesystem *fs = args->fs; 994 995 if (file == NULL) { 996 /* 997 * This is from an open with CREATE flag - the file 998 * is now created so look it up in the file list for this 999 * filesystem. 1000 */ 1001 file = fs_find_file(fs, args->op.open.name); 1002 assert(file != NULL); 1003 args->file = file; 1004 } 1005 1006 file->ref_count++; 1007 TAILQ_INSERT_TAIL(&file->open_requests, req, args.op.open.tailq); 1008 if (file->ref_count == 1) { 1009 assert(file->blob == NULL); 1010 spdk_bs_open_blob(fs->bs, file->blobid, fs_open_blob_done, req); 1011 } else if (file->blob != NULL) { 1012 fs_open_blob_done(req, file->blob, 0); 1013 } else { 1014 /* 1015 * The blob open for this file is in progress due to a previous 1016 * open request. When that open completes, it will invoke the 1017 * open callback for this request. 1018 */ 1019 } 1020 } 1021 1022 void 1023 spdk_fs_open_file_async(struct spdk_filesystem *fs, const char *name, uint32_t flags, 1024 spdk_file_op_with_handle_complete cb_fn, void *cb_arg) 1025 { 1026 struct spdk_file *f = NULL; 1027 struct spdk_fs_request *req; 1028 struct spdk_fs_cb_args *args; 1029 1030 if (strnlen(name, SPDK_FILE_NAME_MAX + 1) == SPDK_FILE_NAME_MAX + 1) { 1031 cb_fn(cb_arg, NULL, -ENAMETOOLONG); 1032 return; 1033 } 1034 1035 f = fs_find_file(fs, name); 1036 if (f == NULL && !(flags & SPDK_BLOBFS_OPEN_CREATE)) { 1037 cb_fn(cb_arg, NULL, -ENOENT); 1038 return; 1039 } 1040 1041 if (f != NULL && f->is_deleted == true) { 1042 cb_fn(cb_arg, NULL, -ENOENT); 1043 return; 1044 } 1045 1046 req = alloc_fs_request(fs->md_target.md_fs_channel); 1047 if (req == NULL) { 1048 cb_fn(cb_arg, NULL, -ENOMEM); 1049 return; 1050 } 1051 1052 args = &req->args; 1053 args->fn.file_op_with_handle = cb_fn; 1054 args->arg = cb_arg; 1055 args->file = f; 1056 args->fs = fs; 1057 args->op.open.name = name; 1058 1059 if (f == NULL) { 1060 spdk_fs_create_file_async(fs, name, fs_open_blob_create_cb, req); 1061 } else { 1062 fs_open_blob_create_cb(req, 0); 1063 } 1064 } 1065 1066 static void 1067 __fs_open_file_done(void *arg, struct spdk_file *file, int bserrno) 1068 { 1069 struct spdk_fs_request *req = arg; 1070 struct spdk_fs_cb_args *args = &req->args; 1071 1072 args->file = file; 1073 args->rc = bserrno; 1074 sem_post(args->sem); 1075 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "file=%s\n", args->op.open.name); 1076 } 1077 1078 static void 1079 __fs_open_file(void *arg) 1080 { 1081 struct spdk_fs_request *req = arg; 1082 struct spdk_fs_cb_args *args = &req->args; 1083 1084 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "file=%s\n", args->op.open.name); 1085 spdk_fs_open_file_async(args->fs, args->op.open.name, args->op.open.flags, 1086 __fs_open_file_done, req); 1087 } 1088 1089 int 1090 spdk_fs_open_file(struct spdk_filesystem *fs, struct spdk_io_channel *_channel, 1091 const char *name, uint32_t flags, struct spdk_file **file) 1092 { 1093 struct spdk_fs_channel *channel = spdk_io_channel_get_ctx(_channel); 1094 struct spdk_fs_request *req; 1095 struct spdk_fs_cb_args *args; 1096 int rc; 1097 1098 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "file=%s\n", name); 1099 1100 req = alloc_fs_request(channel); 1101 assert(req != NULL); 1102 1103 args = &req->args; 1104 args->fs = fs; 1105 args->op.open.name = name; 1106 args->op.open.flags = flags; 1107 args->sem = &channel->sem; 1108 fs->send_request(__fs_open_file, req); 1109 sem_wait(&channel->sem); 1110 rc = args->rc; 1111 if (rc == 0) { 1112 *file = args->file; 1113 } else { 1114 *file = NULL; 1115 } 1116 free_fs_request(req); 1117 1118 return rc; 1119 } 1120 1121 static void 1122 fs_rename_blob_close_cb(void *ctx, int bserrno) 1123 { 1124 struct spdk_fs_request *req = ctx; 1125 struct spdk_fs_cb_args *args = &req->args; 1126 1127 args->fn.fs_op(args->arg, bserrno); 1128 free_fs_request(req); 1129 } 1130 1131 static void 1132 fs_rename_blob_open_cb(void *ctx, struct spdk_blob *blob, int bserrno) 1133 { 1134 struct spdk_fs_request *req = ctx; 1135 struct spdk_fs_cb_args *args = &req->args; 1136 const char *new_name = args->op.rename.new_name; 1137 1138 spdk_blob_set_xattr(blob, "name", new_name, strlen(new_name) + 1); 1139 spdk_blob_close(blob, fs_rename_blob_close_cb, req); 1140 } 1141 1142 static void 1143 __spdk_fs_md_rename_file(struct spdk_fs_request *req) 1144 { 1145 struct spdk_fs_cb_args *args = &req->args; 1146 struct spdk_file *f; 1147 1148 f = fs_find_file(args->fs, args->op.rename.old_name); 1149 if (f == NULL) { 1150 args->fn.fs_op(args->arg, -ENOENT); 1151 free_fs_request(req); 1152 return; 1153 } 1154 1155 free(f->name); 1156 f->name = strdup(args->op.rename.new_name); 1157 args->file = f; 1158 spdk_bs_open_blob(args->fs->bs, f->blobid, fs_rename_blob_open_cb, req); 1159 } 1160 1161 static void 1162 fs_rename_delete_done(void *arg, int fserrno) 1163 { 1164 __spdk_fs_md_rename_file(arg); 1165 } 1166 1167 void 1168 spdk_fs_rename_file_async(struct spdk_filesystem *fs, 1169 const char *old_name, const char *new_name, 1170 spdk_file_op_complete cb_fn, void *cb_arg) 1171 { 1172 struct spdk_file *f; 1173 struct spdk_fs_request *req; 1174 struct spdk_fs_cb_args *args; 1175 1176 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "old=%s new=%s\n", old_name, new_name); 1177 if (strnlen(new_name, SPDK_FILE_NAME_MAX + 1) == SPDK_FILE_NAME_MAX + 1) { 1178 cb_fn(cb_arg, -ENAMETOOLONG); 1179 return; 1180 } 1181 1182 req = alloc_fs_request(fs->md_target.md_fs_channel); 1183 if (req == NULL) { 1184 cb_fn(cb_arg, -ENOMEM); 1185 return; 1186 } 1187 1188 args = &req->args; 1189 args->fn.fs_op = cb_fn; 1190 args->fs = fs; 1191 args->arg = cb_arg; 1192 args->op.rename.old_name = old_name; 1193 args->op.rename.new_name = new_name; 1194 1195 f = fs_find_file(fs, new_name); 1196 if (f == NULL) { 1197 __spdk_fs_md_rename_file(req); 1198 return; 1199 } 1200 1201 /* 1202 * The rename overwrites an existing file. So delete the existing file, then 1203 * do the actual rename. 1204 */ 1205 spdk_fs_delete_file_async(fs, new_name, fs_rename_delete_done, req); 1206 } 1207 1208 static void 1209 __fs_rename_file_done(void *arg, int fserrno) 1210 { 1211 struct spdk_fs_request *req = arg; 1212 struct spdk_fs_cb_args *args = &req->args; 1213 1214 args->rc = fserrno; 1215 sem_post(args->sem); 1216 } 1217 1218 static void 1219 __fs_rename_file(void *arg) 1220 { 1221 struct spdk_fs_request *req = arg; 1222 struct spdk_fs_cb_args *args = &req->args; 1223 1224 spdk_fs_rename_file_async(args->fs, args->op.rename.old_name, args->op.rename.new_name, 1225 __fs_rename_file_done, req); 1226 } 1227 1228 int 1229 spdk_fs_rename_file(struct spdk_filesystem *fs, struct spdk_io_channel *_channel, 1230 const char *old_name, const char *new_name) 1231 { 1232 struct spdk_fs_channel *channel = spdk_io_channel_get_ctx(_channel); 1233 struct spdk_fs_request *req; 1234 struct spdk_fs_cb_args *args; 1235 int rc; 1236 1237 req = alloc_fs_request(channel); 1238 assert(req != NULL); 1239 1240 args = &req->args; 1241 1242 args->fs = fs; 1243 args->op.rename.old_name = old_name; 1244 args->op.rename.new_name = new_name; 1245 args->sem = &channel->sem; 1246 fs->send_request(__fs_rename_file, req); 1247 sem_wait(&channel->sem); 1248 rc = args->rc; 1249 free_fs_request(req); 1250 return rc; 1251 } 1252 1253 static void 1254 blob_delete_cb(void *ctx, int bserrno) 1255 { 1256 struct spdk_fs_request *req = ctx; 1257 struct spdk_fs_cb_args *args = &req->args; 1258 1259 args->fn.file_op(args->arg, bserrno); 1260 free_fs_request(req); 1261 } 1262 1263 void 1264 spdk_fs_delete_file_async(struct spdk_filesystem *fs, const char *name, 1265 spdk_file_op_complete cb_fn, void *cb_arg) 1266 { 1267 struct spdk_file *f; 1268 spdk_blob_id blobid; 1269 struct spdk_fs_request *req; 1270 struct spdk_fs_cb_args *args; 1271 1272 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "file=%s\n", name); 1273 1274 if (strnlen(name, SPDK_FILE_NAME_MAX + 1) == SPDK_FILE_NAME_MAX + 1) { 1275 cb_fn(cb_arg, -ENAMETOOLONG); 1276 return; 1277 } 1278 1279 f = fs_find_file(fs, name); 1280 if (f == NULL) { 1281 cb_fn(cb_arg, -ENOENT); 1282 return; 1283 } 1284 1285 req = alloc_fs_request(fs->md_target.md_fs_channel); 1286 if (req == NULL) { 1287 cb_fn(cb_arg, -ENOMEM); 1288 return; 1289 } 1290 1291 args = &req->args; 1292 args->fn.file_op = cb_fn; 1293 args->arg = cb_arg; 1294 1295 if (f->ref_count > 0) { 1296 /* If the ref > 0, we mark the file as deleted and delete it when we close it. */ 1297 f->is_deleted = true; 1298 spdk_blob_set_xattr(f->blob, "is_deleted", &f->is_deleted, sizeof(bool)); 1299 spdk_blob_sync_md(f->blob, blob_delete_cb, args); 1300 return; 1301 } 1302 1303 TAILQ_REMOVE(&fs->files, f, tailq); 1304 1305 cache_free_buffers(f); 1306 1307 blobid = f->blobid; 1308 1309 free(f->name); 1310 free(f->tree); 1311 free(f); 1312 1313 spdk_bs_delete_blob(fs->bs, blobid, blob_delete_cb, req); 1314 } 1315 1316 static void 1317 __fs_delete_file_done(void *arg, int fserrno) 1318 { 1319 struct spdk_fs_request *req = arg; 1320 struct spdk_fs_cb_args *args = &req->args; 1321 1322 args->rc = fserrno; 1323 sem_post(args->sem); 1324 } 1325 1326 static void 1327 __fs_delete_file(void *arg) 1328 { 1329 struct spdk_fs_request *req = arg; 1330 struct spdk_fs_cb_args *args = &req->args; 1331 1332 spdk_fs_delete_file_async(args->fs, args->op.delete.name, __fs_delete_file_done, req); 1333 } 1334 1335 int 1336 spdk_fs_delete_file(struct spdk_filesystem *fs, struct spdk_io_channel *_channel, 1337 const char *name) 1338 { 1339 struct spdk_fs_channel *channel = spdk_io_channel_get_ctx(_channel); 1340 struct spdk_fs_request *req; 1341 struct spdk_fs_cb_args *args; 1342 int rc; 1343 1344 req = alloc_fs_request(channel); 1345 assert(req != NULL); 1346 1347 args = &req->args; 1348 args->fs = fs; 1349 args->op.delete.name = name; 1350 args->sem = &channel->sem; 1351 fs->send_request(__fs_delete_file, req); 1352 sem_wait(&channel->sem); 1353 rc = args->rc; 1354 free_fs_request(req); 1355 1356 return rc; 1357 } 1358 1359 spdk_fs_iter 1360 spdk_fs_iter_first(struct spdk_filesystem *fs) 1361 { 1362 struct spdk_file *f; 1363 1364 f = TAILQ_FIRST(&fs->files); 1365 return f; 1366 } 1367 1368 spdk_fs_iter 1369 spdk_fs_iter_next(spdk_fs_iter iter) 1370 { 1371 struct spdk_file *f = iter; 1372 1373 if (f == NULL) { 1374 return NULL; 1375 } 1376 1377 f = TAILQ_NEXT(f, tailq); 1378 return f; 1379 } 1380 1381 const char * 1382 spdk_file_get_name(struct spdk_file *file) 1383 { 1384 return file->name; 1385 } 1386 1387 uint64_t 1388 spdk_file_get_length(struct spdk_file *file) 1389 { 1390 assert(file != NULL); 1391 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "file=%s length=0x%jx\n", file->name, file->length); 1392 return file->length; 1393 } 1394 1395 static void 1396 fs_truncate_complete_cb(void *ctx, int bserrno) 1397 { 1398 struct spdk_fs_request *req = ctx; 1399 struct spdk_fs_cb_args *args = &req->args; 1400 1401 args->fn.file_op(args->arg, bserrno); 1402 free_fs_request(req); 1403 } 1404 1405 static uint64_t 1406 __bytes_to_clusters(uint64_t length, uint64_t cluster_sz) 1407 { 1408 return (length + cluster_sz - 1) / cluster_sz; 1409 } 1410 1411 void 1412 spdk_file_truncate_async(struct spdk_file *file, uint64_t length, 1413 spdk_file_op_complete cb_fn, void *cb_arg) 1414 { 1415 struct spdk_filesystem *fs; 1416 size_t num_clusters; 1417 struct spdk_fs_request *req; 1418 struct spdk_fs_cb_args *args; 1419 1420 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "file=%s old=0x%jx new=0x%jx\n", file->name, file->length, length); 1421 if (length == file->length) { 1422 cb_fn(cb_arg, 0); 1423 return; 1424 } 1425 1426 req = alloc_fs_request(file->fs->md_target.md_fs_channel); 1427 if (req == NULL) { 1428 cb_fn(cb_arg, -ENOMEM); 1429 return; 1430 } 1431 1432 args = &req->args; 1433 args->fn.file_op = cb_fn; 1434 args->arg = cb_arg; 1435 args->file = file; 1436 fs = file->fs; 1437 1438 num_clusters = __bytes_to_clusters(length, fs->bs_opts.cluster_sz); 1439 1440 spdk_blob_resize(file->blob, num_clusters); 1441 spdk_blob_set_xattr(file->blob, "length", &length, sizeof(length)); 1442 1443 file->length = length; 1444 if (file->append_pos > file->length) { 1445 file->append_pos = file->length; 1446 } 1447 1448 spdk_blob_sync_md(file->blob, fs_truncate_complete_cb, args); 1449 } 1450 1451 static void 1452 __truncate(void *arg) 1453 { 1454 struct spdk_fs_request *req = arg; 1455 struct spdk_fs_cb_args *args = &req->args; 1456 1457 spdk_file_truncate_async(args->file, args->op.truncate.length, 1458 args->fn.file_op, args->arg); 1459 } 1460 1461 void 1462 spdk_file_truncate(struct spdk_file *file, struct spdk_io_channel *_channel, 1463 uint64_t length) 1464 { 1465 struct spdk_fs_channel *channel = spdk_io_channel_get_ctx(_channel); 1466 struct spdk_fs_request *req; 1467 struct spdk_fs_cb_args *args; 1468 1469 req = alloc_fs_request(channel); 1470 assert(req != NULL); 1471 1472 args = &req->args; 1473 1474 args->file = file; 1475 args->op.truncate.length = length; 1476 args->fn.file_op = __sem_post; 1477 args->arg = &channel->sem; 1478 1479 channel->send_request(__truncate, req); 1480 sem_wait(&channel->sem); 1481 free_fs_request(req); 1482 } 1483 1484 static void 1485 __rw_done(void *ctx, int bserrno) 1486 { 1487 struct spdk_fs_request *req = ctx; 1488 struct spdk_fs_cb_args *args = &req->args; 1489 1490 spdk_dma_free(args->op.rw.pin_buf); 1491 args->fn.file_op(args->arg, bserrno); 1492 free_fs_request(req); 1493 } 1494 1495 static void 1496 __read_done(void *ctx, int bserrno) 1497 { 1498 struct spdk_fs_request *req = ctx; 1499 struct spdk_fs_cb_args *args = &req->args; 1500 1501 if (args->op.rw.is_read) { 1502 memcpy(args->op.rw.user_buf, 1503 args->op.rw.pin_buf + (args->op.rw.offset & 0xFFF), 1504 args->op.rw.length); 1505 __rw_done(req, 0); 1506 } else { 1507 memcpy(args->op.rw.pin_buf + (args->op.rw.offset & 0xFFF), 1508 args->op.rw.user_buf, 1509 args->op.rw.length); 1510 spdk_blob_io_write(args->file->blob, args->op.rw.channel, 1511 args->op.rw.pin_buf, 1512 args->op.rw.start_page, args->op.rw.num_pages, 1513 __rw_done, req); 1514 } 1515 } 1516 1517 static void 1518 __do_blob_read(void *ctx, int fserrno) 1519 { 1520 struct spdk_fs_request *req = ctx; 1521 struct spdk_fs_cb_args *args = &req->args; 1522 1523 spdk_blob_io_read(args->file->blob, args->op.rw.channel, 1524 args->op.rw.pin_buf, 1525 args->op.rw.start_page, args->op.rw.num_pages, 1526 __read_done, req); 1527 } 1528 1529 static void 1530 __get_page_parameters(struct spdk_file *file, uint64_t offset, uint64_t length, 1531 uint64_t *start_page, uint32_t *page_size, uint64_t *num_pages) 1532 { 1533 uint64_t end_page; 1534 1535 *page_size = spdk_bs_get_page_size(file->fs->bs); 1536 *start_page = offset / *page_size; 1537 end_page = (offset + length - 1) / *page_size; 1538 *num_pages = (end_page - *start_page + 1); 1539 } 1540 1541 static void 1542 __readwrite(struct spdk_file *file, struct spdk_io_channel *_channel, 1543 void *payload, uint64_t offset, uint64_t length, 1544 spdk_file_op_complete cb_fn, void *cb_arg, int is_read) 1545 { 1546 struct spdk_fs_request *req; 1547 struct spdk_fs_cb_args *args; 1548 struct spdk_fs_channel *channel = spdk_io_channel_get_ctx(_channel); 1549 uint64_t start_page, num_pages, pin_buf_length; 1550 uint32_t page_size; 1551 1552 if (is_read && offset + length > file->length) { 1553 cb_fn(cb_arg, -EINVAL); 1554 return; 1555 } 1556 1557 req = alloc_fs_request(channel); 1558 if (req == NULL) { 1559 cb_fn(cb_arg, -ENOMEM); 1560 return; 1561 } 1562 1563 args = &req->args; 1564 args->fn.file_op = cb_fn; 1565 args->arg = cb_arg; 1566 args->file = file; 1567 args->op.rw.channel = channel->bs_channel; 1568 args->op.rw.user_buf = payload; 1569 args->op.rw.is_read = is_read; 1570 args->op.rw.offset = offset; 1571 args->op.rw.length = length; 1572 1573 __get_page_parameters(file, offset, length, &start_page, &page_size, &num_pages); 1574 pin_buf_length = num_pages * page_size; 1575 args->op.rw.pin_buf = spdk_dma_malloc(pin_buf_length, 4096, NULL); 1576 1577 args->op.rw.start_page = start_page; 1578 args->op.rw.num_pages = num_pages; 1579 1580 if (!is_read && file->length < offset + length) { 1581 spdk_file_truncate_async(file, offset + length, __do_blob_read, req); 1582 } else { 1583 __do_blob_read(req, 0); 1584 } 1585 } 1586 1587 void 1588 spdk_file_write_async(struct spdk_file *file, struct spdk_io_channel *channel, 1589 void *payload, uint64_t offset, uint64_t length, 1590 spdk_file_op_complete cb_fn, void *cb_arg) 1591 { 1592 __readwrite(file, channel, payload, offset, length, cb_fn, cb_arg, 0); 1593 } 1594 1595 void 1596 spdk_file_read_async(struct spdk_file *file, struct spdk_io_channel *channel, 1597 void *payload, uint64_t offset, uint64_t length, 1598 spdk_file_op_complete cb_fn, void *cb_arg) 1599 { 1600 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "file=%s offset=%jx length=%jx\n", 1601 file->name, offset, length); 1602 __readwrite(file, channel, payload, offset, length, cb_fn, cb_arg, 1); 1603 } 1604 1605 struct spdk_io_channel * 1606 spdk_fs_alloc_io_channel(struct spdk_filesystem *fs) 1607 { 1608 struct spdk_io_channel *io_channel; 1609 struct spdk_fs_channel *fs_channel; 1610 1611 io_channel = spdk_get_io_channel(&fs->io_target); 1612 fs_channel = spdk_io_channel_get_ctx(io_channel); 1613 fs_channel->bs_channel = spdk_bs_alloc_io_channel(fs->bs); 1614 fs_channel->send_request = __send_request_direct; 1615 1616 return io_channel; 1617 } 1618 1619 struct spdk_io_channel * 1620 spdk_fs_alloc_io_channel_sync(struct spdk_filesystem *fs) 1621 { 1622 struct spdk_io_channel *io_channel; 1623 struct spdk_fs_channel *fs_channel; 1624 1625 io_channel = spdk_get_io_channel(&fs->io_target); 1626 fs_channel = spdk_io_channel_get_ctx(io_channel); 1627 fs_channel->send_request = fs->send_request; 1628 fs_channel->sync = 1; 1629 pthread_spin_init(&fs_channel->lock, 0); 1630 1631 return io_channel; 1632 } 1633 1634 void 1635 spdk_fs_free_io_channel(struct spdk_io_channel *channel) 1636 { 1637 spdk_put_io_channel(channel); 1638 } 1639 1640 void 1641 spdk_fs_set_cache_size(uint64_t size_in_mb) 1642 { 1643 g_fs_cache_size = size_in_mb * 1024 * 1024; 1644 } 1645 1646 uint64_t 1647 spdk_fs_get_cache_size(void) 1648 { 1649 return g_fs_cache_size / (1024 * 1024); 1650 } 1651 1652 static void __file_flush(void *_args); 1653 1654 static void * 1655 alloc_cache_memory_buffer(struct spdk_file *context) 1656 { 1657 struct spdk_file *file; 1658 void *buf; 1659 1660 buf = spdk_mempool_get(g_cache_pool); 1661 if (buf != NULL) { 1662 return buf; 1663 } 1664 1665 pthread_spin_lock(&g_caches_lock); 1666 TAILQ_FOREACH(file, &g_caches, cache_tailq) { 1667 if (!file->open_for_writing && 1668 file->priority == SPDK_FILE_PRIORITY_LOW && 1669 file != context) { 1670 break; 1671 } 1672 } 1673 pthread_spin_unlock(&g_caches_lock); 1674 if (file != NULL) { 1675 cache_free_buffers(file); 1676 buf = spdk_mempool_get(g_cache_pool); 1677 if (buf != NULL) { 1678 return buf; 1679 } 1680 } 1681 1682 pthread_spin_lock(&g_caches_lock); 1683 TAILQ_FOREACH(file, &g_caches, cache_tailq) { 1684 if (!file->open_for_writing && file != context) { 1685 break; 1686 } 1687 } 1688 pthread_spin_unlock(&g_caches_lock); 1689 if (file != NULL) { 1690 cache_free_buffers(file); 1691 buf = spdk_mempool_get(g_cache_pool); 1692 if (buf != NULL) { 1693 return buf; 1694 } 1695 } 1696 1697 pthread_spin_lock(&g_caches_lock); 1698 TAILQ_FOREACH(file, &g_caches, cache_tailq) { 1699 if (file != context) { 1700 break; 1701 } 1702 } 1703 pthread_spin_unlock(&g_caches_lock); 1704 if (file != NULL) { 1705 cache_free_buffers(file); 1706 buf = spdk_mempool_get(g_cache_pool); 1707 if (buf != NULL) { 1708 return buf; 1709 } 1710 } 1711 1712 return NULL; 1713 } 1714 1715 static struct cache_buffer * 1716 cache_insert_buffer(struct spdk_file *file, uint64_t offset) 1717 { 1718 struct cache_buffer *buf; 1719 int count = 0; 1720 1721 buf = calloc(1, sizeof(*buf)); 1722 if (buf == NULL) { 1723 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "calloc failed\n"); 1724 return NULL; 1725 } 1726 1727 buf->buf = alloc_cache_memory_buffer(file); 1728 while (buf->buf == NULL) { 1729 /* 1730 * TODO: alloc_cache_memory_buffer() should eventually free 1731 * some buffers. Need a more sophisticated check here, instead 1732 * of just bailing if 100 tries does not result in getting a 1733 * free buffer. This will involve using the sync channel's 1734 * semaphore to block until a buffer becomes available. 1735 */ 1736 if (count++ == 100) { 1737 SPDK_ERRLOG("could not allocate cache buffer\n"); 1738 assert(false); 1739 free(buf); 1740 return NULL; 1741 } 1742 buf->buf = alloc_cache_memory_buffer(file); 1743 } 1744 1745 buf->buf_size = CACHE_BUFFER_SIZE; 1746 buf->offset = offset; 1747 1748 pthread_spin_lock(&g_caches_lock); 1749 if (file->tree->present_mask == 0) { 1750 TAILQ_INSERT_TAIL(&g_caches, file, cache_tailq); 1751 } 1752 file->tree = spdk_tree_insert_buffer(file->tree, buf); 1753 pthread_spin_unlock(&g_caches_lock); 1754 1755 return buf; 1756 } 1757 1758 static struct cache_buffer * 1759 cache_append_buffer(struct spdk_file *file) 1760 { 1761 struct cache_buffer *last; 1762 1763 assert(file->last == NULL || file->last->bytes_filled == file->last->buf_size); 1764 assert((file->append_pos % CACHE_BUFFER_SIZE) == 0); 1765 1766 last = cache_insert_buffer(file, file->append_pos); 1767 if (last == NULL) { 1768 SPDK_DEBUGLOG(SPDK_LOG_BLOBFS, "cache_insert_buffer failed\n"); 1769 return NULL; 1770 } 1771 1772 file->last = last; 1773 1774 return last; 1775 } 1776 1777 static void 1778 __wake_caller(struct spdk_fs_cb_args *args) 1779 { 1780 sem_post(args->sem); 1781 } 1782 1783 static void __check_sync_reqs(struct spdk_file *file); 1784 1785 static void 1786 __file_cache_finish_sync(struct spdk_file *file) 1787 { 1788 struct spdk_fs_request *sync_req; 1789 struct spdk_fs_cb_args *sync_args; 1790 1791 pthread_spin_lock(&file->lock); 1792 sync_req = TAILQ_FIRST(&file->sync_requests); 1793 sync_args = &sync_req->args; 1794 assert(sync_args->op.sync.offset <= file->length_flushed); 1795 BLOBFS_TRACE(file, "sync done offset=%jx\n", sync_args->op.sync.offset); 1796 TAILQ_REMOVE(&file->sync_requests, sync_req, args.op.sync.tailq); 1797 pthread_spin_unlock(&file->lock); 1798 1799 sync_args->fn.file_op(sync_args->arg, 0); 1800 __check_sync_reqs(file); 1801 1802 pthread_spin_lock(&file->lock); 1803 free_fs_request(sync_req); 1804 pthread_spin_unlock(&file->lock); 1805 } 1806 1807 static void 1808 __file_cache_finish_sync_bs_cb(void *ctx, int bserrno) 1809 { 1810 struct spdk_file *file = ctx; 1811 1812 __file_cache_finish_sync(file); 1813 } 1814 1815 static void 1816 __free_args(struct spdk_fs_cb_args *args) 1817 { 1818 struct spdk_fs_request *req; 1819 1820 if (!args->from_request) { 1821 free(args); 1822 } else { 1823 /* Depends on args being at the start of the spdk_fs_request structure. */ 1824 req = (struct spdk_fs_request *)args; 1825 free_fs_request(req); 1826 } 1827 } 1828 1829 static void 1830 __check_sync_reqs(struct spdk_file *file) 1831 { 1832 struct spdk_fs_request *sync_req; 1833 1834 pthread_spin_lock(&file->lock); 1835 1836 TAILQ_FOREACH(sync_req, &file->sync_requests, args.op.sync.tailq) { 1837 if (sync_req->args.op.sync.offset <= file->length_flushed) { 1838 break; 1839 } 1840 } 1841 1842 if (sync_req != NULL && !sync_req->args.op.sync.xattr_in_progress) { 1843 BLOBFS_TRACE(file, "set xattr length 0x%jx\n", file->length_flushed); 1844 sync_req->args.op.sync.xattr_in_progress = true; 1845 spdk_blob_set_xattr(file->blob, "length", &file->length_flushed, 1846 sizeof(file->length_flushed)); 1847 1848 pthread_spin_unlock(&file->lock); 1849 spdk_blob_sync_md(file->blob, __file_cache_finish_sync_bs_cb, file); 1850 } else { 1851 pthread_spin_unlock(&file->lock); 1852 } 1853 } 1854 1855 static void 1856 __file_flush_done(void *arg, int bserrno) 1857 { 1858 struct spdk_fs_cb_args *args = arg; 1859 struct spdk_file *file = args->file; 1860 struct cache_buffer *next = args->op.flush.cache_buffer; 1861 1862 BLOBFS_TRACE(file, "length=%jx\n", args->op.flush.length); 1863 1864 pthread_spin_lock(&file->lock); 1865 next->in_progress = false; 1866 next->bytes_flushed += args->op.flush.length; 1867 file->length_flushed += args->op.flush.length; 1868 if (file->length_flushed > file->length) { 1869 file->length = file->length_flushed; 1870 } 1871 if (next->bytes_flushed == next->buf_size) { 1872 BLOBFS_TRACE(file, "write buffer fully flushed 0x%jx\n", file->length_flushed); 1873 next = spdk_tree_find_buffer(file->tree, file->length_flushed); 1874 } 1875 1876 /* 1877 * Assert that there is no cached data that extends past the end of the underlying 1878 * blob. 1879 */ 1880 assert(next == NULL || next->offset < __file_get_blob_size(file) || 1881 next->bytes_filled == 0); 1882 1883 pthread_spin_unlock(&file->lock); 1884 1885 __check_sync_reqs(file); 1886 1887 __file_flush(args); 1888 } 1889 1890 static void 1891 __file_flush(void *_args) 1892 { 1893 struct spdk_fs_cb_args *args = _args; 1894 struct spdk_file *file = args->file; 1895 struct cache_buffer *next; 1896 uint64_t offset, length, start_page, num_pages; 1897 uint32_t page_size; 1898 1899 pthread_spin_lock(&file->lock); 1900 next = spdk_tree_find_buffer(file->tree, file->length_flushed); 1901 if (next == NULL || next->in_progress) { 1902 /* 1903 * There is either no data to flush, or a flush I/O is already in 1904 * progress. So return immediately - if a flush I/O is in 1905 * progress we will flush more data after that is completed. 1906 */ 1907 __free_args(args); 1908 pthread_spin_unlock(&file->lock); 1909 return; 1910 } 1911 1912 offset = next->offset + next->bytes_flushed; 1913 length = next->bytes_filled - next->bytes_flushed; 1914 if (length == 0) { 1915 __free_args(args); 1916 pthread_spin_unlock(&file->lock); 1917 return; 1918 } 1919 args->op.flush.length = length; 1920 args->op.flush.cache_buffer = next; 1921 1922 __get_page_parameters(file, offset, length, &start_page, &page_size, &num_pages); 1923 1924 next->in_progress = true; 1925 BLOBFS_TRACE(file, "offset=%jx length=%jx page start=%jx num=%jx\n", 1926 offset, length, start_page, num_pages); 1927 pthread_spin_unlock(&file->lock); 1928 spdk_blob_io_write(file->blob, file->fs->sync_target.sync_fs_channel->bs_channel, 1929 next->buf + (start_page * page_size) - next->offset, 1930 start_page, num_pages, __file_flush_done, args); 1931 } 1932 1933 static void 1934 __file_extend_done(void *arg, int bserrno) 1935 { 1936 struct spdk_fs_cb_args *args = arg; 1937 1938 __wake_caller(args); 1939 } 1940 1941 static void 1942 __file_extend_blob(void *_args) 1943 { 1944 struct spdk_fs_cb_args *args = _args; 1945 struct spdk_file *file = args->file; 1946 1947 spdk_blob_resize(file->blob, args->op.resize.num_clusters); 1948 1949 spdk_blob_sync_md(file->blob, __file_extend_done, args); 1950 } 1951 1952 static void 1953 __rw_from_file_done(void *arg, int bserrno) 1954 { 1955 struct spdk_fs_cb_args *args = arg; 1956 1957 __wake_caller(args); 1958 __free_args(args); 1959 } 1960 1961 static void 1962 __rw_from_file(void *_args) 1963 { 1964 struct spdk_fs_cb_args *args = _args; 1965 struct spdk_file *file = args->file; 1966 1967 if (args->op.rw.is_read) { 1968 spdk_file_read_async(file, file->fs->sync_target.sync_io_channel, args->op.rw.user_buf, 1969 args->op.rw.offset, args->op.rw.length, 1970 __rw_from_file_done, args); 1971 } else { 1972 spdk_file_write_async(file, file->fs->sync_target.sync_io_channel, args->op.rw.user_buf, 1973 args->op.rw.offset, args->op.rw.length, 1974 __rw_from_file_done, args); 1975 } 1976 } 1977 1978 static int 1979 __send_rw_from_file(struct spdk_file *file, sem_t *sem, void *payload, 1980 uint64_t offset, uint64_t length, bool is_read) 1981 { 1982 struct spdk_fs_cb_args *args; 1983 1984 args = calloc(1, sizeof(*args)); 1985 if (args == NULL) { 1986 sem_post(sem); 1987 return -ENOMEM; 1988 } 1989 1990 args->file = file; 1991 args->sem = sem; 1992 args->op.rw.user_buf = payload; 1993 args->op.rw.offset = offset; 1994 args->op.rw.length = length; 1995 args->op.rw.is_read = is_read; 1996 file->fs->send_request(__rw_from_file, args); 1997 return 0; 1998 } 1999 2000 int 2001 spdk_file_write(struct spdk_file *file, struct spdk_io_channel *_channel, 2002 void *payload, uint64_t offset, uint64_t length) 2003 { 2004 struct spdk_fs_channel *channel = spdk_io_channel_get_ctx(_channel); 2005 struct spdk_fs_cb_args *args; 2006 uint64_t rem_length, copy, blob_size, cluster_sz; 2007 uint32_t cache_buffers_filled = 0; 2008 uint8_t *cur_payload; 2009 struct cache_buffer *last; 2010 2011 BLOBFS_TRACE_RW(file, "offset=%jx length=%jx\n", offset, length); 2012 2013 if (length == 0) { 2014 return 0; 2015 } 2016 2017 if (offset != file->append_pos) { 2018 BLOBFS_TRACE(file, " error offset=%jx append_pos=%jx\n", offset, file->append_pos); 2019 return -EINVAL; 2020 } 2021 2022 pthread_spin_lock(&file->lock); 2023 file->open_for_writing = true; 2024 2025 if (file->last == NULL) { 2026 if (file->append_pos % CACHE_BUFFER_SIZE == 0) { 2027 cache_append_buffer(file); 2028 } else { 2029 int rc; 2030 2031 file->append_pos += length; 2032 pthread_spin_unlock(&file->lock); 2033 rc = __send_rw_from_file(file, &channel->sem, payload, 2034 offset, length, false); 2035 sem_wait(&channel->sem); 2036 return rc; 2037 } 2038 } 2039 2040 blob_size = __file_get_blob_size(file); 2041 2042 if ((offset + length) > blob_size) { 2043 struct spdk_fs_cb_args extend_args = {}; 2044 2045 cluster_sz = file->fs->bs_opts.cluster_sz; 2046 extend_args.sem = &channel->sem; 2047 extend_args.op.resize.num_clusters = __bytes_to_clusters((offset + length), cluster_sz); 2048 extend_args.file = file; 2049 BLOBFS_TRACE(file, "start resize to %u clusters\n", extend_args.op.resize.num_clusters); 2050 pthread_spin_unlock(&file->lock); 2051 file->fs->send_request(__file_extend_blob, &extend_args); 2052 sem_wait(&channel->sem); 2053 } 2054 2055 last = file->last; 2056 rem_length = length; 2057 cur_payload = payload; 2058 while (rem_length > 0) { 2059 copy = last->buf_size - last->bytes_filled; 2060 if (copy > rem_length) { 2061 copy = rem_length; 2062 } 2063 BLOBFS_TRACE_RW(file, " fill offset=%jx length=%jx\n", file->append_pos, copy); 2064 memcpy(&last->buf[last->bytes_filled], cur_payload, copy); 2065 file->append_pos += copy; 2066 if (file->length < file->append_pos) { 2067 file->length = file->append_pos; 2068 } 2069 cur_payload += copy; 2070 last->bytes_filled += copy; 2071 rem_length -= copy; 2072 if (last->bytes_filled == last->buf_size) { 2073 cache_buffers_filled++; 2074 last = cache_append_buffer(file); 2075 if (last == NULL) { 2076 BLOBFS_TRACE(file, "nomem\n"); 2077 pthread_spin_unlock(&file->lock); 2078 return -ENOMEM; 2079 } 2080 } 2081 } 2082 2083 pthread_spin_unlock(&file->lock); 2084 2085 if (cache_buffers_filled == 0) { 2086 return 0; 2087 } 2088 2089 args = calloc(1, sizeof(*args)); 2090 if (args == NULL) { 2091 return -ENOMEM; 2092 } 2093 2094 args->file = file; 2095 file->fs->send_request(__file_flush, args); 2096 return 0; 2097 } 2098 2099 static void 2100 __readahead_done(void *arg, int bserrno) 2101 { 2102 struct spdk_fs_cb_args *args = arg; 2103 struct cache_buffer *cache_buffer = args->op.readahead.cache_buffer; 2104 struct spdk_file *file = args->file; 2105 2106 BLOBFS_TRACE(file, "offset=%jx\n", cache_buffer->offset); 2107 2108 pthread_spin_lock(&file->lock); 2109 cache_buffer->bytes_filled = args->op.readahead.length; 2110 cache_buffer->bytes_flushed = args->op.readahead.length; 2111 cache_buffer->in_progress = false; 2112 pthread_spin_unlock(&file->lock); 2113 2114 __free_args(args); 2115 } 2116 2117 static void 2118 __readahead(void *_args) 2119 { 2120 struct spdk_fs_cb_args *args = _args; 2121 struct spdk_file *file = args->file; 2122 uint64_t offset, length, start_page, num_pages; 2123 uint32_t page_size; 2124 2125 offset = args->op.readahead.offset; 2126 length = args->op.readahead.length; 2127 assert(length > 0); 2128 2129 __get_page_parameters(file, offset, length, &start_page, &page_size, &num_pages); 2130 2131 BLOBFS_TRACE(file, "offset=%jx length=%jx page start=%jx num=%jx\n", 2132 offset, length, start_page, num_pages); 2133 spdk_blob_io_read(file->blob, file->fs->sync_target.sync_fs_channel->bs_channel, 2134 args->op.readahead.cache_buffer->buf, 2135 start_page, num_pages, __readahead_done, args); 2136 } 2137 2138 static uint64_t 2139 __next_cache_buffer_offset(uint64_t offset) 2140 { 2141 return (offset + CACHE_BUFFER_SIZE) & ~(CACHE_TREE_LEVEL_MASK(0)); 2142 } 2143 2144 static void 2145 check_readahead(struct spdk_file *file, uint64_t offset) 2146 { 2147 struct spdk_fs_cb_args *args; 2148 2149 offset = __next_cache_buffer_offset(offset); 2150 if (spdk_tree_find_buffer(file->tree, offset) != NULL || file->length <= offset) { 2151 return; 2152 } 2153 2154 args = calloc(1, sizeof(*args)); 2155 if (args == NULL) { 2156 return; 2157 } 2158 2159 BLOBFS_TRACE(file, "offset=%jx\n", offset); 2160 2161 args->file = file; 2162 args->op.readahead.offset = offset; 2163 args->op.readahead.cache_buffer = cache_insert_buffer(file, offset); 2164 args->op.readahead.cache_buffer->in_progress = true; 2165 if (file->length < (offset + CACHE_BUFFER_SIZE)) { 2166 args->op.readahead.length = file->length & (CACHE_BUFFER_SIZE - 1); 2167 } else { 2168 args->op.readahead.length = CACHE_BUFFER_SIZE; 2169 } 2170 file->fs->send_request(__readahead, args); 2171 } 2172 2173 static int 2174 __file_read(struct spdk_file *file, void *payload, uint64_t offset, uint64_t length, sem_t *sem) 2175 { 2176 struct cache_buffer *buf; 2177 int rc; 2178 2179 buf = spdk_tree_find_filled_buffer(file->tree, offset); 2180 if (buf == NULL) { 2181 pthread_spin_unlock(&file->lock); 2182 rc = __send_rw_from_file(file, sem, payload, offset, length, true); 2183 pthread_spin_lock(&file->lock); 2184 return rc; 2185 } 2186 2187 if ((offset + length) > (buf->offset + buf->bytes_filled)) { 2188 length = buf->offset + buf->bytes_filled - offset; 2189 } 2190 BLOBFS_TRACE(file, "read %p offset=%ju length=%ju\n", payload, offset, length); 2191 memcpy(payload, &buf->buf[offset - buf->offset], length); 2192 if ((offset + length) % CACHE_BUFFER_SIZE == 0) { 2193 pthread_spin_lock(&g_caches_lock); 2194 spdk_tree_remove_buffer(file->tree, buf); 2195 if (file->tree->present_mask == 0) { 2196 TAILQ_REMOVE(&g_caches, file, cache_tailq); 2197 } 2198 pthread_spin_unlock(&g_caches_lock); 2199 } 2200 2201 sem_post(sem); 2202 return 0; 2203 } 2204 2205 int64_t 2206 spdk_file_read(struct spdk_file *file, struct spdk_io_channel *_channel, 2207 void *payload, uint64_t offset, uint64_t length) 2208 { 2209 struct spdk_fs_channel *channel = spdk_io_channel_get_ctx(_channel); 2210 uint64_t final_offset, final_length; 2211 uint32_t sub_reads = 0; 2212 int rc = 0; 2213 2214 pthread_spin_lock(&file->lock); 2215 2216 BLOBFS_TRACE_RW(file, "offset=%ju length=%ju\n", offset, length); 2217 2218 file->open_for_writing = false; 2219 2220 if (length == 0 || offset >= file->append_pos) { 2221 pthread_spin_unlock(&file->lock); 2222 return 0; 2223 } 2224 2225 if (offset + length > file->append_pos) { 2226 length = file->append_pos - offset; 2227 } 2228 2229 if (offset != file->next_seq_offset) { 2230 file->seq_byte_count = 0; 2231 } 2232 file->seq_byte_count += length; 2233 file->next_seq_offset = offset + length; 2234 if (file->seq_byte_count >= CACHE_READAHEAD_THRESHOLD) { 2235 check_readahead(file, offset); 2236 check_readahead(file, offset + CACHE_BUFFER_SIZE); 2237 } 2238 2239 final_length = 0; 2240 final_offset = offset + length; 2241 while (offset < final_offset) { 2242 length = NEXT_CACHE_BUFFER_OFFSET(offset) - offset; 2243 if (length > (final_offset - offset)) { 2244 length = final_offset - offset; 2245 } 2246 rc = __file_read(file, payload, offset, length, &channel->sem); 2247 if (rc == 0) { 2248 final_length += length; 2249 } else { 2250 break; 2251 } 2252 payload += length; 2253 offset += length; 2254 sub_reads++; 2255 } 2256 pthread_spin_unlock(&file->lock); 2257 while (sub_reads-- > 0) { 2258 sem_wait(&channel->sem); 2259 } 2260 if (rc == 0) { 2261 return final_length; 2262 } else { 2263 return rc; 2264 } 2265 } 2266 2267 static void 2268 _file_sync(struct spdk_file *file, struct spdk_fs_channel *channel, 2269 spdk_file_op_complete cb_fn, void *cb_arg) 2270 { 2271 struct spdk_fs_request *sync_req; 2272 struct spdk_fs_request *flush_req; 2273 struct spdk_fs_cb_args *sync_args; 2274 struct spdk_fs_cb_args *flush_args; 2275 2276 BLOBFS_TRACE(file, "offset=%jx\n", file->append_pos); 2277 2278 pthread_spin_lock(&file->lock); 2279 if (file->append_pos <= file->length_flushed || file->last == NULL) { 2280 BLOBFS_TRACE(file, "done - no data to flush\n"); 2281 pthread_spin_unlock(&file->lock); 2282 cb_fn(cb_arg, 0); 2283 return; 2284 } 2285 2286 sync_req = alloc_fs_request(channel); 2287 assert(sync_req != NULL); 2288 sync_args = &sync_req->args; 2289 2290 flush_req = alloc_fs_request(channel); 2291 assert(flush_req != NULL); 2292 flush_args = &flush_req->args; 2293 2294 sync_args->file = file; 2295 sync_args->fn.file_op = cb_fn; 2296 sync_args->arg = cb_arg; 2297 sync_args->op.sync.offset = file->append_pos; 2298 sync_args->op.sync.xattr_in_progress = false; 2299 TAILQ_INSERT_TAIL(&file->sync_requests, sync_req, args.op.sync.tailq); 2300 pthread_spin_unlock(&file->lock); 2301 2302 flush_args->file = file; 2303 channel->send_request(__file_flush, flush_args); 2304 } 2305 2306 int 2307 spdk_file_sync(struct spdk_file *file, struct spdk_io_channel *_channel) 2308 { 2309 struct spdk_fs_channel *channel = spdk_io_channel_get_ctx(_channel); 2310 2311 _file_sync(file, channel, __sem_post, &channel->sem); 2312 sem_wait(&channel->sem); 2313 2314 return 0; 2315 } 2316 2317 void 2318 spdk_file_sync_async(struct spdk_file *file, struct spdk_io_channel *_channel, 2319 spdk_file_op_complete cb_fn, void *cb_arg) 2320 { 2321 struct spdk_fs_channel *channel = spdk_io_channel_get_ctx(_channel); 2322 2323 _file_sync(file, channel, cb_fn, cb_arg); 2324 } 2325 2326 void 2327 spdk_file_set_priority(struct spdk_file *file, uint32_t priority) 2328 { 2329 BLOBFS_TRACE(file, "priority=%u\n", priority); 2330 file->priority = priority; 2331 2332 } 2333 2334 /* 2335 * Close routines 2336 */ 2337 2338 static void 2339 __file_close_async_done(void *ctx, int bserrno) 2340 { 2341 struct spdk_fs_request *req = ctx; 2342 struct spdk_fs_cb_args *args = &req->args; 2343 struct spdk_file *file = args->file; 2344 2345 if (file->is_deleted) { 2346 spdk_fs_delete_file_async(file->fs, file->name, blob_delete_cb, ctx); 2347 return; 2348 } 2349 2350 args->fn.file_op(args->arg, bserrno); 2351 free_fs_request(req); 2352 } 2353 2354 static void 2355 __file_close_async(struct spdk_file *file, struct spdk_fs_request *req) 2356 { 2357 struct spdk_blob *blob; 2358 2359 pthread_spin_lock(&file->lock); 2360 if (file->ref_count == 0) { 2361 pthread_spin_unlock(&file->lock); 2362 __file_close_async_done(req, -EBADF); 2363 return; 2364 } 2365 2366 file->ref_count--; 2367 if (file->ref_count > 0) { 2368 pthread_spin_unlock(&file->lock); 2369 req->args.fn.file_op(req->args.arg, 0); 2370 free_fs_request(req); 2371 return; 2372 } 2373 2374 pthread_spin_unlock(&file->lock); 2375 2376 blob = file->blob; 2377 file->blob = NULL; 2378 spdk_blob_close(blob, __file_close_async_done, req); 2379 } 2380 2381 static void 2382 __file_close_async__sync_done(void *arg, int fserrno) 2383 { 2384 struct spdk_fs_request *req = arg; 2385 struct spdk_fs_cb_args *args = &req->args; 2386 2387 __file_close_async(args->file, req); 2388 } 2389 2390 void 2391 spdk_file_close_async(struct spdk_file *file, spdk_file_op_complete cb_fn, void *cb_arg) 2392 { 2393 struct spdk_fs_request *req; 2394 struct spdk_fs_cb_args *args; 2395 2396 req = alloc_fs_request(file->fs->md_target.md_fs_channel); 2397 if (req == NULL) { 2398 cb_fn(cb_arg, -ENOMEM); 2399 return; 2400 } 2401 2402 args = &req->args; 2403 args->file = file; 2404 args->fn.file_op = cb_fn; 2405 args->arg = cb_arg; 2406 2407 spdk_file_sync_async(file, file->fs->md_target.md_io_channel, __file_close_async__sync_done, req); 2408 } 2409 2410 static void 2411 __file_close_done(void *arg, int fserrno) 2412 { 2413 struct spdk_fs_cb_args *args = arg; 2414 2415 args->rc = fserrno; 2416 sem_post(args->sem); 2417 } 2418 2419 static void 2420 __file_close(void *arg) 2421 { 2422 struct spdk_fs_request *req = arg; 2423 struct spdk_fs_cb_args *args = &req->args; 2424 struct spdk_file *file = args->file; 2425 2426 __file_close_async(file, req); 2427 } 2428 2429 int 2430 spdk_file_close(struct spdk_file *file, struct spdk_io_channel *_channel) 2431 { 2432 struct spdk_fs_channel *channel = spdk_io_channel_get_ctx(_channel); 2433 struct spdk_fs_request *req; 2434 struct spdk_fs_cb_args *args; 2435 2436 req = alloc_fs_request(channel); 2437 assert(req != NULL); 2438 2439 args = &req->args; 2440 2441 spdk_file_sync(file, _channel); 2442 BLOBFS_TRACE(file, "name=%s\n", file->name); 2443 args->file = file; 2444 args->sem = &channel->sem; 2445 args->fn.file_op = __file_close_done; 2446 args->arg = req; 2447 channel->send_request(__file_close, req); 2448 sem_wait(&channel->sem); 2449 2450 return args->rc; 2451 } 2452 2453 static void 2454 cache_free_buffers(struct spdk_file *file) 2455 { 2456 BLOBFS_TRACE(file, "free=%s\n", file->name); 2457 pthread_spin_lock(&file->lock); 2458 pthread_spin_lock(&g_caches_lock); 2459 if (file->tree->present_mask == 0) { 2460 pthread_spin_unlock(&g_caches_lock); 2461 pthread_spin_unlock(&file->lock); 2462 return; 2463 } 2464 spdk_tree_free_buffers(file->tree); 2465 2466 TAILQ_REMOVE(&g_caches, file, cache_tailq); 2467 /* If not freed, put it in the end of the queue */ 2468 if (file->tree->present_mask != 0) { 2469 TAILQ_INSERT_TAIL(&g_caches, file, cache_tailq); 2470 } 2471 file->last = NULL; 2472 pthread_spin_unlock(&g_caches_lock); 2473 pthread_spin_unlock(&file->lock); 2474 } 2475 2476 SPDK_LOG_REGISTER_COMPONENT("blobfs", SPDK_LOG_BLOBFS) 2477 SPDK_LOG_REGISTER_COMPONENT("blobfs_rw", SPDK_LOG_BLOBFS_RW) 2478