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 #include "spdk/thread.h" 36 #include "spdk/env.h" 37 #include "spdk/event.h" 38 #include "spdk/log.h" 39 #include "spdk/string.h" 40 #include "spdk/accel_engine.h" 41 42 static uint64_t g_tsc_rate; 43 static uint64_t g_tsc_us_rate; 44 static uint64_t g_tsc_end; 45 static int g_xfer_size_bytes = 4096; 46 static int g_queue_depth = 32; 47 static int g_time_in_sec = 5; 48 static bool g_verify = false; 49 static struct worker_thread *g_workers = NULL; 50 static int g_num_workers = 0; 51 static pthread_mutex_t g_workers_lock = PTHREAD_MUTEX_INITIALIZER; 52 53 struct worker_thread { 54 struct spdk_io_channel *ch; 55 uint64_t xfer_completed; 56 uint64_t xfer_failed; 57 uint64_t current_queue_depth; 58 struct spdk_mempool *data_pool; 59 struct spdk_mempool *task_pool; 60 struct worker_thread *next; 61 unsigned core; 62 struct spdk_thread *thread; 63 bool is_draining; 64 struct spdk_poller *is_draining_poller; 65 struct spdk_poller *stop_poller; 66 }; 67 68 struct ap_task { 69 void *src; 70 void *dst; 71 struct worker_thread *worker; 72 }; 73 74 inline static struct ap_task * 75 __ap_task_from_accel_task(struct spdk_accel_task *at) 76 { 77 return (struct ap_task *)((uintptr_t)at - sizeof(struct ap_task)); 78 } 79 80 inline static struct spdk_accel_task * 81 __accel_task_from_ap_task(struct ap_task *ap) 82 { 83 return (struct spdk_accel_task *)((uintptr_t)ap + sizeof(struct ap_task)); 84 } 85 86 static void 87 dump_user_config(struct spdk_app_opts *opts) 88 { 89 printf("SPDK Configuration:\n"); 90 printf("Core mask: %s\n\n", opts->reactor_mask); 91 printf("Accel Perf Configuration:\n"); 92 printf("Transfer size: %u bytes\n", g_xfer_size_bytes); 93 printf("Queue depth: %u\n", g_queue_depth); 94 printf("Run time: %u seconds\n", g_time_in_sec); 95 printf("Verify: %s\n\n", g_verify ? "Yes" : "No"); 96 } 97 98 static void 99 usage(void) 100 { 101 printf("accel_perf options:\n"); 102 printf("\t[-h help message]\n"); 103 printf("\t[-q queue depth]\n"); 104 printf("\t[-n number of channels]\n"); 105 printf("\t[-o transfer size in bytes]\n"); 106 printf("\t[-t time in seconds]\n"); 107 printf("\t[-y verify copy result if this switch is on]\n"); 108 } 109 110 static int 111 parse_args(int argc, char *argv) 112 { 113 switch (argc) { 114 case 'o': 115 g_xfer_size_bytes = spdk_strtol(optarg, 10); 116 break; 117 case 'q': 118 g_queue_depth = spdk_strtol(optarg, 10); 119 break; 120 case 't': 121 g_time_in_sec = spdk_strtol(optarg, 10); 122 break; 123 case 'y': 124 g_verify = true; 125 break; 126 default: 127 usage(); 128 return 1; 129 } 130 return 0; 131 } 132 133 static void 134 unregister_worker(void *arg1) 135 { 136 struct worker_thread *worker = arg1; 137 138 spdk_mempool_free(worker->data_pool); 139 spdk_mempool_free(worker->task_pool); 140 spdk_put_io_channel(worker->ch); 141 pthread_mutex_lock(&g_workers_lock); 142 assert(g_num_workers >= 1); 143 if (--g_num_workers == 0) { 144 pthread_mutex_unlock(&g_workers_lock); 145 spdk_app_stop(0); 146 } 147 pthread_mutex_unlock(&g_workers_lock); 148 } 149 150 static void accel_done(void *ref, int status); 151 152 static void 153 _submit_single(void *arg1, void *arg2) 154 { 155 struct worker_thread *worker = arg1; 156 struct ap_task *task = arg2; 157 158 assert(worker); 159 160 if (g_verify) { 161 memset(task->src, 0x5a, g_xfer_size_bytes); 162 memset(task->dst, 0x0, g_xfer_size_bytes); 163 } 164 task->worker = worker; 165 task->worker->current_queue_depth++; 166 spdk_accel_submit_copy(__accel_task_from_ap_task(task), 167 worker->ch, task->dst, 168 task->src, g_xfer_size_bytes, accel_done); 169 } 170 171 static void 172 _accel_done(void *arg1) 173 { 174 struct ap_task *task = arg1; 175 struct worker_thread *worker = task->worker; 176 177 assert(worker); 178 assert(worker->current_queue_depth > 0); 179 180 if (g_verify) { 181 if (memcmp(task->src, task->dst, g_xfer_size_bytes)) { 182 SPDK_NOTICELOG("Data miscompare\n"); 183 worker->xfer_failed++; 184 /* TODO: cleanup */ 185 exit(-1); 186 } 187 } 188 worker->xfer_completed++; 189 worker->current_queue_depth--; 190 191 if (!worker->is_draining) { 192 _submit_single(worker, task); 193 } else { 194 spdk_mempool_put(worker->data_pool, task->src); 195 spdk_mempool_put(worker->data_pool, task->dst); 196 spdk_mempool_put(worker->task_pool, task); 197 } 198 } 199 200 static int 201 dump_result(void) 202 { 203 uint64_t total_completed = 0; 204 uint64_t total_failed = 0; 205 uint64_t total_xfer_per_sec, total_bw_in_MiBps; 206 struct worker_thread *worker = g_workers; 207 208 printf("\nCore Transfers Bandwidth Failed\n"); 209 printf("-------------------------------------------------\n"); 210 while (worker != NULL) { 211 212 uint64_t xfer_per_sec = worker->xfer_completed / g_time_in_sec; 213 uint64_t bw_in_MiBps = (worker->xfer_completed * g_xfer_size_bytes) / 214 (g_time_in_sec * 1024 * 1024); 215 216 total_completed += worker->xfer_completed; 217 total_failed += worker->xfer_failed; 218 219 if (xfer_per_sec) { 220 printf("%10d%12" PRIu64 "/s%8" PRIu64 " MiB/s%11" PRIu64 "\n", 221 worker->core, xfer_per_sec, 222 bw_in_MiBps, worker->xfer_failed); 223 } 224 225 worker = worker->next; 226 } 227 228 total_xfer_per_sec = total_completed / g_time_in_sec; 229 total_bw_in_MiBps = (total_completed * g_xfer_size_bytes) / 230 (g_time_in_sec * 1024 * 1024); 231 232 printf("=================================================\n"); 233 printf("Total:%16" PRIu64 "/s%8" PRIu64 " MiB/s%11" PRIu64 "\n\n", 234 total_xfer_per_sec, total_bw_in_MiBps, total_failed); 235 236 return total_failed ? 1 : 0; 237 } 238 239 static int 240 _check_draining(void *arg) 241 { 242 struct worker_thread *worker = arg; 243 244 assert(worker); 245 246 if (worker->current_queue_depth == 0) { 247 spdk_poller_unregister(&worker->is_draining_poller); 248 unregister_worker(worker); 249 } 250 251 return -1; 252 } 253 254 static int 255 _worker_stop(void *arg) 256 { 257 struct worker_thread *worker = arg; 258 259 assert(worker); 260 261 spdk_poller_unregister(&worker->stop_poller); 262 263 /* now let the worker drain and check it's outstanding IO with a poller */ 264 worker->is_draining = true; 265 worker->is_draining_poller = SPDK_POLLER_REGISTER(_check_draining, worker, 0); 266 267 return 0; 268 } 269 270 static void 271 _init_thread_done(void *ctx) 272 { 273 } 274 275 static void 276 _init_thread(void *arg1) 277 { 278 struct worker_thread *worker; 279 char buf_pool_name[30], task_pool_name[30]; 280 struct ap_task *task; 281 int i; 282 283 worker = calloc(1, sizeof(*worker)); 284 if (worker == NULL) { 285 fprintf(stderr, "Unable to allocate worker\n"); 286 return; 287 } 288 289 worker->core = spdk_env_get_current_core(); 290 worker->thread = spdk_get_thread(); 291 worker->next = g_workers; 292 worker->ch = spdk_accel_engine_get_io_channel(); 293 snprintf(buf_pool_name, sizeof(buf_pool_name), "buf_pool_%d", g_num_workers); 294 snprintf(task_pool_name, sizeof(task_pool_name), "task_pool_%d", g_num_workers); 295 worker->data_pool = spdk_mempool_create(buf_pool_name, 296 g_queue_depth * 2, /* src + dst */ 297 g_xfer_size_bytes, 298 SPDK_MEMPOOL_DEFAULT_CACHE_SIZE, 299 SPDK_ENV_SOCKET_ID_ANY); 300 worker->task_pool = spdk_mempool_create(task_pool_name, 301 g_queue_depth, 302 spdk_accel_task_size() + sizeof(struct ap_task), 303 SPDK_MEMPOOL_DEFAULT_CACHE_SIZE, 304 SPDK_ENV_SOCKET_ID_ANY); 305 if (!worker->data_pool || !worker->task_pool) { 306 fprintf(stderr, "Could not allocate buffer pool.\n"); 307 spdk_mempool_free(worker->data_pool); 308 spdk_mempool_free(worker->task_pool); 309 free(worker); 310 return; 311 } 312 313 /* Register a poller that will stop the worker at time elapsed */ 314 worker->stop_poller = SPDK_POLLER_REGISTER(_worker_stop, worker, 315 g_time_in_sec * 1000000ULL); 316 317 g_workers = worker; 318 pthread_mutex_lock(&g_workers_lock); 319 g_num_workers++; 320 pthread_mutex_unlock(&g_workers_lock); 321 322 for (i = 0; i < g_queue_depth; i++) { 323 task = spdk_mempool_get(worker->task_pool); 324 if (!task) { 325 fprintf(stderr, "Unable to get accel_task\n"); 326 return; 327 } 328 task->src = spdk_mempool_get(worker->data_pool); 329 task->dst = spdk_mempool_get(worker->data_pool); 330 _submit_single(worker, task); 331 } 332 } 333 334 static void 335 accel_done(void *ref, int status) 336 { 337 struct ap_task *task = __ap_task_from_accel_task(ref); 338 struct worker_thread *worker = task->worker; 339 340 assert(worker); 341 342 spdk_thread_send_msg(worker->thread, _accel_done, task); 343 } 344 345 static void 346 accel_perf_start(void *arg1) 347 { 348 g_tsc_rate = spdk_get_ticks_hz(); 349 g_tsc_us_rate = g_tsc_rate / (1000 * 1000); 350 g_tsc_end = spdk_get_ticks() + g_time_in_sec * g_tsc_rate; 351 352 printf("Running for %d seconds...\n", g_time_in_sec); 353 fflush(stdout); 354 355 spdk_for_each_thread(_init_thread, NULL, _init_thread_done); 356 } 357 358 int 359 main(int argc, char **argv) 360 { 361 struct spdk_app_opts opts = {}; 362 struct worker_thread *worker, *tmp; 363 int rc = 0; 364 365 pthread_mutex_init(&g_workers_lock, NULL); 366 spdk_app_opts_init(&opts); 367 opts.reactor_mask = "0x1"; 368 if ((rc = spdk_app_parse_args(argc, argv, &opts, "o:q:t:y", NULL, parse_args, 369 usage)) != SPDK_APP_PARSE_ARGS_SUCCESS) { 370 rc = -1; 371 goto cleanup; 372 } 373 374 dump_user_config(&opts); 375 rc = spdk_app_start(&opts, accel_perf_start, NULL); 376 if (rc) { 377 SPDK_ERRLOG("ERROR starting application\n"); 378 } else { 379 dump_result(); 380 } 381 382 pthread_mutex_destroy(&g_workers_lock); 383 384 worker = g_workers; 385 while (worker) { 386 tmp = worker->next; 387 free(worker); 388 worker = tmp; 389 } 390 cleanup: 391 spdk_app_fini(); 392 return rc; 393 } 394