d/numam
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

device driver abstraction

Browse Source
This commit is contained in:
quackerd 2022-03-21 19:43:49 +08:00
parent 2ecfacff11
commit 27c6cd188d
23 changed files with 468 additions and 189 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CMakeLists.txt
View File

@ -51,7 +51,7 @@ add_executable(rat EXCLUDE_FROM_ALL net/rat.cc)
target_link_libraries(rat PRIVATE pthread nm ntr gen ${DPDK_LINK_LIBRARIES})
target_compile_options(rat PRIVATE ${CC_FLAGS} ${DPDK_CFLAGS})
add_executable(birb EXCLUDE_FROM_ALL storage/birb.cc storage/io_gen.cc)
add_executable(birb EXCLUDE_FROM_ALL storage/birb.cc storage/io_gen.cc storage/bdev.cc storage/bdev_thread.cc)
target_include_directories(birb PRIVATE ${SPDK_INCLUDE_DIRS} ${DPDK_INCLUDE_DIRS} ${UUID_INCLUDE_DIRS})
target_compile_options(birb PRIVATE ${CC_FLAGS} ${SPDK_CFLAGS} ${UUID_CFLAGS})
target_link_directories(birb PRIVATE ${SPDK_LIBRARY_DIRS} ${SPDK_SYS_STATIC_LIBRARY_DIRS} ${UUID_LIBRARY_DIRS})

1
inc/defs.h → inc/defs.hh
View File

