1 /*- 2 * BSD LICENSE 3 * 4 * Copyright(c) 2016-2017 Intel Corporation. All rights reserved. 5 * 6 * Redistribution and use in source and binary forms, with or without 7 * modification, are permitted provided that the following conditions 8 * are met: 9 * 10 * * Redistributions of source code must retain the above copyright 11 * notice, this list of conditions and the following disclaimer. 12 * * Redistributions in binary form must reproduce the above copyright 13 * notice, this list of conditions and the following disclaimer in 14 * the documentation and/or other materials provided with the 15 * distribution. 16 * * Neither the name of Intel Corporation nor the names of its 17 * contributors may be used to endorse or promote products derived 18 * from this software without specific prior written permission. 19 * 20 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 21 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 22 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR 23 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT 24 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 25 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT 26 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, 27 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 28 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT 29 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 30 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 31 */ 32 33 #include <string.h> 34 35 #include <rte_vdev.h> 36 #include <rte_memzone.h> 37 #include <rte_kvargs.h> 38 #include <rte_ring.h> 39 #include <rte_errno.h> 40 41 #include "sw_evdev.h" 42 #include "iq_ring.h" 43 #include "event_ring.h" 44 45 #define EVENTDEV_NAME_SW_PMD event_sw 46 #define NUMA_NODE_ARG "numa_node" 47 #define SCHED_QUANTA_ARG "sched_quanta" 48 #define CREDIT_QUANTA_ARG "credit_quanta" 49 50 static void 51 sw_info_get(struct rte_eventdev *dev, struct rte_event_dev_info *info); 52 53 static int 54 sw_port_link(struct rte_eventdev *dev, void *port, const uint8_t queues[], 55 const uint8_t priorities[], uint16_t num) 56 { 57 struct sw_port *p = port; 58 struct sw_evdev *sw = sw_pmd_priv(dev); 59 int i; 60 61 RTE_SET_USED(priorities); 62 for (i = 0; i < num; i++) { 63 struct sw_qid *q = &sw->qids[queues[i]]; 64 65 /* check for qid map overflow */ 66 if (q->cq_num_mapped_cqs >= RTE_DIM(q->cq_map)) { 67 rte_errno = -EDQUOT; 68 break; 69 } 70 71 if (p->is_directed && p->num_qids_mapped > 0) { 72 rte_errno = -EDQUOT; 73 break; 74 } 75 76 if (q->type == SW_SCHED_TYPE_DIRECT) { 77 /* check directed qids only map to one port */ 78 if (p->num_qids_mapped > 0) { 79 rte_errno = -EDQUOT; 80 break; 81 } 82 /* check port only takes a directed flow */ 83 if (num > 1) { 84 rte_errno = -EDQUOT; 85 break; 86 } 87 88 p->is_directed = 1; 89 p->num_qids_mapped = 1; 90 } else if (q->type == RTE_SCHED_TYPE_ORDERED) { 91 p->num_ordered_qids++; 92 p->num_qids_mapped++; 93 } else if (q->type == RTE_SCHED_TYPE_ATOMIC) { 94 p->num_qids_mapped++; 95 } 96 97 q->cq_map[q->cq_num_mapped_cqs] = p->id; 98 rte_smp_wmb(); 99 q->cq_num_mapped_cqs++; 100 } 101 return i; 102 } 103 104 static int 105 sw_port_unlink(struct rte_eventdev *dev, void *port, uint8_t queues[], 106 uint16_t nb_unlinks) 107 { 108 struct sw_port *p = port; 109 struct sw_evdev *sw = sw_pmd_priv(dev); 110 unsigned int i, j; 111 112 int unlinked = 0; 113 for (i = 0; i < nb_unlinks; i++) { 114 struct sw_qid *q = &sw->qids[queues[i]]; 115 for (j = 0; j < q->cq_num_mapped_cqs; j++) { 116 if (q->cq_map[j] == p->id) { 117 q->cq_map[j] = 118 q->cq_map[q->cq_num_mapped_cqs - 1]; 119 rte_smp_wmb(); 120 q->cq_num_mapped_cqs--; 121 unlinked++; 122 123 p->num_qids_mapped--; 124 125 if (q->type == RTE_SCHED_TYPE_ORDERED) 126 p->num_ordered_qids--; 127 128 continue; 129 } 130 } 131 } 132 return unlinked; 133 } 134 135 static int 136 sw_port_setup(struct rte_eventdev *dev, uint8_t port_id, 137 const struct rte_event_port_conf *conf) 138 { 139 struct sw_evdev *sw = sw_pmd_priv(dev); 140 struct sw_port *p = &sw->ports[port_id]; 141 char buf[QE_RING_NAMESIZE]; 142 unsigned int i; 143 144 struct rte_event_dev_info info; 145 sw_info_get(dev, &info); 146 147 /* detect re-configuring and return credits to instance if needed */ 148 if (p->initialized) { 149 /* taking credits from pool is done one quanta at a time, and 150 * credits may be spend (counted in p->inflights) or still 151 * available in the port (p->inflight_credits). We must return 152 * the sum to no leak credits 153 */ 154 int possible_inflights = p->inflight_credits + p->inflights; 155 rte_atomic32_sub(&sw->inflights, possible_inflights); 156 } 157 158 *p = (struct sw_port){0}; /* zero entire structure */ 159 p->id = port_id; 160 p->sw = sw; 161 162 snprintf(buf, sizeof(buf), "sw%d_%s", dev->data->dev_id, 163 "rx_worker_ring"); 164 p->rx_worker_ring = qe_ring_create(buf, MAX_SW_PROD_Q_DEPTH, 165 dev->data->socket_id); 166 if (p->rx_worker_ring == NULL) { 167 SW_LOG_ERR("Error creating RX worker ring for port %d\n", 168 port_id); 169 return -1; 170 } 171 172 p->inflight_max = conf->new_event_threshold; 173 174 snprintf(buf, sizeof(buf), "sw%d_%s", dev->data->dev_id, 175 "cq_worker_ring"); 176 p->cq_worker_ring = qe_ring_create(buf, conf->dequeue_depth, 177 dev->data->socket_id); 178 if (p->cq_worker_ring == NULL) { 179 qe_ring_destroy(p->rx_worker_ring); 180 SW_LOG_ERR("Error creating CQ worker ring for port %d\n", 181 port_id); 182 return -1; 183 } 184 sw->cq_ring_space[port_id] = conf->dequeue_depth; 185 186 /* set hist list contents to empty */ 187 for (i = 0; i < SW_PORT_HIST_LIST; i++) { 188 p->hist_list[i].fid = -1; 189 p->hist_list[i].qid = -1; 190 } 191 dev->data->ports[port_id] = p; 192 193 rte_smp_wmb(); 194 p->initialized = 1; 195 return 0; 196 } 197 198 static void 199 sw_port_release(void *port) 200 { 201 struct sw_port *p = (void *)port; 202 if (p == NULL) 203 return; 204 205 qe_ring_destroy(p->rx_worker_ring); 206 qe_ring_destroy(p->cq_worker_ring); 207 memset(p, 0, sizeof(*p)); 208 } 209 210 static int32_t 211 qid_init(struct sw_evdev *sw, unsigned int idx, int type, 212 const struct rte_event_queue_conf *queue_conf) 213 { 214 unsigned int i; 215 int dev_id = sw->data->dev_id; 216 int socket_id = sw->data->socket_id; 217 char buf[IQ_RING_NAMESIZE]; 218 struct sw_qid *qid = &sw->qids[idx]; 219 220 for (i = 0; i < SW_IQS_MAX; i++) { 221 snprintf(buf, sizeof(buf), "q_%u_iq_%d", idx, i); 222 qid->iq[i] = iq_ring_create(buf, socket_id); 223 if (!qid->iq[i]) { 224 SW_LOG_DBG("ring create failed"); 225 goto cleanup; 226 } 227 } 228 229 /* Initialize the FID structures to no pinning (-1), and zero packets */ 230 const struct sw_fid_t fid = {.cq = -1, .pcount = 0}; 231 for (i = 0; i < RTE_DIM(qid->fids); i++) 232 qid->fids[i] = fid; 233 234 qid->id = idx; 235 qid->type = type; 236 qid->priority = queue_conf->priority; 237 238 if (qid->type == RTE_SCHED_TYPE_ORDERED) { 239 char ring_name[RTE_RING_NAMESIZE]; 240 uint32_t window_size; 241 242 /* rte_ring and window_size_mask require require window_size to 243 * be a power-of-2. 244 */ 245 window_size = rte_align32pow2( 246 queue_conf->nb_atomic_order_sequences); 247 248 qid->window_size = window_size - 1; 249 250 if (!window_size) { 251 SW_LOG_DBG( 252 "invalid reorder_window_size for ordered queue\n" 253 ); 254 goto cleanup; 255 } 256 257 snprintf(buf, sizeof(buf), "sw%d_iq_%d_rob", dev_id, i); 258 qid->reorder_buffer = rte_zmalloc_socket(buf, 259 window_size * sizeof(qid->reorder_buffer[0]), 260 0, socket_id); 261 if (!qid->reorder_buffer) { 262 SW_LOG_DBG("reorder_buffer malloc failed\n"); 263 goto cleanup; 264 } 265 266 memset(&qid->reorder_buffer[0], 267 0, 268 window_size * sizeof(qid->reorder_buffer[0])); 269 270 snprintf(ring_name, sizeof(ring_name), "sw%d_q%d_freelist", 271 dev_id, idx); 272 273 /* lookup the ring, and if it already exists, free it */ 274 struct rte_ring *cleanup = rte_ring_lookup(ring_name); 275 if (cleanup) 276 rte_ring_free(cleanup); 277 278 qid->reorder_buffer_freelist = rte_ring_create(ring_name, 279 window_size, 280 socket_id, 281 RING_F_SP_ENQ | RING_F_SC_DEQ); 282 if (!qid->reorder_buffer_freelist) { 283 SW_LOG_DBG("freelist ring create failed"); 284 goto cleanup; 285 } 286 287 /* Populate the freelist with reorder buffer entries. Enqueue 288 * 'window_size - 1' entries because the rte_ring holds only 289 * that many. 290 */ 291 for (i = 0; i < window_size - 1; i++) { 292 if (rte_ring_sp_enqueue(qid->reorder_buffer_freelist, 293 &qid->reorder_buffer[i]) < 0) 294 goto cleanup; 295 } 296 297 qid->reorder_buffer_index = 0; 298 qid->cq_next_tx = 0; 299 } 300 301 qid->initialized = 1; 302 303 return 0; 304 305 cleanup: 306 for (i = 0; i < SW_IQS_MAX; i++) { 307 if (qid->iq[i]) 308 iq_ring_destroy(qid->iq[i]); 309 } 310 311 if (qid->reorder_buffer) { 312 rte_free(qid->reorder_buffer); 313 qid->reorder_buffer = NULL; 314 } 315 316 if (qid->reorder_buffer_freelist) { 317 rte_ring_free(qid->reorder_buffer_freelist); 318 qid->reorder_buffer_freelist = NULL; 319 } 320 321 return -EINVAL; 322 } 323 324 static int 325 sw_queue_setup(struct rte_eventdev *dev, uint8_t queue_id, 326 const struct rte_event_queue_conf *conf) 327 { 328 int type; 329 330 /* SINGLE_LINK can be OR-ed with other types, so handle first */ 331 if (RTE_EVENT_QUEUE_CFG_SINGLE_LINK & conf->event_queue_cfg) { 332 type = SW_SCHED_TYPE_DIRECT; 333 } else { 334 switch (conf->event_queue_cfg) { 335 case RTE_EVENT_QUEUE_CFG_ATOMIC_ONLY: 336 type = RTE_SCHED_TYPE_ATOMIC; 337 break; 338 case RTE_EVENT_QUEUE_CFG_ORDERED_ONLY: 339 type = RTE_SCHED_TYPE_ORDERED; 340 break; 341 case RTE_EVENT_QUEUE_CFG_PARALLEL_ONLY: 342 type = RTE_SCHED_TYPE_PARALLEL; 343 break; 344 case RTE_EVENT_QUEUE_CFG_ALL_TYPES: 345 SW_LOG_ERR("QUEUE_CFG_ALL_TYPES not supported\n"); 346 return -ENOTSUP; 347 default: 348 SW_LOG_ERR("Unknown queue type %d requested\n", 349 conf->event_queue_cfg); 350 return -EINVAL; 351 } 352 } 353 354 struct sw_evdev *sw = sw_pmd_priv(dev); 355 return qid_init(sw, queue_id, type, conf); 356 } 357 358 static void 359 sw_queue_release(struct rte_eventdev *dev, uint8_t id) 360 { 361 struct sw_evdev *sw = sw_pmd_priv(dev); 362 struct sw_qid *qid = &sw->qids[id]; 363 uint32_t i; 364 365 for (i = 0; i < SW_IQS_MAX; i++) 366 iq_ring_destroy(qid->iq[i]); 367 368 if (qid->type == RTE_SCHED_TYPE_ORDERED) { 369 rte_free(qid->reorder_buffer); 370 rte_ring_free(qid->reorder_buffer_freelist); 371 } 372 memset(qid, 0, sizeof(*qid)); 373 } 374 375 static void 376 sw_queue_def_conf(struct rte_eventdev *dev, uint8_t queue_id, 377 struct rte_event_queue_conf *conf) 378 { 379 RTE_SET_USED(dev); 380 RTE_SET_USED(queue_id); 381 382 static const struct rte_event_queue_conf default_conf = { 383 .nb_atomic_flows = 4096, 384 .nb_atomic_order_sequences = 1, 385 .event_queue_cfg = RTE_EVENT_QUEUE_CFG_ATOMIC_ONLY, 386 .priority = RTE_EVENT_DEV_PRIORITY_NORMAL, 387 }; 388 389 *conf = default_conf; 390 } 391 392 static void 393 sw_port_def_conf(struct rte_eventdev *dev, uint8_t port_id, 394 struct rte_event_port_conf *port_conf) 395 { 396 RTE_SET_USED(dev); 397 RTE_SET_USED(port_id); 398 399 port_conf->new_event_threshold = 1024; 400 port_conf->dequeue_depth = 16; 401 port_conf->enqueue_depth = 16; 402 } 403 404 static int 405 sw_dev_configure(const struct rte_eventdev *dev) 406 { 407 struct sw_evdev *sw = sw_pmd_priv(dev); 408 const struct rte_eventdev_data *data = dev->data; 409 const struct rte_event_dev_config *conf = &data->dev_conf; 410 411 sw->qid_count = conf->nb_event_queues; 412 sw->port_count = conf->nb_event_ports; 413 sw->nb_events_limit = conf->nb_events_limit; 414 rte_atomic32_set(&sw->inflights, 0); 415 416 if (conf->event_dev_cfg & RTE_EVENT_DEV_CFG_PER_DEQUEUE_TIMEOUT) 417 return -ENOTSUP; 418 419 return 0; 420 } 421 422 static void 423 sw_info_get(struct rte_eventdev *dev, struct rte_event_dev_info *info) 424 { 425 RTE_SET_USED(dev); 426 427 static const struct rte_event_dev_info evdev_sw_info = { 428 .driver_name = SW_PMD_NAME, 429 .max_event_queues = RTE_EVENT_MAX_QUEUES_PER_DEV, 430 .max_event_queue_flows = SW_QID_NUM_FIDS, 431 .max_event_queue_priority_levels = SW_Q_PRIORITY_MAX, 432 .max_event_priority_levels = SW_IQS_MAX, 433 .max_event_ports = SW_PORTS_MAX, 434 .max_event_port_dequeue_depth = MAX_SW_CONS_Q_DEPTH, 435 .max_event_port_enqueue_depth = MAX_SW_PROD_Q_DEPTH, 436 .max_num_events = SW_INFLIGHT_EVENTS_TOTAL, 437 .event_dev_cap = (RTE_EVENT_DEV_CAP_QUEUE_QOS | 438 RTE_EVENT_DEV_CAP_EVENT_QOS), 439 }; 440 441 *info = evdev_sw_info; 442 } 443 444 static void 445 sw_dump(struct rte_eventdev *dev, FILE *f) 446 { 447 const struct sw_evdev *sw = sw_pmd_priv(dev); 448 449 static const char * const q_type_strings[] = { 450 "Ordered", "Atomic", "Parallel", "Directed" 451 }; 452 uint32_t i; 453 fprintf(f, "EventDev %s: ports %d, qids %d\n", "todo-fix-name", 454 sw->port_count, sw->qid_count); 455 456 fprintf(f, "\trx %"PRIu64"\n\tdrop %"PRIu64"\n\ttx %"PRIu64"\n", 457 sw->stats.rx_pkts, sw->stats.rx_dropped, sw->stats.tx_pkts); 458 fprintf(f, "\tsched calls: %"PRIu64"\n", sw->sched_called); 459 fprintf(f, "\tsched cq/qid call: %"PRIu64"\n", sw->sched_cq_qid_called); 460 fprintf(f, "\tsched no IQ enq: %"PRIu64"\n", sw->sched_no_iq_enqueues); 461 fprintf(f, "\tsched no CQ enq: %"PRIu64"\n", sw->sched_no_cq_enqueues); 462 uint32_t inflights = rte_atomic32_read(&sw->inflights); 463 uint32_t credits = sw->nb_events_limit - inflights; 464 fprintf(f, "\tinflight %d, credits: %d\n", inflights, credits); 465 466 #define COL_RED "\x1b[31m" 467 #define COL_RESET "\x1b[0m" 468 469 for (i = 0; i < sw->port_count; i++) { 470 int max, j; 471 const struct sw_port *p = &sw->ports[i]; 472 if (!p->initialized) { 473 fprintf(f, " %sPort %d not initialized.%s\n", 474 COL_RED, i, COL_RESET); 475 continue; 476 } 477 fprintf(f, " Port %d %s\n", i, 478 p->is_directed ? " (SingleCons)" : ""); 479 fprintf(f, "\trx %"PRIu64"\tdrop %"PRIu64"\ttx %"PRIu64 480 "\t%sinflight %d%s\n", sw->ports[i].stats.rx_pkts, 481 sw->ports[i].stats.rx_dropped, 482 sw->ports[i].stats.tx_pkts, 483 (p->inflights == p->inflight_max) ? 484 COL_RED : COL_RESET, 485 sw->ports[i].inflights, COL_RESET); 486 487 fprintf(f, "\tMax New: %u" 488 "\tAvg cycles PP: %"PRIu64"\tCredits: %u\n", 489 sw->ports[i].inflight_max, 490 sw->ports[i].avg_pkt_ticks, 491 sw->ports[i].inflight_credits); 492 fprintf(f, "\tReceive burst distribution:\n"); 493 float zp_percent = p->zero_polls * 100.0 / p->total_polls; 494 fprintf(f, zp_percent < 10 ? "\t\t0:%.02f%% " : "\t\t0:%.0f%% ", 495 zp_percent); 496 for (max = (int)RTE_DIM(p->poll_buckets); max-- > 0;) 497 if (p->poll_buckets[max] != 0) 498 break; 499 for (j = 0; j <= max; j++) { 500 if (p->poll_buckets[j] != 0) { 501 float poll_pc = p->poll_buckets[j] * 100.0 / 502 p->total_polls; 503 fprintf(f, "%u-%u:%.02f%% ", 504 ((j << SW_DEQ_STAT_BUCKET_SHIFT) + 1), 505 ((j+1) << SW_DEQ_STAT_BUCKET_SHIFT), 506 poll_pc); 507 } 508 } 509 fprintf(f, "\n"); 510 511 if (p->rx_worker_ring) { 512 uint64_t used = qe_ring_count(p->rx_worker_ring); 513 uint64_t space = qe_ring_free_count(p->rx_worker_ring); 514 const char *col = (space == 0) ? COL_RED : COL_RESET; 515 fprintf(f, "\t%srx ring used: %4"PRIu64"\tfree: %4" 516 PRIu64 COL_RESET"\n", col, used, space); 517 } else 518 fprintf(f, "\trx ring not initialized.\n"); 519 520 if (p->cq_worker_ring) { 521 uint64_t used = qe_ring_count(p->cq_worker_ring); 522 uint64_t space = qe_ring_free_count(p->cq_worker_ring); 523 const char *col = (space == 0) ? COL_RED : COL_RESET; 524 fprintf(f, "\t%scq ring used: %4"PRIu64"\tfree: %4" 525 PRIu64 COL_RESET"\n", col, used, space); 526 } else 527 fprintf(f, "\tcq ring not initialized.\n"); 528 } 529 530 for (i = 0; i < sw->qid_count; i++) { 531 const struct sw_qid *qid = &sw->qids[i]; 532 if (!qid->initialized) { 533 fprintf(f, " %sQueue %d not initialized.%s\n", 534 COL_RED, i, COL_RESET); 535 continue; 536 } 537 int affinities_per_port[SW_PORTS_MAX] = {0}; 538 uint32_t inflights = 0; 539 540 fprintf(f, " Queue %d (%s)\n", i, q_type_strings[qid->type]); 541 fprintf(f, "\trx %"PRIu64"\tdrop %"PRIu64"\ttx %"PRIu64"\n", 542 qid->stats.rx_pkts, qid->stats.rx_dropped, 543 qid->stats.tx_pkts); 544 if (qid->type == RTE_SCHED_TYPE_ORDERED) { 545 struct rte_ring *rob_buf_free = 546 qid->reorder_buffer_freelist; 547 if (rob_buf_free) 548 fprintf(f, "\tReorder entries in use: %u\n", 549 rte_ring_free_count(rob_buf_free)); 550 else 551 fprintf(f, 552 "\tReorder buffer not initialized\n"); 553 } 554 555 uint32_t flow; 556 for (flow = 0; flow < RTE_DIM(qid->fids); flow++) 557 if (qid->fids[flow].cq != -1) { 558 affinities_per_port[qid->fids[flow].cq]++; 559 inflights += qid->fids[flow].pcount; 560 } 561 562 uint32_t cq; 563 fprintf(f, "\tInflights: %u\tFlows pinned per port: ", 564 inflights); 565 for (cq = 0; cq < sw->port_count; cq++) 566 fprintf(f, "%d ", affinities_per_port[cq]); 567 fprintf(f, "\n"); 568 569 uint32_t iq; 570 uint32_t iq_printed = 0; 571 for (iq = 0; iq < SW_IQS_MAX; iq++) { 572 if (!qid->iq[iq]) { 573 fprintf(f, "\tiq %d is not initialized.\n", iq); 574 iq_printed = 1; 575 continue; 576 } 577 uint32_t used = iq_ring_count(qid->iq[iq]); 578 uint32_t free = iq_ring_free_count(qid->iq[iq]); 579 const char *col = (free == 0) ? COL_RED : COL_RESET; 580 if (used > 0) { 581 fprintf(f, "\t%siq %d: Used %d\tFree %d" 582 COL_RESET"\n", col, iq, used, free); 583 iq_printed = 1; 584 } 585 } 586 if (iq_printed == 0) 587 fprintf(f, "\t-- iqs empty --\n"); 588 } 589 } 590 591 static int 592 sw_start(struct rte_eventdev *dev) 593 { 594 unsigned int i, j; 595 struct sw_evdev *sw = sw_pmd_priv(dev); 596 /* check all ports are set up */ 597 for (i = 0; i < sw->port_count; i++) 598 if (sw->ports[i].rx_worker_ring == NULL) { 599 SW_LOG_ERR("Port %d not configured\n", i); 600 return -ESTALE; 601 } 602 603 /* check all queues are configured and mapped to ports*/ 604 for (i = 0; i < sw->qid_count; i++) 605 if (sw->qids[i].iq[0] == NULL || 606 sw->qids[i].cq_num_mapped_cqs == 0) { 607 SW_LOG_ERR("Queue %d not configured\n", i); 608 return -ENOLINK; 609 } 610 611 /* build up our prioritized array of qids */ 612 /* We don't use qsort here, as if all/multiple entries have the same 613 * priority, the result is non-deterministic. From "man 3 qsort": 614 * "If two members compare as equal, their order in the sorted 615 * array is undefined." 616 */ 617 uint32_t qidx = 0; 618 for (j = 0; j <= RTE_EVENT_DEV_PRIORITY_LOWEST; j++) { 619 for (i = 0; i < sw->qid_count; i++) { 620 if (sw->qids[i].priority == j) { 621 sw->qids_prioritized[qidx] = &sw->qids[i]; 622 qidx++; 623 } 624 } 625 } 626 627 if (sw_xstats_init(sw) < 0) 628 return -EINVAL; 629 630 rte_smp_wmb(); 631 sw->started = 1; 632 633 return 0; 634 } 635 636 static void 637 sw_stop(struct rte_eventdev *dev) 638 { 639 struct sw_evdev *sw = sw_pmd_priv(dev); 640 sw_xstats_uninit(sw); 641 sw->started = 0; 642 rte_smp_wmb(); 643 } 644 645 static int 646 sw_close(struct rte_eventdev *dev) 647 { 648 struct sw_evdev *sw = sw_pmd_priv(dev); 649 uint32_t i; 650 651 for (i = 0; i < sw->qid_count; i++) 652 sw_queue_release(dev, i); 653 sw->qid_count = 0; 654 655 for (i = 0; i < sw->port_count; i++) 656 sw_port_release(&sw->ports[i]); 657 sw->port_count = 0; 658 659 memset(&sw->stats, 0, sizeof(sw->stats)); 660 sw->sched_called = 0; 661 sw->sched_no_iq_enqueues = 0; 662 sw->sched_no_cq_enqueues = 0; 663 sw->sched_cq_qid_called = 0; 664 665 return 0; 666 } 667 668 static int 669 assign_numa_node(const char *key __rte_unused, const char *value, void *opaque) 670 { 671 int *socket_id = opaque; 672 *socket_id = atoi(value); 673 if (*socket_id >= RTE_MAX_NUMA_NODES) 674 return -1; 675 return 0; 676 } 677 678 static int 679 set_sched_quanta(const char *key __rte_unused, const char *value, void *opaque) 680 { 681 int *quanta = opaque; 682 *quanta = atoi(value); 683 if (*quanta < 0 || *quanta >= 4096) 684 return -1; 685 return 0; 686 } 687 688 static int 689 set_credit_quanta(const char *key __rte_unused, const char *value, void *opaque) 690 { 691 int *credit = opaque; 692 *credit = atoi(value); 693 if (*credit < 0 || *credit >= 128) 694 return -1; 695 return 0; 696 } 697 698 static int 699 sw_probe(const char *name, const char *params) 700 { 701 static const struct rte_eventdev_ops evdev_sw_ops = { 702 .dev_configure = sw_dev_configure, 703 .dev_infos_get = sw_info_get, 704 .dev_close = sw_close, 705 .dev_start = sw_start, 706 .dev_stop = sw_stop, 707 .dump = sw_dump, 708 709 .queue_def_conf = sw_queue_def_conf, 710 .queue_setup = sw_queue_setup, 711 .queue_release = sw_queue_release, 712 .port_def_conf = sw_port_def_conf, 713 .port_setup = sw_port_setup, 714 .port_release = sw_port_release, 715 .port_link = sw_port_link, 716 .port_unlink = sw_port_unlink, 717 718 .xstats_get = sw_xstats_get, 719 .xstats_get_names = sw_xstats_get_names, 720 .xstats_get_by_name = sw_xstats_get_by_name, 721 .xstats_reset = sw_xstats_reset, 722 }; 723 724 static const char *const args[] = { 725 NUMA_NODE_ARG, 726 SCHED_QUANTA_ARG, 727 CREDIT_QUANTA_ARG, 728 NULL 729 }; 730 struct rte_eventdev *dev; 731 struct sw_evdev *sw; 732 int socket_id = rte_socket_id(); 733 int sched_quanta = SW_DEFAULT_SCHED_QUANTA; 734 int credit_quanta = SW_DEFAULT_CREDIT_QUANTA; 735 736 if (params != NULL && params[0] != '\0') { 737 struct rte_kvargs *kvlist = rte_kvargs_parse(params, args); 738 739 if (!kvlist) { 740 SW_LOG_INFO( 741 "Ignoring unsupported parameters when creating device '%s'\n", 742 name); 743 } else { 744 int ret = rte_kvargs_process(kvlist, NUMA_NODE_ARG, 745 assign_numa_node, &socket_id); 746 if (ret != 0) { 747 SW_LOG_ERR( 748 "%s: Error parsing numa node parameter", 749 name); 750 rte_kvargs_free(kvlist); 751 return ret; 752 } 753 754 ret = rte_kvargs_process(kvlist, SCHED_QUANTA_ARG, 755 set_sched_quanta, &sched_quanta); 756 if (ret != 0) { 757 SW_LOG_ERR( 758 "%s: Error parsing sched quanta parameter", 759 name); 760 rte_kvargs_free(kvlist); 761 return ret; 762 } 763 764 ret = rte_kvargs_process(kvlist, CREDIT_QUANTA_ARG, 765 set_credit_quanta, &credit_quanta); 766 if (ret != 0) { 767 SW_LOG_ERR( 768 "%s: Error parsing credit quanta parameter", 769 name); 770 rte_kvargs_free(kvlist); 771 return ret; 772 } 773 774 rte_kvargs_free(kvlist); 775 } 776 } 777 778 SW_LOG_INFO( 779 "Creating eventdev sw device %s, numa_node=%d, sched_quanta=%d, credit_quanta=%d\n", 780 name, socket_id, sched_quanta, credit_quanta); 781 782 dev = rte_event_pmd_vdev_init(name, 783 sizeof(struct sw_evdev), socket_id); 784 if (dev == NULL) { 785 SW_LOG_ERR("eventdev vdev init() failed"); 786 return -EFAULT; 787 } 788 dev->dev_ops = &evdev_sw_ops; 789 dev->enqueue = sw_event_enqueue; 790 dev->enqueue_burst = sw_event_enqueue_burst; 791 dev->dequeue = sw_event_dequeue; 792 dev->dequeue_burst = sw_event_dequeue_burst; 793 dev->schedule = sw_event_schedule; 794 795 if (rte_eal_process_type() != RTE_PROC_PRIMARY) 796 return 0; 797 798 sw = dev->data->dev_private; 799 sw->data = dev->data; 800 801 /* copy values passed from vdev command line to instance */ 802 sw->credit_update_quanta = credit_quanta; 803 sw->sched_quanta = sched_quanta; 804 805 return 0; 806 } 807 808 static int 809 sw_remove(const char *name) 810 { 811 if (name == NULL) 812 return -EINVAL; 813 814 SW_LOG_INFO("Closing eventdev sw device %s\n", name); 815 816 return rte_event_pmd_vdev_uninit(name); 817 } 818 819 static struct rte_vdev_driver evdev_sw_pmd_drv = { 820 .probe = sw_probe, 821 .remove = sw_remove 822 }; 823 824 RTE_PMD_REGISTER_VDEV(EVENTDEV_NAME_SW_PMD, evdev_sw_pmd_drv); 825 RTE_PMD_REGISTER_PARAM_STRING(event_sw, NUMA_NODE_ARG "=<int> " 826 SCHED_QUANTA_ARG "=<int>" CREDIT_QUANTA_ARG "=<int>"); 827