drivers/event/dsw/dsw_evdev.c

   1 /* SPDX-License-Identifier: BSD-3-Clause
   2  * Copyright(c) 2018 Ericsson AB
   3  */
   4
   5 #include <stdbool.h>
   6
   7 #include <rte_cycles.h>
   8 #include <rte_eventdev_pmd.h>
   9 #include <rte_eventdev_pmd_vdev.h>
  10 #include <rte_random.h>
  11
  12 #include "dsw_evdev.h"
  13
  14 #define EVENTDEV_NAME_DSW_PMD event_dsw
  15
  16 static int
  17 dsw_port_setup(struct rte_eventdev *dev, uint8_t port_id,
  18                const struct rte_event_port_conf *conf)
  19 {
  20         struct dsw_evdev *dsw = dsw_pmd_priv(dev);
  21         struct dsw_port *port;
  22         struct rte_event_ring *in_ring;
  23         struct rte_ring *ctl_in_ring;
  24         char ring_name[RTE_RING_NAMESIZE];
  25
  26         port = &dsw->ports[port_id];
  27
  28         *port = (struct dsw_port) {
  29                 .id = port_id,
  30                 .dsw = dsw,
  31                 .dequeue_depth = conf->dequeue_depth,
  32                 .enqueue_depth = conf->enqueue_depth,
  33                 .new_event_threshold = conf->new_event_threshold
  34         };
  35
  36         snprintf(ring_name, sizeof(ring_name), "dsw%d_p%u", dev->data->dev_id,
  37                  port_id);
  38
  39         in_ring = rte_event_ring_create(ring_name, DSW_IN_RING_SIZE,
  40                                         dev->data->socket_id,
  41                                         RING_F_SC_DEQ|RING_F_EXACT_SZ);
  42
  43         if (in_ring == NULL)
  44                 return -ENOMEM;
  45
  46         snprintf(ring_name, sizeof(ring_name), "dswctl%d_p%u",
  47                  dev->data->dev_id, port_id);
  48
  49         ctl_in_ring = rte_ring_create(ring_name, DSW_CTL_IN_RING_SIZE,
  50                                       dev->data->socket_id,
  51                                       RING_F_SC_DEQ|RING_F_EXACT_SZ);
  52
  53         if (ctl_in_ring == NULL) {
  54                 rte_event_ring_free(in_ring);
  55                 return -ENOMEM;
  56         }
  57
  58         port->in_ring = in_ring;
  59         port->ctl_in_ring = ctl_in_ring;
  60
  61         rte_atomic16_init(&port->load);
  62
  63         port->load_update_interval =
  64                 (DSW_LOAD_UPDATE_INTERVAL * rte_get_timer_hz()) / US_PER_S;
  65
  66         port->migration_interval =
  67                 (DSW_MIGRATION_INTERVAL * rte_get_timer_hz()) / US_PER_S;
  68
  69         dev->data->ports[port_id] = port;
  70
  71         return 0;
  72 }
  73
  74 static void
  75 dsw_port_def_conf(struct rte_eventdev *dev __rte_unused,
  76                   uint8_t port_id __rte_unused,
  77                   struct rte_event_port_conf *port_conf)
  78 {
  79         *port_conf = (struct rte_event_port_conf) {
  80                 .new_event_threshold = 1024,
  81                 .dequeue_depth = DSW_MAX_PORT_DEQUEUE_DEPTH / 4,
  82                 .enqueue_depth = DSW_MAX_PORT_ENQUEUE_DEPTH / 4
  83         };
  84 }
  85
  86 static void
  87 dsw_port_release(void *p)
  88 {
  89         struct dsw_port *port = p;
  90
  91         rte_event_ring_free(port->in_ring);
  92         rte_ring_free(port->ctl_in_ring);
  93 }
  94
  95 static int
  96 dsw_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
  97                 const struct rte_event_queue_conf *conf)
  98 {
  99         struct dsw_evdev *dsw = dsw_pmd_priv(dev);
 100         struct dsw_queue *queue = &dsw->queues[queue_id];
 101
 102         if (RTE_EVENT_QUEUE_CFG_ALL_TYPES & conf->event_queue_cfg)
 103                 return -ENOTSUP;
 104
 105         if (conf->schedule_type == RTE_SCHED_TYPE_ORDERED)
 106                 return -ENOTSUP;
 107
 108         /* SINGLE_LINK is better off treated as TYPE_ATOMIC, since it
 109          * avoid the "fake" TYPE_PARALLEL flow_id assignment. Since
 110          * the queue will only have a single serving port, no
 111          * migration will ever happen, so the extra TYPE_ATOMIC
 112          * migration overhead is avoided.
 113          */
 114         if (RTE_EVENT_QUEUE_CFG_SINGLE_LINK & conf->event_queue_cfg)
 115                 queue->schedule_type = RTE_SCHED_TYPE_ATOMIC;
 116         else /* atomic or parallel */
 117                 queue->schedule_type = conf->schedule_type;
 118
 119         queue->num_serving_ports = 0;
 120
 121         return 0;
 122 }
 123
 124 static void
 125 dsw_queue_def_conf(struct rte_eventdev *dev __rte_unused,
 126                    uint8_t queue_id __rte_unused,
 127                    struct rte_event_queue_conf *queue_conf)
 128 {
 129         *queue_conf = (struct rte_event_queue_conf) {
 130                 .nb_atomic_flows = 4096,
 131                 .schedule_type = RTE_SCHED_TYPE_ATOMIC,
 132                 .priority = RTE_EVENT_DEV_PRIORITY_NORMAL
 133         };
 134 }
 135
 136 static void
 137 dsw_queue_release(struct rte_eventdev *dev __rte_unused,
 138                   uint8_t queue_id __rte_unused)
 139 {
 140 }
 141
 142 static void
 143 queue_add_port(struct dsw_queue *queue, uint16_t port_id)
 144 {
 145         queue->serving_ports[queue->num_serving_ports] = port_id;
 146         queue->num_serving_ports++;
 147 }
 148
 149 static bool
 150 queue_remove_port(struct dsw_queue *queue, uint16_t port_id)
 151 {
 152         uint16_t i;
 153
 154         for (i = 0; i < queue->num_serving_ports; i++)
 155                 if (queue->serving_ports[i] == port_id) {
 156                         uint16_t last_idx = queue->num_serving_ports - 1;
 157                         if (i != last_idx)
 158                                 queue->serving_ports[i] =
 159                                         queue->serving_ports[last_idx];
 160                         queue->num_serving_ports--;
 161                         return true;
 162                 }
 163         return false;
 164 }
 165
 166 static int
 167 dsw_port_link_unlink(struct rte_eventdev *dev, void *port,
 168                      const uint8_t queues[], uint16_t num, bool link)
 169 {
 170         struct dsw_evdev *dsw = dsw_pmd_priv(dev);
 171         struct dsw_port *p = port;
 172         uint16_t i;
 173         uint16_t count = 0;
 174
 175         for (i = 0; i < num; i++) {
 176                 uint8_t qid = queues[i];
 177                 struct dsw_queue *q = &dsw->queues[qid];
 178                 if (link) {
 179                         queue_add_port(q, p->id);
 180                         count++;
 181                 } else {
 182                         bool removed = queue_remove_port(q, p->id);
 183                         if (removed)
 184                                 count++;
 185                 }
 186         }
 187
 188         return count;
 189 }
 190
 191 static int
 192 dsw_port_link(struct rte_eventdev *dev, void *port, const uint8_t queues[],
 193               const uint8_t priorities[] __rte_unused, uint16_t num)
 194 {
 195         return dsw_port_link_unlink(dev, port, queues, num, true);
 196 }
 197
 198 static int
 199 dsw_port_unlink(struct rte_eventdev *dev, void *port, uint8_t queues[],
 200                 uint16_t num)
 201 {
 202         return dsw_port_link_unlink(dev, port, queues, num, false);
 203 }
 204
 205 static void
 206 dsw_info_get(struct rte_eventdev *dev __rte_unused,
 207              struct rte_event_dev_info *info)
 208 {
 209         *info = (struct rte_event_dev_info) {
 210                 .driver_name = DSW_PMD_NAME,
 211                 .max_event_queues = DSW_MAX_QUEUES,
 212                 .max_event_queue_flows = DSW_MAX_FLOWS,
 213                 .max_event_queue_priority_levels = 1,
 214                 .max_event_priority_levels = 1,
 215                 .max_event_ports = DSW_MAX_PORTS,
 216                 .max_event_port_dequeue_depth = DSW_MAX_PORT_DEQUEUE_DEPTH,
 217                 .max_event_port_enqueue_depth = DSW_MAX_PORT_ENQUEUE_DEPTH,
 218                 .max_num_events = DSW_MAX_EVENTS,
 219                 .event_dev_cap = RTE_EVENT_DEV_CAP_BURST_MODE|
 220                 RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED
 221         };
 222 }
 223
 224 static int
 225 dsw_configure(const struct rte_eventdev *dev)
 226 {
 227         struct dsw_evdev *dsw = dsw_pmd_priv(dev);
 228         const struct rte_event_dev_config *conf = &dev->data->dev_conf;
 229         int32_t min_max_in_flight;
 230
 231         dsw->num_ports = conf->nb_event_ports;
 232         dsw->num_queues = conf->nb_event_queues;
 233
 234         /* Avoid a situation where consumer ports are holding all the
 235          * credits, without making use of them.
 236          */
 237         min_max_in_flight = conf->nb_event_ports * DSW_PORT_MAX_CREDITS;
 238
 239         dsw->max_inflight = RTE_MAX(conf->nb_events_limit, min_max_in_flight);
 240
 241         return 0;
 242 }
 243
 244
 245 static void
 246 initial_flow_to_port_assignment(struct dsw_evdev *dsw)
 247 {
 248         uint8_t queue_id;
 249         for (queue_id = 0; queue_id < dsw->num_queues; queue_id++) {
 250                 struct dsw_queue *queue = &dsw->queues[queue_id];
 251                 uint16_t flow_hash;
 252                 for (flow_hash = 0; flow_hash < DSW_MAX_FLOWS; flow_hash++) {
 253                         uint8_t port_idx =
 254                                 rte_rand() % queue->num_serving_ports;
 255                         uint8_t port_id =
 256                                 queue->serving_ports[port_idx];
 257                         dsw->queues[queue_id].flow_to_port_map[flow_hash] =
 258                                 port_id;
 259                 }
 260         }
 261 }
 262
 263 static int
 264 dsw_start(struct rte_eventdev *dev)
 265 {
 266         struct dsw_evdev *dsw = dsw_pmd_priv(dev);
 267         uint16_t i;
 268         uint64_t now;
 269
 270         rte_atomic32_init(&dsw->credits_on_loan);
 271
 272         initial_flow_to_port_assignment(dsw);
 273
 274         now = rte_get_timer_cycles();
 275         for (i = 0; i < dsw->num_ports; i++) {
 276                 dsw->ports[i].measurement_start = now;
 277                 dsw->ports[i].busy_start = now;
 278         }
 279
 280         return 0;
 281 }
 282
 283 static void
 284 dsw_port_drain_buf(uint8_t dev_id, struct rte_event *buf, uint16_t buf_len,
 285                    eventdev_stop_flush_t flush, void *flush_arg)
 286 {
 287         uint16_t i;
 288
 289         for (i = 0; i < buf_len; i++)
 290                 flush(dev_id, buf[i], flush_arg);
 291 }
 292
 293 static void
 294 dsw_port_drain_paused(uint8_t dev_id, struct dsw_port *port,
 295                       eventdev_stop_flush_t flush, void *flush_arg)
 296 {
 297         dsw_port_drain_buf(dev_id, port->paused_events, port->paused_events_len,
 298                            flush, flush_arg);
 299 }
 300
 301 static void
 302 dsw_port_drain_out(uint8_t dev_id, struct dsw_evdev *dsw, struct dsw_port *port,
 303                    eventdev_stop_flush_t flush, void *flush_arg)
 304 {
 305         uint16_t dport_id;
 306
 307         for (dport_id = 0; dport_id < dsw->num_ports; dport_id++)
 308                 if (dport_id != port->id)
 309                         dsw_port_drain_buf(dev_id, port->out_buffer[dport_id],
 310                                            port->out_buffer_len[dport_id],
 311                                            flush, flush_arg);
 312 }
 313
 314 static void
 315 dsw_port_drain_in_ring(uint8_t dev_id, struct dsw_port *port,
 316                        eventdev_stop_flush_t flush, void *flush_arg)
 317 {
 318         struct rte_event ev;
 319
 320         while (rte_event_ring_dequeue_burst(port->in_ring, &ev, 1, NULL))
 321                 flush(dev_id, ev, flush_arg);
 322 }
 323
 324 static void
 325 dsw_drain(uint8_t dev_id, struct dsw_evdev *dsw,
 326           eventdev_stop_flush_t flush, void *flush_arg)
 327 {
 328         uint16_t port_id;
 329
 330         if (flush == NULL)
 331                 return;
 332
 333         for (port_id = 0; port_id < dsw->num_ports; port_id++) {
 334                 struct dsw_port *port = &dsw->ports[port_id];
 335
 336                 dsw_port_drain_out(dev_id, dsw, port, flush, flush_arg);
 337                 dsw_port_drain_paused(dev_id, port, flush, flush_arg);
 338                 dsw_port_drain_in_ring(dev_id, port, flush, flush_arg);
 339         }
 340 }
 341
 342 static void
 343 dsw_stop(struct rte_eventdev *dev)
 344 {
 345         struct dsw_evdev *dsw = dsw_pmd_priv(dev);
 346         uint8_t dev_id;
 347         eventdev_stop_flush_t flush;
 348         void *flush_arg;
 349
 350         dev_id = dev->data->dev_id;
 351         flush = dev->dev_ops->dev_stop_flush;
 352         flush_arg = dev->data->dev_stop_flush_arg;
 353
 354         dsw_drain(dev_id, dsw, flush, flush_arg);
 355 }
 356
 357 static int
 358 dsw_close(struct rte_eventdev *dev)
 359 {
 360         struct dsw_evdev *dsw = dsw_pmd_priv(dev);
 361
 362         dsw->num_ports = 0;
 363         dsw->num_queues = 0;
 364
 365         return 0;
 366 }
 367
 368 static struct rte_eventdev_ops dsw_evdev_ops = {
 369         .port_setup = dsw_port_setup,
 370         .port_def_conf = dsw_port_def_conf,
 371         .port_release = dsw_port_release,
 372         .queue_setup = dsw_queue_setup,
 373         .queue_def_conf = dsw_queue_def_conf,
 374         .queue_release = dsw_queue_release,
 375         .port_link = dsw_port_link,
 376         .port_unlink = dsw_port_unlink,
 377         .dev_infos_get = dsw_info_get,
 378         .dev_configure = dsw_configure,
 379         .dev_start = dsw_start,
 380         .dev_stop = dsw_stop,
 381         .dev_close = dsw_close,
 382         .xstats_get = dsw_xstats_get,
 383         .xstats_get_names = dsw_xstats_get_names,
 384         .xstats_get_by_name = dsw_xstats_get_by_name
 385 };
 386
 387 static int
 388 dsw_probe(struct rte_vdev_device *vdev)
 389 {
 390         const char *name;
 391         struct rte_eventdev *dev;
 392         struct dsw_evdev *dsw;
 393
 394         name = rte_vdev_device_name(vdev);
 395
 396         dev = rte_event_pmd_vdev_init(name, sizeof(struct dsw_evdev),
 397                                       rte_socket_id());
 398         if (dev == NULL)
 399                 return -EFAULT;
 400
 401         dev->dev_ops = &dsw_evdev_ops;
 402         dev->enqueue = dsw_event_enqueue;
 403         dev->enqueue_burst = dsw_event_enqueue_burst;
 404         dev->enqueue_new_burst = dsw_event_enqueue_new_burst;
 405         dev->enqueue_forward_burst = dsw_event_enqueue_forward_burst;
 406         dev->dequeue = dsw_event_dequeue;
 407         dev->dequeue_burst = dsw_event_dequeue_burst;
 408
 409         if (rte_eal_process_type() != RTE_PROC_PRIMARY)
 410                 return 0;
 411
 412         dsw = dev->data->dev_private;
 413         dsw->data = dev->data;
 414
 415         return 0;
 416 }
 417
 418 static int
 419 dsw_remove(struct rte_vdev_device *vdev)
 420 {
 421         const char *name;
 422
 423         name = rte_vdev_device_name(vdev);
 424         if (name == NULL)
 425                 return -EINVAL;
 426
 427         return rte_event_pmd_vdev_uninit(name);
 428 }
 429
 430 static struct rte_vdev_driver evdev_dsw_pmd_drv = {
 431         .probe = dsw_probe,
 432         .remove = dsw_remove
 433 };
 434
 435 RTE_PMD_REGISTER_VDEV(EVENTDEV_NAME_DSW_PMD, evdev_dsw_pmd_drv);