@ -15,6 +15,7 @@ constexpr static unsigned long S2US = 1000000UL;
constexpr static unsigned long MS2NS = 1000000UL;
constexpr static int NEXT_CPU_NULL = -1;
static inline int
cmask_get_next_cpu(uint64_t *mask)
{

0
inc/gen.h → inc/gen.hh
View File

4
inc/net/pkt.h → inc/net/pkt.hh
View File

@ -10,8 +10,8 @@
#include <rte_udp.h>
#include <unistd.h>
#include "nm.h"
#include "util.h"
#include "nm.hh"
#include "util.hh"
#include <random>

0
inc/net/util.h → inc/net/util.hh
View File

4
inc/nm.h → inc/nm.hh
View File

@ -1,7 +1,7 @@
#pragma once
#include "gen.h"
#include "defs.h"
#include "gen.hh"
#include "defs.hh"
#include <atomic>
#include <cstdint>

46
inc/storage/driver.hh Normal file
View File

@ -0,0 +1,46 @@
#pragma once
#include "defs.hh"
#include "spdk/thread.h"
#include <cstdlib>
class birb_driver
{
private:
DISALLOW_EVIL_CONSTRUCTORS(birb_driver);
public:
enum birb_driver_status{
BIRB_SUCCESS,
BIRB_FAIL
};
enum birb_driver_type{
BIRB_DRV_NVME,
BIRB_DRV_BDEV
};
virtual size_t get_capacity() = 0;
virtual birb_driver_status get_status() = 0;
virtual size_t get_align() = 0;
virtual birb_driver_type get_type() = 0;
virtual ~birb_driver() = default;
protected:
birb_driver() = default;
};
class birb_driver_thread_context
{
private:
DISALLOW_EVIL_CONSTRUCTORS(birb_driver_thread_context);
public:
using callback = void (*)(bool, void *);
virtual int read(size_t offset, size_t size, char * buffer, callback callback, void * context) = 0;
virtual int write(size_t offset, size_t size, char * buffer, callback callback, void * context) = 0;
virtual birb_driver::birb_driver_status get_status() = 0;
virtual ~birb_driver_thread_context() = default;
protected:
birb_driver_thread_context() = default;
};

56
inc/storage/driver_bdev.hh Normal file
View File

@ -0,0 +1,56 @@
#pragma once
#include "storage/driver.hh"
#include "spdk/bdev.h"
#include "spdk/bdev_zone.h"
#include "spdk/thread.h"
class birb_bdev_driver : public birb_driver
{
public:
birb_bdev_driver(const char * dev_name);
~birb_bdev_driver() override;
size_t get_capacity() override;
birb_driver_status get_status() override;
struct spdk_bdev * get_bdev();
struct spdk_bdev_desc * get_bdev_desc();
birb_driver_type get_type() override;
size_t get_align() override;
private:
DISALLOW_EVIL_CONSTRUCTORS(birb_bdev_driver);
struct spdk_bdev_desc * bdev_desc;
struct spdk_bdev * bdev;
size_t block_sz;
size_t block_num;
birb_driver_status status;
static void print_all_bdev();
static void bdev_event_cb(enum spdk_bdev_event_type type, struct spdk_bdev * bdev,
void * event_ctx);
};
class birb_bdev_thread_context : public birb_driver_thread_context
{
public:
birb_bdev_thread_context(birb_bdev_driver * driver);
~birb_bdev_thread_context() override;
int read(size_t offset, size_t size, char * buffer, callback callback, void * context) override;
int write(size_t offset, size_t size, char * buffer, callback callback, void * context) override;
birb_driver::birb_driver_status get_status() override;
private:
struct cb_context {
callback cb;
void * ctx;
struct spdk_io_channel * ch;
};
DISALLOW_EVIL_CONSTRUCTORS(birb_bdev_thread_context);
spdk_io_channel * io_channel;
birb_driver::birb_driver_status status;
birb_bdev_driver * driver;
static void io_callback(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg);
};

4
inc/storage/io_gen.hh
View File

@ -1,8 +1,8 @@
#pragma once
#include <sys/endian.h>
#include <sys/types.h>
#include "defs.h"
#include "gen.h"
#include "defs.hh"
#include "gen.hh"
#include <random>
enum io_generator_opcode {

2
libgen/generator.cc
View File

@ -1,6 +1,6 @@
// modified from mutilate
#include "gen.h"
#include "gen.hh"
Generator *
createFacebookKey()

2
libnm/alloc.cc
View File

@ -7,7 +7,7 @@
#include <cerrno>
#include "nmp.h"
#include "nmp.hh"
static pthread_mutex_t alloc_lock;
static constexpr unsigned int MEM_OBJ_SIZE = 4096; // 4k

2
libnm/loadgen.cc
View File

@ -1,4 +1,4 @@
#include "nmp.h"
#include "nmp.hh"
#include <sys/cpuset.h>
#include <sys/domainset.h>

5
libnm/nm.cc
View File

@ -8,9 +8,8 @@
#include <vector>
#include <cerrno>
#include "nm.h"
#include "nmp.h"
#include "defs.h"
#include "nmp.hh"
#include "defs.hh"
static const char *SYSCTL_TSC = "machdep.tsc_freq";
static uint64_t sysctl_tsc_freq = 0;

2
libnm/nmp.h → libnm/nmp.hh
View File

@ -1,6 +1,6 @@
#pragma once
#include "nm.h"
#include "nm.hh"
int nm_topo_init();

3
libnm/topo.cc
View File

@ -1,5 +1,4 @@
#include "nm.h"
#include "nmp.h"
#include "nmp.hh"
#include <cstdio>
#include <cstdlib>

8
net/cat.cc
View File

@ -10,11 +10,11 @@
#include <rte_mbuf.h>
#include <unistd.h>
#include "gen.h"
#include "nm.h"
#include "gen.hh"
#include "nm.hh"
#include "ntr.h"
#include "net/pkt.h"
#include "net/util.h"
#include "net/pkt.hh"
#include "net/util.hh"
#include <atomic>
#include <ctime>

6
net/khat.cc
View File

@ -9,10 +9,10 @@
#include <rte_mbuf.h>
#include <unistd.h>
#include "nm.h"
#include "nm.hh"
#include "ntr.h"
#include "net/pkt.h"
#include "net/util.h"
#include "net/pkt.hh"
#include "net/util.hh"
#include <atomic>
#include <ctime>

8
net/rat.cc
View File

@ -9,11 +9,11 @@
#include <rte_mbuf.h>
#include <unistd.h>
#include "gen.h"
#include "nm.h"
#include "gen.hh"
#include "nm.hh"
#include "ntr.h"
#include "net/pkt.h"
#include "net/util.h"
#include "net/pkt.hh"
#include "net/util.hh"
#include <atomic>
#include <list>

95
storage/bdev.cc Normal file
View File

@ -0,0 +1,95 @@
#include <sys/endian.h>
#include "storage/driver_bdev.hh"
#include "ntr.h"
#include "spdk/bdev.h"
#include "spdk/thread.h"
size_t
birb_bdev_driver::get_capacity()
{
return block_num * block_sz;
}
birb_driver::birb_driver_status
birb_bdev_driver::get_status()
{
return this->status;
}
void
birb_bdev_driver::bdev_event_cb(enum spdk_bdev_event_type type, struct spdk_bdev * bdev UNUSED,
void * event_ctx UNUSED)
{
ntr(NTR_DEP_USER1, NTR_LEVEL_WARNING, "bdev_event_cb: unsupported bdev event: type %d\n", type);
}
void
birb_bdev_driver::print_all_bdev()
{
struct spdk_bdev * cur = spdk_bdev_first();
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "birb_bdev_driver: all registered block devices: ");
while(cur != NULL) {
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "%s, ", spdk_bdev_get_name(cur));
cur = spdk_bdev_next(cur);
}
}
birb_bdev_driver::birb_bdev_driver(const char * dev_name) : bdev_desc(nullptr),
bdev(nullptr),
block_sz(0),
block_num(0),
status(BIRB_FAIL)
{
int rc;
rc = spdk_bdev_open_ext(dev_name, true, birb_bdev_driver::bdev_event_cb, NULL, &this->bdev_desc);
if (rc != 0) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "birb_bdev_driver: failed to open bdev: %d\n", rc);
return;
}
/* A bdev pointer is valid while the bdev is opened. */
this->bdev = spdk_bdev_desc_get_bdev(this->bdev_desc);
this->block_sz = spdk_bdev_get_block_size(this->bdev);
this->block_num = spdk_bdev_get_num_blocks(this->bdev);
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "birb_bdev_driver: bdev block size %zu bytes, blocks count %zu\n", this->block_sz, this->block_num);
this->status = BIRB_SUCCESS;
}
birb_bdev_driver::~birb_bdev_driver()
{
if (this->status == BIRB_SUCCESS) {
spdk_bdev_close(this->bdev_desc);
}
}
birb_driver::birb_driver_type
birb_bdev_driver::get_type()
{
return BIRB_DRV_BDEV;
}
size_t
birb_bdev_driver::get_align()
{
return spdk_bdev_get_buf_align(this->bdev);
}
struct spdk_bdev *
birb_bdev_driver::get_bdev()
{
return this->bdev;
}
struct spdk_bdev_desc *
birb_bdev_driver::get_bdev_desc()
{
return this->bdev_desc;
}

