Skip to content

Support non-contiguous multi-extent files on the CPU/P2P read path #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
karlowich merged 2 commits into from
Jul 8, 2026
Merged

Support non-contiguous multi-extent files on the CPU/P2P read path #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9197902
refactor(io): support non-contiguous multi-extent files
karlowich Jul 8, 2026
78897c3
fix(io): validate iosize against LBA size on the CPU/P2P path
karlowich Jul 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions include/fil_io.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,6 @@ struct fil_data {
uint64_t index;
};

struct fil_cpu_io {
uint64_t *slbas;
uint64_t *elbas;
};

struct fil_file_io {
char prefix[PATH_MAX];
char path[PATH_MAX];
Expand Down Expand Up @@ -49,6 +44,11 @@ void
gpu_io_launch(struct xnvme_cuda_queue **queues_dev, uint32_t n_queues,
struct xnvme_spec_cmd *cmds_dev, uint32_t n_io, uint32_t queue_depth);

struct xnvme_cmd_ctx;

void
fil_io_cb(struct xnvme_cmd_ctx *ctx, void *cb_arg);

int
fil_cpu_submit(struct fil_iter *iter);

Expand Down
2 changes: 1 addition & 1 deletion include/fil_iter.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,8 @@ struct fil_dev {
struct fil_gpu_io gpu_io;
struct xal *xal;
struct xal_inode *root_inode;
struct fil_cpu_io *cpu_io;
struct fil_file_io *file_io;
uint32_t io_errors; ///< Reads that failed in the current batch (CPU/P2P path)
const char *data_dir;
void **buffers;
uint64_t buf;
Expand Down
279 changes: 132 additions & 147 deletions src/io.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,125 +20,66 @@
#include <cuda_runtime.h>
#include <cufile.h>

struct _range {
uint64_t slba;
uint64_t elba;
void *dbuf;
};

struct _work {
uint32_t opc;
uint32_t nlb;
uint64_t nbytes;

uint32_t n_ranges;
uint32_t cur_range;

struct _range *ranges;
uint32_t errors;
};

static int
_submit(struct _work *work, struct xnvme_cmd_ctx *ctx)
/**
* Queue completion callback: reap a finished read, tally any failure into the
* error counter passed at queue setup, and release the context. Registered once
* per queue in _xnvme_setup.
*/
void
fil_io_cb(struct xnvme_cmd_ctx *ctx, void *cb_arg)
{
int err;
uint32_t *errors = cb_arg;

ctx->cmd.common.opcode = work->opc;
ctx->cmd.common.nsid = xnvme_dev_get_nsid(ctx->dev);
if (work->ranges[work->cur_range].slba > work->ranges[work->cur_range].elba) {
work->cur_range += 1;
}
if (work->cur_range >= work->n_ranges) {
xnvme_queue_put_cmd_ctx(ctx->async.queue, ctx);
return 0;
}

ctx->cmd.nvm.slba = work->ranges[work->cur_range].slba;
ctx->cmd.nvm.nlb = work->nlb;

retry:
err = xnvme_cmd_pass(ctx, work->ranges[work->cur_range].dbuf, work->nbytes, NULL, 0);
if (err == -EBUSY || err == -EAGAIN) {
xnvme_queue_poke(ctx->async.queue, 0);
goto retry;
}
if (err) {
printf("Failed to submit, err: %d\n", err);
xnvme_queue_put_cmd_ctx(ctx->async.queue, ctx);
return err;
}
work->ranges[work->cur_range].slba += (work->nlb + 1);
work->ranges[work->cur_range].dbuf =
(uint8_t *)work->ranges[work->cur_range].dbuf + work->nbytes;
return 0;
}

static void
_cb_fn(struct xnvme_cmd_ctx *ctx, void *cb_arg)
{
struct _work *work = cb_arg;
if (xnvme_cmd_ctx_cpl_status(ctx)) {
xnvme_cmd_ctx_pr(ctx, XNVME_PR_DEF);
work->errors++;
xnvme_queue_put_cmd_ctx(ctx->async.queue, ctx);
return;
(*errors)++;
}

if (work->cur_range >= work->n_ranges) {
xnvme_queue_put_cmd_ctx(ctx->async.queue, ctx);
return;
}
_submit(work, ctx);
xnvme_queue_put_cmd_ctx(ctx->async.queue, ctx);
}

/**
* Read a physically-contiguous run of 'nblocks' device blocks starting at
* 'slba' into '*dbuf', split into commands of at most 'io_nblocks' blocks (the
* configured iosize). Advances '*dbuf' past the bytes read. Completions are
* reaped asynchronously by fil_io_cb; when the queue is full 'poke' drains it.
*/
static int
_io_range_submit(struct xnvme_queue *queue, uint32_t opc, uint64_t *slbas, uint64_t *elbas,
uint32_t nlb, uint64_t nbytes, void **dbufs, uint32_t n_ranges)
_submit_run(struct xnvme_queue *queue, uint32_t nsid, uint64_t slba, uint64_t nblocks,
uint32_t io_nblocks, uint64_t blocksize, void **dbuf)
{
struct _work work = {0};
struct _range ranges[n_ranges];
struct _range *range;
uint32_t n_blocks;
int err, capacity;

capacity = xnvme_queue_get_capacity(queue);
work.nlb = nlb;
work.n_ranges = n_ranges;
work.nbytes = nbytes;
work.opc = opc;
work.ranges = ranges;

err = xnvme_queue_set_cb(queue, _cb_fn, &work);
if (err) {
printf("Failed to set queue callback, err: %d\n", err);
return err;
}
int err;

for (uint32_t i = 0; i < n_ranges; i++) {
range = &work.ranges[i];
range->slba = slbas[i];
range->elba = elbas[i];
range->dbuf = dbufs[i];
n_blocks = (range->elba - range->slba) + 1;
if (n_blocks % (nlb + 1) != 0) {
printf("n_blocks (%u) is not divisible by nlb + 1 (%u)\n", n_blocks,
nlb + 1);
return -EINVAL;
while (nblocks) {
struct xnvme_cmd_ctx *ctx;
uint64_t n = nblocks < io_nblocks ? nblocks : io_nblocks;
uint64_t nbytes = n * blocksize;

while ((ctx = xnvme_queue_get_cmd_ctx(queue)) == NULL) {
xnvme_queue_poke(queue, 0);
}
}

for (int i = 0; i < capacity - 1; i++) {
struct xnvme_cmd_ctx *ctx = xnvme_queue_get_cmd_ctx(queue);
err = _submit(&work, ctx);
ctx->cmd.common.opcode = XNVME_SPEC_NVM_OPC_READ;
ctx->cmd.common.nsid = nsid;
ctx->cmd.nvm.slba = slba;
ctx->cmd.nvm.nlb = n - 1;

do {
err = xnvme_cmd_pass(ctx, *dbuf, nbytes, NULL, 0);
if (err == -EBUSY || err == -EAGAIN) {
xnvme_queue_poke(queue, 0);
}
} while (err == -EBUSY || err == -EAGAIN);
if (err) {
break;
fprintf(stderr, "Failed to submit, err: %d\n", err);
xnvme_queue_put_cmd_ctx(queue, ctx);
return err;
}

slba += n;
*dbuf = (uint8_t *)*dbuf + nbytes;
nblocks -= n;
}
xnvme_queue_drain(queue);
if (work.errors) {
return -EIO;
}
return err;
return 0;
}

/**
Expand Down Expand Up @@ -177,65 +118,109 @@ fil_extent_slba(struct fil_dev *device, const struct xal_extent *extent, uint64_
return xal_fsbno_offset(device->xal, extent->start_block) / blocksize;
}

/**
* Read a coalesced run and tally the commands it takes into stats. A zero-length
* run is a no-op, so the boundary and final flushes in _submit_device can call
* this unconditionally.
*/
static int
_flush_run(struct fil_iter *iter, struct fil_dev *device, uint32_t nsid, uint64_t slba,
uint64_t run, uint32_t io_nblocks, uint64_t blocksize, void **dbuf)
{
int err;

if (!run) {
return 0;
}
err = _submit_run(device->queue, nsid, slba, run, io_nblocks, blocksize, dbuf);
if (err) {
return err;
}
iter->stats->io += (run + io_nblocks - 1) / io_nblocks;
return 0;
}

/**
* Submit all of a device's per-buffer reads, coalescing physically-adjacent
* extents into runs. Always drains the queue before returning -- including on
* the error path -- so no outstanding command is left referencing a buffer.
*/
static int
_submit_device(struct fil_iter *iter, struct fil_dev *device, uint32_t dev_id)
{
uint64_t blocksize = xnvme_dev_get_geo(device->dev)->lba_nbytes;
uint32_t xal_blksize = xal_get_sb_blocksize(device->xal);
uint32_t nsid = xnvme_dev_get_nsid(device->dev);
uint32_t io_nblocks;
int err = 0;

if (iter->opts->iosize < blocksize || iter->opts->iosize % blocksize) {
fprintf(stderr, "iosize (%lu) must be a multiple of the device LBA size (%lu)\n",
iter->opts->iosize, blocksize);
return EINVAL;
}
io_nblocks = iter->opts->iosize / blocksize;

for (uint32_t j = 0; j < device->n_buffers; j++) {
struct xal_inode *file = fil_next_file(iter, device, dev_id, j, NULL);
void *dbuf = device->buffers[j];
uint64_t slba = 0, run = 0;

/* Coalesce physically-adjacent extents into a single run and read
* it back in iosize-sized commands; a run is flushed whenever the
* next extent is not contiguous with it. */
for (uint32_t k = 0; k < file->content.extents.count; k++) {
const struct xal_extent *ext =
xal_extent_at(device->xal, file->content.extents.extent_idx + k);
uint64_t ext_slba = fil_extent_slba(device, ext, blocksize);
uint64_t ext_blocks = (uint64_t)ext->nblocks * xal_blksize / blocksize;

if (run && ext_slba == slba + run) {
run += ext_blocks;
continue;
}
err = _flush_run(iter, device, nsid, slba, run, io_nblocks, blocksize,
&dbuf);
if (err) {
goto drain;
}
slba = ext_slba;
run = ext_blocks;
}
err = _flush_run(iter, device, nsid, slba, run, io_nblocks, blocksize, &dbuf);
if (err) {
goto drain;
}
}

drain:
xnvme_queue_drain(device->queue);
return err;
}

int
fil_cpu_submit(struct fil_iter *iter)
{
struct xal_inode *dir, *file;
struct xal_extent extent, next_extent;
uint64_t nblocks, nbytes, blocksize, next_slba;
uint32_t xal_blksize, nlb;
struct timespec start, end;
int err;

for (uint32_t i = 0; i < iter->n_devs; i++) {
clock_gettime(CLOCK_MONOTONIC_RAW, &start);
struct fil_dev *device = iter->devs[i];
blocksize = xnvme_dev_get_geo(device->dev)->lba_nbytes;
xal_blksize = xal_get_sb_blocksize(device->xal);
nlb = iter->opts->iosize / blocksize - 1;

for (uint32_t j = 0; j < device->n_buffers; j++) {
file = fil_next_file(iter, device, i, j, &dir);
extent = *xal_extent_at(device->xal, file->content.extents.extent_idx);
device->cpu_io->slbas[j] = fil_extent_slba(device, &extent, blocksize);

nbytes = extent.nblocks * xal_blksize;
nblocks = nbytes / blocksize;
device->io_errors = 0;
clock_gettime(CLOCK_MONOTONIC_RAW, &start);

for (uint32_t k = 1; k < file->content.extents.count; k++) {
next_extent = *xal_extent_at(device->xal,
file->content.extents.extent_idx + k);
next_slba = fil_extent_slba(device, &next_extent, blocksize);
if (next_slba != device->cpu_io->slbas[j] + nblocks) {
fprintf(stderr,
"File: %s, in dir: %s, has non contiguous "
"extents\n",
file->name, dir->name);
fprintf(stderr,
"extent[%d].elba: %lu, extent[%d].slba: %lu\n",
k - 1, device->cpu_io->slbas[j] + nblocks - 1, k,
next_slba);
return ENOTSUP;
}
nbytes += next_extent.nblocks * xal_blksize;
nblocks = nbytes / blocksize;
}
device->cpu_io->elbas[j] = device->cpu_io->slbas[j] + nblocks - 1;
iter->stats->io += nblocks / (nlb + 1);
}
clock_gettime(CLOCK_MONOTONIC_RAW, &end);
iter->stats->prep_time += ELAPSED(start, end);
err = _submit_device(iter, device, i);

clock_gettime(CLOCK_MONOTONIC_RAW, &start);
err = _io_range_submit(device->queue, XNVME_SPEC_NVM_OPC_READ,
device->cpu_io->slbas, device->cpu_io->elbas, nlb,
iter->opts->iosize, device->buffers, device->n_buffers);
clock_gettime(CLOCK_MONOTONIC_RAW, &end);
iter->stats->io_time += ELAPSED(start, end);
if (err) {
fprintf(stderr, "IO failed: %d\n", err);
return err;
}
if (device->io_errors) {
fprintf(stderr, "IO failed: %u\n", device->io_errors);
return -EIO;
}
}
return 0;
}
Expand Down
Loading
Loading