66
storage/bdev_thread.cc Normal file
View File

@ -0,0 +1,66 @@
#include <sys/endian.h>
#include "storage/driver_bdev.hh"
#include "ntr.h"
#include "spdk/bdev.h"
#include "spdk/thread.h"
birb_bdev_thread_context::birb_bdev_thread_context(birb_bdev_driver * driver) : io_channel(nullptr),
status(birb_driver::BIRB_FAIL),
driver(driver)
{
struct spdk_bdev_desc * desc = driver->get_bdev_desc();
// obtain io channel
this->io_channel = spdk_bdev_get_io_channel(desc);
if (io_channel == nullptr) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "birb_bdev_thread_context: could not create bdev I/O channel!\n");
}
this->status = birb_driver::BIRB_SUCCESS;
}
birb_driver::birb_driver_status
birb_bdev_thread_context::get_status()
{
return this->status;
}
birb_bdev_thread_context::~birb_bdev_thread_context()
{
if (this->io_channel != nullptr) {
spdk_put_io_channel(this->io_channel);
}
}
/*
* Callback function for io completion.
*/
void
birb_bdev_thread_context::io_callback(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg)
{
spdk_bdev_free_io(bdev_io);
auto ctx = reinterpret_cast<struct cb_context *>(cb_arg);
ctx->cb(success, ctx->ctx);
delete ctx;
}
int
birb_bdev_thread_context::read(size_t offset, size_t size, char * buffer, callback callback, void * context)
{
auto ctx = new struct cb_context;
ctx->cb = callback;
ctx->ctx = context;
return spdk_bdev_read(driver->get_bdev_desc(), this->io_channel, buffer, offset, size, io_callback, reinterpret_cast<void*>(ctx));
}
int
birb_bdev_thread_context::write(size_t offset, size_t size, char * buffer, callback callback, void * context)
{
auto ctx = new struct cb_context;
ctx->cb = callback;
ctx->ctx = context;
return spdk_bdev_write(driver->get_bdev_desc(), this->io_channel, buffer, offset, size, io_callback, reinterpret_cast<void*>(ctx));
}

337
storage/birb.cc
View File

@ -21,20 +21,18 @@
#include "spdk/cpuset.h"
#include "spdk/stdinc.h"
#include "spdk/thread.h"
#include "spdk/bdev.h"
#include "spdk/env.h"
#include "spdk/event.h"
#include "spdk/log.h"
#include "spdk/string.h"
#include "spdk/bdev_zone.h"
#include "gen.h"
#include "gen.hh"
#include "ntr.h"
#include "defs.h"
#include "nm.h"
#include "defs.hh"
#include "nm.hh"
#include "storage/io_gen.hh"
#include "storage/driver.hh"
#include "storage/driver_bdev.hh"
static inline uint64_t get_cur_ts_nano()
{
@ -47,20 +45,23 @@ static inline uint64_t get_cur_ts_nano()
* our events and callbacks.
*/
static constexpr unsigned long MAX_SPEC_LEN = 32;
static constexpr unsigned long MAX_BDEV_NAME_LEN = 32;
static constexpr unsigned long MAX_DEV_NAME_LEN = 32;
static constexpr unsigned long MAX_OUTPUT_FILE_LEN = 256;
struct options_t {
// args
int verbosity = NTR_LEVEL_DEFAULT;
int num_threads = 1;
unsigned long cpumask = 1;
char pattern_spec[MAX_SPEC_LEN];
char ia_spec[MAX_SPEC_LEN];
char pattern_spec[MAX_SPEC_LEN] = "R,100";
char ia_spec[MAX_SPEC_LEN] = "fixed";
unsigned int time = 5;
unsigned int warmup = 2;
unsigned int queue_depth = 1;
char bdev_name[MAX_BDEV_NAME_LEN];
char dev_name[MAX_DEV_NAME_LEN] = "Malloc0";
char driver_name[MAX_DEV_NAME_LEN] = "bdev";
unsigned int read_pct = 0;
io_generator_address_mode addr_mode = IOGEN_ADDR_UNIFORM_RANDOM;
char output_file[MAX_OUTPUT_FILE_LEN] = "output.txt";
@ -95,48 +96,49 @@ struct io_request {
char * dma_buf;
};
// thread local states (RO by worker threads)
struct thread_context {
unsigned int tid;
unsigned int coreid;
unsigned int sockid;
pthread_t sys_thread;
char thread_name[32];
struct spdk_thread *main_thread;
struct spdk_bdev *bdev;
struct spdk_bdev_desc *bdev_desc;
struct spdk_thread *s_thread;
struct spdk_thread * main_thread;
birb_driver * driver;
const char * ia_gen_desc;
unsigned long start_region_offset;
unsigned long start_region_length;
unsigned int read_pct;
io_generator_address_mode addr_mode;
/* modified by worker threads */
struct spdk_thread * sp_thread;
std::list<io_record *> *io_records;
};
static void dump_options()
{
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "main: Options:\n"
" bdev name: %s\n"
" dev name: %s\n"
" driver name: %s\n"
" worker threads: 0x%lx\n"
" number of threads: %d\n"
" IO request size: %lu\n"
" IO requests per second: %lu\n"
" IO pattern: %s\n"
" IO queue depth: %d\n"
" IO addressing mode: %d\n"
" read percent: %u\n"
" inter-arrival dist: %s\n"
" run time: %d\n"
" warmup time: %d\n"
" output file: %s\n",
options.bdev_name,
options.dev_name,
options.driver_name,
options.cpumask,
options.num_threads,
options.req_size,
options.rps,
options.pattern_spec,
options.queue_depth,
options.addr_mode,
options.read_pct,
options.ia_spec,
options.time,
options.warmup,
@ -148,7 +150,8 @@ static void usage()
{
fprintf(stdout,
" -V(VV): verbose mode\n"
" -D: bdev name\n"
" -D: dev name\n"
" -k: driver to use (default bdev)\n"
" -a: worker threads spec (0x3 = spawn 2 threads on core 1 & 2)\n"
" -b: IO request size\n"
" -q: IO requests per second\n"
@ -168,7 +171,10 @@ static int parse_arg(int c, char *arg)
ntr_get_level(NTR_DEP_USER1) + 1);
break;
case 'D':
strncpy(options.bdev_name, arg, MAX_BDEV_NAME_LEN);
strncpy(options.dev_name, arg, MAX_DEV_NAME_LEN);
break;
case 'k':
strncpy(options.driver_name, arg, MAX_DEV_NAME_LEN);
break;
case 'a':
options.cpumask = strtoull(optarg, nullptr, 16);
@ -218,22 +224,56 @@ static int parse_arg(int c, char *arg)
return 0;
}
static birb_driver *
birb_create_driver(const char * driver_name, void * context)
{
if (strcmp(driver_name, "bdev") == 0) {
return new birb_bdev_driver((const char *)context);
} else if (strcmp(driver_name, "nvme") == 0) {
return nullptr;
} else {
return nullptr;
}
}
static birb_driver_thread_context *
birb_create_thread_context(birb_driver * driver)
{
if (driver->get_type() == birb_driver::BIRB_DRV_BDEV) {
return new birb_bdev_thread_context(dynamic_cast<birb_bdev_driver *>(driver));
} else if (driver->get_type() == birb_driver::BIRB_DRV_NVME) {
return nullptr;
} else {
return nullptr;
}
}
static void
birb_destroy_driver(birb_driver * drv)
{
delete drv;
}
static void
birb_destroy_thread_context(birb_driver_thread_context * ctx)
{
delete ctx;
}
/*
* Callback function for io completion.
*/
static void
worker_io_complete(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg)
worker_io_complete(bool success, void *cb_arg)
{
auto vars = (struct worker_thread_cb_vars *)cb_vars;
auto req = (struct io_request *)cb_arg;
spdk_bdev_free_io(bdev_io);
uint64_t end_ts = get_cur_ts_nano();
if (!success) {
// XXX: print warning for errors for now
ntr(NTR_DEP_USER1, NTR_LEVEL_WARNING, "thread %d <worker_read_complete>: io request failed\n", vars->ctx->tid);
ntr(NTR_DEP_USER1, NTR_LEVEL_WARNING, "thread %d <worker_io_complete>: io request failed\n", vars->ctx->tid);
} else {
auto rec = new struct io_record;
rec->start_ts = req->start_ts;
@ -244,18 +284,12 @@ worker_io_complete(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg)
memcpy(req->user_buf, req->dma_buf, options.req_size);
}
ntr(NTR_DEP_USER1, NTR_LEVEL_DEBUG, "thread %d <worker_read_complete>: completed io request type %d\n", vars->ctx->tid, req->op);
ntr(NTR_DEP_USER1, NTR_LEVEL_DEBUG, "thread %d <worker_io_complete>: completed io request type %d\n", vars->ctx->tid, req->op);
}
vars->free_ios->push_back(req);
}
static void
bdev_event_cb(enum spdk_bdev_event_type type, struct spdk_bdev * bdev UNUSED,
void * event_ctx UNUSED)
{
ntr(NTR_DEP_USER1, NTR_LEVEL_WARNING, "unsupported bdev event: type %d\n", type);
}
static void
cb_notify_main_init(void * arg)
@ -319,11 +353,14 @@ worker_thread_main(void * arg)
{
int rc = 0;
constexpr static unsigned int SPDK_THREAD_NAME_SZ = 16;
struct worker_thread_cb_vars vars;
auto *ctx = (struct thread_context *)arg;
struct spdk_io_channel *io_channel = nullptr;
const unsigned long buf_align = spdk_bdev_get_buf_align(ctx->bdev);
birb_driver_thread_context * driver_thread_ctx;
std::list<struct io_request *> free_ios;
char spdk_thread_name[SPDK_THREAD_NAME_SZ];
struct spdk_cpuset * cpuset;
Generator * ia_gen = nullptr;
io_generator * io_gen = nullptr;
@ -333,13 +370,35 @@ worker_thread_main(void * arg)
uint64_t a_offset;
ntr(NTR_DEP_USER1, NTR_LEVEL_DEBUG, "thread %d: init...\n", ctx->tid);
// associate current thread with spdk thread
spdk_set_thread(ctx->s_thread);
// create spdk thread
cpuset = spdk_cpuset_alloc();
if (cpuset == nullptr) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "main: failed to alloc cpuset\n");
rc = ENOMEM;
goto cleanup;
}
spdk_cpuset_zero(cpuset);
spdk_cpuset_set_cpu(cpuset, ctx->coreid, true);
snprintf(spdk_thread_name, SPDK_THREAD_NAME_SZ, "birb_worker_%u", ctx->tid);
ctx->sp_thread = spdk_thread_create(spdk_thread_name, cpuset);
if (ctx->sp_thread == nullptr) {
rc = ENOMEM;
goto cleanup;
}
spdk_set_thread(ctx->sp_thread);
// create thread context
driver_thread_ctx = birb_create_thread_context(ctx->driver);
if (driver_thread_ctx == nullptr || driver_thread_ctx->get_status() != birb_driver::BIRB_SUCCESS) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "thread %d: could not create thread context!\n", ctx->tid);
rc = EINVAL;
goto cleanup;
}
// create io request objects
for (unsigned int i = 0; i < options.queue_depth; i++) {
auto dma_buf = (char *)spdk_dma_zmalloc_socket(options.req_size, buf_align, NULL, ctx->sockid);
auto dma_buf = (char *)spdk_dma_zmalloc_socket(options.req_size, ctx->driver->get_align(), NULL, ctx->sockid);
auto user_buf = (char *)nm_malloc(ctx->sockid, options.req_size);
if (dma_buf == nullptr || user_buf == nullptr) {
@ -358,16 +417,8 @@ worker_thread_main(void * arg)
// init thread local states
worker_thread_cb_vars_init(&vars, ctx, &free_ios);
cb_vars = &vars;
// obtain io channel
io_channel = spdk_bdev_get_io_channel(ctx->bdev_desc);
if (io_channel == nullptr) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "thread %d: could not create bdev I/O channel!\n", ctx->tid);
rc = EINVAL;
goto cleanup;
}
ia_gen = createGenerator(ctx->ia_gen_desc);
ia_gen = createGenerator(options.ia_spec);
if (ia_gen == nullptr) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "thread %d: could not allocate ia generator!\n", ctx->tid);
rc = EINVAL;
@ -375,7 +426,7 @@ worker_thread_main(void * arg)
}
ia_gen->set_lambda((double)options.rps / (double)(options.num_threads));
io_gen = new io_generator(options.req_size, ctx->start_region_length, ctx->read_pct, ctx->addr_mode);
io_gen = new io_generator(options.req_size, ctx->start_region_length, options.read_pct, options.addr_mode);
if (io_gen == nullptr) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "thread %d: could not allocate ia generator!\n", ctx->tid);
rc = EINVAL;
@ -426,11 +477,9 @@ worker_thread_main(void * arg)
io_req->op = io_ctx.op;
if(io_ctx.op == IOGEN_READ) {
rc = spdk_bdev_read(ctx->bdev_desc, io_channel, io_req->dma_buf,
a_offset, io_ctx.size, worker_io_complete, io_req);
rc = driver_thread_ctx->read(a_offset, io_ctx.size, io_req->dma_buf, worker_io_complete, io_req);
} else {
rc = spdk_bdev_write(ctx->bdev_desc, io_channel, io_req->dma_buf,
a_offset, io_ctx.size, worker_io_complete, io_req);
rc = driver_thread_ctx->write(a_offset, io_ctx.size, io_req->dma_buf, worker_io_complete, io_req);
}
if (rc != 0) {
@ -452,10 +501,6 @@ cleanup:
nm_free(ctx->sockid, req->user_buf);
}
if (io_channel != nullptr) {
spdk_put_io_channel(io_channel);
}
if (ia_gen != nullptr) {
delete ia_gen;
}
@ -464,22 +509,43 @@ cleanup:
delete io_gen;
}
if (cpuset != nullptr) {
spdk_cpuset_free(cpuset);
}
if (driver_thread_ctx != nullptr) {
birb_destroy_thread_context(driver_thread_ctx);
}
if (rc == 0) {
if ((rc = spdk_thread_send_msg(ctx->main_thread, cb_notify_main_stop, ctx)) != 0) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "thread %d: could not send message %d\n", ctx->tid, rc);
}
}
spdk_thread_exit(ctx->sp_thread);
while (!spdk_thread_is_exited(ctx->sp_thread)) {
spdk_thread_poll(ctx->sp_thread, 0, 0);
};
if (ctx->sp_thread != nullptr) {
spdk_set_thread(nullptr);
spdk_thread_destroy(ctx->sp_thread);
}
ntr(NTR_DEP_USER1, NTR_LEVEL_DEBUG, "thread %d: stopped...\n", ctx->tid);
if (rc != 0) {
spdk_app_stop(rc);
}
if ((rc = spdk_thread_send_msg(ctx->main_thread, cb_notify_main_stop, ctx)) != 0) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "thread %d: could not send message %d\n", ctx->tid, rc);
}
return nullptr;
}
static void
parse_pattern(char * pattern, int * read_pct, io_generator_address_mode * addr_mode)
parse_pattern(char * pattern, unsigned int * read_pct, io_generator_address_mode * addr_mode)
{
char * token = strtok(pattern, ",");
@ -493,90 +559,48 @@ parse_pattern(char * pattern, int * read_pct, io_generator_address_mode * addr_m
*read_pct = strtoull(token, nullptr, 10);
}
static void
print_all_bdev()
{
struct spdk_bdev * cur = spdk_bdev_first();
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "main: All registered block devices: ");
while(cur != NULL) {
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "%s, ", spdk_bdev_get_name(cur));
cur = spdk_bdev_next(cur);
}
}
static void
birb_main(void * arg1 UNUSED)
{
int rc = 0;
struct spdk_bdev * bdev;
struct spdk_bdev_desc * bdev_desc;
std::list<struct thread_context *> worker_threads;
std::ofstream output_file;
uint64_t total_reqs = 0;
io_generator_address_mode addr_mode = IOGEN_ADDR_MONOTONIC_INCREASING;
int read_pct = 0;
struct main_thread_cb_vars vars;
birb_driver * drv = nullptr;
unsigned long record_cutoff_time = 0;
unsigned long current_s = 0;
unsigned int total_reqs = 0;
unsigned int tid = 0;
unsigned long per_thread_cap = 0;
int cur_core = cmask_get_next_cpu(&options.cpumask);
/* initialize driver */
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "main: initializing device driver for device %s\n", options.dev_name);
drv = birb_create_driver(options.driver_name, options.dev_name);
if (drv == nullptr || drv->get_status() != birb_driver::BIRB_SUCCESS) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "main: failed to create device driver.\n");
rc = EINVAL;
goto end;
}
per_thread_cap = drv->get_capacity() / options.num_threads;
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "main: initialized device with capacity %zu bytes ~= %zu MB\n", drv->get_capacity(), drv->get_capacity() / 1024 / 1024);
/* misc init */
main_thread_cb_vars_init(&vars);
cb_vars = &vars;
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "main: successfully started the application\n");
parse_pattern(options.pattern_spec, &options.read_pct, &options.addr_mode);
dump_options();
/* process spec */
parse_pattern(options.pattern_spec, &read_pct, &addr_mode);
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "main: read percent %d, address mode %d\n", read_pct, addr_mode);
/*
* There can be many bdevs configured, but this application will only use
* the one input by the user at runtime.
*
* Open the bdev by calling spdk_bdev_open_ext() with its name.
* The function will return a descriptor
*/
print_all_bdev();
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "main: opening block device %s\n", options.bdev_name);
rc = spdk_bdev_open_ext(options.bdev_name, true, bdev_event_cb, NULL, &bdev_desc);
if (rc != 0) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "main: failed to open bdev: %d\n", rc);
spdk_app_stop(rc);
return;
}
output_file.open(options.output_file, std::ofstream::out);
if (!output_file) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "main: failed to open output file %s\n", options.output_file);
spdk_app_stop(EINVAL);
return;
}
/* A bdev pointer is valid while the bdev is opened. */
bdev = spdk_bdev_desc_get_bdev(bdev_desc);
const uint32_t blk_size = spdk_bdev_get_block_size(bdev);
const unsigned long bdev_capacity = blk_size * spdk_bdev_get_num_blocks(bdev);
const unsigned long per_thread_cap = bdev_capacity / options.num_threads;
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "main: bdev block size %d bytes, # blocks %lu, per_thread_cap %lu\n", blk_size, spdk_bdev_get_num_blocks(bdev), per_thread_cap);
unsigned long record_cutoff_time = 0;
unsigned long current_s = 0;
/* create worker threads */
unsigned int tid = 0;
int cur_core = cmask_get_next_cpu(&options.cpumask);
struct spdk_cpuset * cpuset = spdk_cpuset_alloc();
if (cpuset == NULL) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "main: failed to alloc cpuset\n");
rc = ENOMEM;
rc = EINVAL;
goto end;
}
while(cur_core != NEXT_CPU_NULL) {
auto * ctx = new struct thread_context;
memset(ctx, 0, sizeof(struct thread_context));
@ -588,27 +612,13 @@ birb_main(void * arg1 UNUSED)
}
ctx->tid = tid++;
ctx->driver = drv;
ctx->main_thread = spdk_get_thread();
snprintf(ctx->thread_name, 32, "birb_wrk_%d", ctx->tid);
ctx->sockid = rte_lcore_to_socket_id(cur_core);
ctx->coreid = cur_core;
ctx->bdev = bdev;
ctx->bdev_desc = bdev_desc;
ctx->io_records = new std::list<struct io_record *>();
ctx->start_region_length = per_thread_cap;
ctx->start_region_offset = per_thread_cap * ctx->tid;
ctx->ia_gen_desc = options.ia_spec;
ctx->addr_mode = addr_mode;
ctx->read_pct = read_pct;
// create spdk thread
spdk_cpuset_zero(cpuset);
spdk_cpuset_set_cpu(cpuset, cur_core, true);
ctx->s_thread = spdk_thread_create(ctx->thread_name, cpuset);
if (ctx->s_thread == nullptr) {
rc = ENOMEM;
goto end;
}
// create sys thread
pthread_attr_t attr;
@ -617,11 +627,11 @@ birb_main(void * arg1 UNUSED)
CPU_SET(cur_core, &scpuset);
pthread_attr_init(&attr);
pthread_attr_setaffinity_np(&attr, sizeof(cpuset_t), &scpuset);
rc = pthread_create(&ctx->sys_thread, NULL, worker_thread_main, ctx);
rc = pthread_create(&ctx->sys_thread, nullptr, worker_thread_main, ctx);
if (rc != 0) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "main: failed to create sys thread: %d\n", rc);
spdk_app_stop(EINVAL);
return;
rc = EINVAL;
goto end;
}
worker_threads.push_back(ctx);
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "main: created worker thread %d on core %d socket %d offset 0x%lx length %ld\n", ctx->tid, cur_core, ctx->sockid,
@ -631,16 +641,14 @@ birb_main(void * arg1 UNUSED)
cur_core = cmask_get_next_cpu(&options.cpumask);
}
spdk_cpuset_free(cpuset);
ntr(NTR_DEP_USER1, NTR_LEVEL_DEBUG, "main: waiting for worker thread preinit...\n");
ntr(NTR_DEP_USER1, NTR_LEVEL_DEBUG, "main: waiting for worker thread init...\n");
while(vars.worker_thread_init_cnt < (uint32_t)options.num_threads) {
spdk_thread_poll(spdk_get_thread(), 0, 0);
}
ntr(NTR_DEP_USER1, NTR_LEVEL_DEBUG, "main: starting worker threads...\n");
for (struct thread_context * tctx : worker_threads) {
rc = spdk_thread_send_msg(tctx->s_thread, cb_notify_worker_start, tctx);
rc = spdk_thread_send_msg(tctx->sp_thread, cb_notify_worker_start, tctx);
if (rc != 0) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "main: failed to send message %d\n", rc);
@ -648,6 +656,7 @@ birb_main(void * arg1 UNUSED)
}
}
/* main event loop */
while(current_s < options.time) {
if (current_s >= options.warmup && record_cutoff_time == 0) {
record_cutoff_time = get_cur_ts_nano();
@ -658,7 +667,7 @@ birb_main(void * arg1 UNUSED)
ntr(NTR_DEP_USER1, NTR_LEVEL_DEBUG, "main: stopping worker threads...\n");
for (struct thread_context * tctx : worker_threads) {
rc = spdk_thread_send_msg(tctx->s_thread, cb_notify_worker_stop, tctx);
rc = spdk_thread_send_msg(tctx->sp_thread, cb_notify_worker_stop, tctx);
if (rc != 0) {
ntr(NTR_DEP_USER1, NTR_LEVEL_ERROR, "main: failed to send message %d\n", rc);
@ -672,32 +681,40 @@ birb_main(void * arg1 UNUSED)
// keep stats
for (struct thread_context * tctx : worker_threads) {
uint64_t last_ts = 0;
for (struct io_record * r : *tctx->io_records) {
if (r->start_ts >= record_cutoff_time) {
if (r->end_ts > last_ts) {
last_ts = r->end_ts;
}
output_file << r->end_ts - r->start_ts << std::endl;
total_reqs++;
}
}
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "main: thread %d processed requests: %zu, last request %lu\n",
tctx->tid, tctx->io_records->size(), last_ts);
}
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "main: total requests: %lu, bytes per second: %lu\n", total_reqs, total_reqs * options.req_size / (options.time - options.warmup));
ntr(NTR_DEP_USER1, NTR_LEVEL_INFO, "main: total requests: %u, bytes per second: %lu\n",
total_reqs, total_reqs * options.req_size / (options.time - options.warmup));
end:
if (drv != nullptr) {
birb_destroy_driver(drv);
}
output_file.close();
for (struct thread_context * tctx : worker_threads) {
for (struct io_record * r : *tctx->io_records) {
delete r;
}
spdk_thread_destroy(tctx->s_thread);
delete tctx->io_records;
delete tctx;
}
if (bdev_desc != nullptr) {
spdk_bdev_close(bdev_desc);
}
exit(0);
spdk_app_stop(rc);
return;
@ -720,7 +737,7 @@ main(int argc, char **argv)
* Parse built-in SPDK command line parameters as well
* as our custom one(s).
*/
if ((rc = spdk_app_parse_args(argc, argv, &opts, "VD:a:b:q:Q:P:I:t:w:o:", NULL, parse_arg,
if ((rc = spdk_app_parse_args(argc, argv, &opts, "VD:k:a:b:q:Q:P:I:t:w:o:", NULL, parse_arg,
usage)) != SPDK_APP_PARSE_ARGS_SUCCESS) {
exit(rc);
}

2
storage/io_gen.cc
View File

@ -1,7 +1,7 @@
#include <sys/endian.h>
#include <random>
#include "nm.h"
#include "nm.hh"
#include "storage/io_gen.hh"
io_generator::io_generator(

2
util/memloadgen.cc
View File

@ -1,4 +1,4 @@
#include "nm.h"
#include "nm.hh"
#include "ntr.h"
#include <getopt.h>
#include <unistd.h>