examples/ip_pipeline/thread.c

   1 /* SPDX-License-Identifier: BSD-3-Clause
   2  * Copyright(c) 2010-2018 Intel Corporation
   3  */
   4
   5 #include <stdlib.h>
   6
   7 #include <rte_common.h>
   8 #include <rte_cycles.h>
   9 #include <rte_lcore.h>
  10 #include <rte_ring.h>
  11
  12 #include <rte_table_acl.h>
  13 #include <rte_table_array.h>
  14 #include <rte_table_hash.h>
  15 #include <rte_table_lpm.h>
  16 #include <rte_table_lpm_ipv6.h>
  17
  18 #include "common.h"
  19 #include "thread.h"
  20 #include "pipeline.h"
  21
  22 #ifndef THREAD_PIPELINES_MAX
  23 #define THREAD_PIPELINES_MAX                               256
  24 #endif
  25
  26 #ifndef THREAD_MSGQ_SIZE
  27 #define THREAD_MSGQ_SIZE                                   64
  28 #endif
  29
  30 #ifndef THREAD_TIMER_PERIOD_MS
  31 #define THREAD_TIMER_PERIOD_MS                             100
  32 #endif
  33
  34 /**
  35  * Master thead: data plane thread context
  36  */
  37 struct thread {
  38         struct rte_ring *msgq_req;
  39         struct rte_ring *msgq_rsp;
  40
  41         uint32_t enabled;
  42 };
  43
  44 static struct thread thread[RTE_MAX_LCORE];
  45
  46 /**
  47  * Data plane threads: context
  48  */
  49 struct table_data {
  50         struct rte_table_action *a;
  51 };
  52
  53 struct pipeline_data {
  54         struct rte_pipeline *p;
  55         struct table_data table_data[RTE_PIPELINE_TABLE_MAX];
  56         uint32_t n_tables;
  57
  58         struct rte_ring *msgq_req;
  59         struct rte_ring *msgq_rsp;
  60         uint64_t timer_period; /* Measured in CPU cycles. */
  61         uint64_t time_next;
  62
  63         uint8_t buffer[TABLE_RULE_ACTION_SIZE_MAX];
  64 };
  65
  66 struct thread_data {
  67         struct rte_pipeline *p[THREAD_PIPELINES_MAX];
  68         uint32_t n_pipelines;
  69
  70         struct pipeline_data pipeline_data[THREAD_PIPELINES_MAX];
  71         struct rte_ring *msgq_req;
  72         struct rte_ring *msgq_rsp;
  73         uint64_t timer_period; /* Measured in CPU cycles. */
  74         uint64_t time_next;
  75         uint64_t time_next_min;
  76 } __rte_cache_aligned;
  77
  78 static struct thread_data thread_data[RTE_MAX_LCORE];
  79
  80 /**
  81  * Master thread: data plane thread init
  82  */
  83 static void
  84 thread_free(void)
  85 {
  86         uint32_t i;
  87
  88         for (i = 0; i < RTE_MAX_LCORE; i++) {
  89                 struct thread *t = &thread[i];
  90
  91                 if (!rte_lcore_is_enabled(i))
  92                         continue;
  93
  94                 /* MSGQs */
  95                 if (t->msgq_req)
  96                         rte_ring_free(t->msgq_req);
  97
  98                 if (t->msgq_rsp)
  99                         rte_ring_free(t->msgq_rsp);
 100         }
 101 }
 102
 103 int
 104 thread_init(void)
 105 {
 106         uint32_t i;
 107
 108         RTE_LCORE_FOREACH_SLAVE(i) {
 109                 char name[NAME_MAX];
 110                 struct rte_ring *msgq_req, *msgq_rsp;
 111                 struct thread *t = &thread[i];
 112                 struct thread_data *t_data = &thread_data[i];
 113                 uint32_t cpu_id = rte_lcore_to_socket_id(i);
 114
 115                 /* MSGQs */
 116                 snprintf(name, sizeof(name), "THREAD-%04x-MSGQ-REQ", i);
 117
 118                 msgq_req = rte_ring_create(name,
 119                         THREAD_MSGQ_SIZE,
 120                         cpu_id,
 121                         RING_F_SP_ENQ | RING_F_SC_DEQ);
 122
 123                 if (msgq_req == NULL) {
 124                         thread_free();
 125                         return -1;
 126                 }
 127
 128                 snprintf(name, sizeof(name), "THREAD-%04x-MSGQ-RSP", i);
 129
 130                 msgq_rsp = rte_ring_create(name,
 131                         THREAD_MSGQ_SIZE,
 132                         cpu_id,
 133                         RING_F_SP_ENQ | RING_F_SC_DEQ);
 134
 135                 if (msgq_rsp == NULL) {
 136                         thread_free();
 137                         return -1;
 138                 }
 139
 140                 /* Master thread records */
 141                 t->msgq_req = msgq_req;
 142                 t->msgq_rsp = msgq_rsp;
 143                 t->enabled = 1;
 144
 145                 /* Data plane thread records */
 146                 t_data->n_pipelines = 0;
 147                 t_data->msgq_req = msgq_req;
 148                 t_data->msgq_rsp = msgq_rsp;
 149                 t_data->timer_period =
 150                         (rte_get_tsc_hz() * THREAD_TIMER_PERIOD_MS) / 1000;
 151                 t_data->time_next = rte_get_tsc_cycles() + t_data->timer_period;
 152                 t_data->time_next_min = t_data->time_next;
 153         }
 154
 155         return 0;
 156 }
 157
 158 /**
 159  * Master thread & data plane threads: message passing
 160  */
 161 enum thread_req_type {
 162         THREAD_REQ_PIPELINE_ENABLE = 0,
 163         THREAD_REQ_PIPELINE_DISABLE,
 164         THREAD_REQ_MAX
 165 };
 166
 167 struct thread_msg_req {
 168         enum thread_req_type type;
 169
 170         union {
 171                 struct {
 172                         struct rte_pipeline *p;
 173                         struct {
 174                                 struct rte_table_action *a;
 175                         } table[RTE_PIPELINE_TABLE_MAX];
 176                         struct rte_ring *msgq_req;
 177                         struct rte_ring *msgq_rsp;
 178                         uint32_t timer_period_ms;
 179                         uint32_t n_tables;
 180                 } pipeline_enable;
 181
 182                 struct {
 183                         struct rte_pipeline *p;
 184                 } pipeline_disable;
 185         };
 186 };
 187
 188 struct thread_msg_rsp {
 189         int status;
 190 };
 191
 192 /**
 193  * Master thread
 194  */
 195 static struct thread_msg_req *
 196 thread_msg_alloc(void)
 197 {
 198         size_t size = RTE_MAX(sizeof(struct thread_msg_req),
 199                 sizeof(struct thread_msg_rsp));
 200
 201         return calloc(1, size);
 202 }
 203
 204 static void
 205 thread_msg_free(struct thread_msg_rsp *rsp)
 206 {
 207         free(rsp);
 208 }
 209
 210 static struct thread_msg_rsp *
 211 thread_msg_send_recv(uint32_t thread_id,
 212         struct thread_msg_req *req)
 213 {
 214         struct thread *t = &thread[thread_id];
 215         struct rte_ring *msgq_req = t->msgq_req;
 216         struct rte_ring *msgq_rsp = t->msgq_rsp;
 217         struct thread_msg_rsp *rsp;
 218         int status;
 219
 220         /* send */
 221         do {
 222                 status = rte_ring_sp_enqueue(msgq_req, req);
 223         } while (status == -ENOBUFS);
 224
 225         /* recv */
 226         do {
 227                 status = rte_ring_sc_dequeue(msgq_rsp, (void **) &rsp);
 228         } while (status != 0);
 229
 230         return rsp;
 231 }
 232
 233 int
 234 thread_pipeline_enable(uint32_t thread_id,
 235         const char *pipeline_name)
 236 {
 237         struct pipeline *p = pipeline_find(pipeline_name);
 238         struct thread *t;
 239         struct thread_msg_req *req;
 240         struct thread_msg_rsp *rsp;
 241         uint32_t i;
 242         int status;
 243
 244         /* Check input params */
 245         if ((thread_id >= RTE_MAX_LCORE) ||
 246                 (p == NULL) ||
 247                 (p->n_ports_in == 0) ||
 248                 (p->n_ports_out == 0) ||
 249                 (p->n_tables == 0))
 250                 return -1;
 251
 252         t = &thread[thread_id];
 253         if ((t->enabled == 0) ||
 254                 p->enabled)
 255                 return -1;
 256
 257         /* Allocate request */
 258         req = thread_msg_alloc();
 259         if (req == NULL)
 260                 return -1;
 261
 262         /* Write request */
 263         req->type = THREAD_REQ_PIPELINE_ENABLE;
 264         req->pipeline_enable.p = p->p;
 265         for (i = 0; i < p->n_tables; i++)
 266                 req->pipeline_enable.table[i].a =
 267                         p->table[i].a;
 268         req->pipeline_enable.msgq_req = p->msgq_req;
 269         req->pipeline_enable.msgq_rsp = p->msgq_rsp;
 270         req->pipeline_enable.timer_period_ms = p->timer_period_ms;
 271         req->pipeline_enable.n_tables = p->n_tables;
 272
 273         /* Send request and wait for response */
 274         rsp = thread_msg_send_recv(thread_id, req);
 275         if (rsp == NULL)
 276                 return -1;
 277
 278         /* Read response */
 279         status = rsp->status;
 280
 281         /* Free response */
 282         thread_msg_free(rsp);
 283
 284         /* Request completion */
 285         if (status)
 286                 return status;
 287
 288         p->thread_id = thread_id;
 289         p->enabled = 1;
 290
 291         return 0;
 292 }
 293
 294 int
 295 thread_pipeline_disable(uint32_t thread_id,
 296         const char *pipeline_name)
 297 {
 298         struct pipeline *p = pipeline_find(pipeline_name);
 299         struct thread *t;
 300         struct thread_msg_req *req;
 301         struct thread_msg_rsp *rsp;
 302         int status;
 303
 304         /* Check input params */
 305         if ((thread_id >= RTE_MAX_LCORE) ||
 306                 (p == NULL))
 307                 return -1;
 308
 309         t = &thread[thread_id];
 310         if (t->enabled == 0)
 311                 return -1;
 312
 313         if (p->enabled == 0)
 314                 return 0;
 315
 316         if (p->thread_id != thread_id)
 317                 return -1;
 318
 319         /* Allocate request */
 320         req = thread_msg_alloc();
 321         if (req == NULL)
 322                 return -1;
 323
 324         /* Write request */
 325         req->type = THREAD_REQ_PIPELINE_DISABLE;
 326         req->pipeline_disable.p = p->p;
 327
 328         /* Send request and wait for response */
 329         rsp = thread_msg_send_recv(thread_id, req);
 330         if (rsp == NULL)
 331                 return -1;
 332
 333         /* Read response */
 334         status = rsp->status;
 335
 336         /* Free response */
 337         thread_msg_free(rsp);
 338
 339         /* Request completion */
 340         if (status)
 341                 return status;
 342
 343         p->enabled = 0;
 344
 345         return 0;
 346 }
 347
 348 /**
 349  * Data plane threads: message handling
 350  */
 351 static inline struct thread_msg_req *
 352 thread_msg_recv(struct rte_ring *msgq_req)
 353 {
 354         struct thread_msg_req *req;
 355
 356         int status = rte_ring_sc_dequeue(msgq_req, (void **) &req);
 357
 358         if (status != 0)
 359                 return NULL;
 360
 361         return req;
 362 }
 363
 364 static inline void
 365 thread_msg_send(struct rte_ring *msgq_rsp,
 366         struct thread_msg_rsp *rsp)
 367 {
 368         int status;
 369
 370         do {
 371                 status = rte_ring_sp_enqueue(msgq_rsp, rsp);
 372         } while (status == -ENOBUFS);
 373 }
 374
 375 static struct thread_msg_rsp *
 376 thread_msg_handle_pipeline_enable(struct thread_data *t,
 377         struct thread_msg_req *req)
 378 {
 379         struct thread_msg_rsp *rsp = (struct thread_msg_rsp *) req;
 380         struct pipeline_data *p = &t->pipeline_data[t->n_pipelines];
 381         uint32_t i;
 382
 383         /* Request */
 384         if (t->n_pipelines >= THREAD_PIPELINES_MAX) {
 385                 rsp->status = -1;
 386                 return rsp;
 387         }
 388
 389         t->p[t->n_pipelines] = req->pipeline_enable.p;
 390
 391         p->p = req->pipeline_enable.p;
 392         for (i = 0; i < req->pipeline_enable.n_tables; i++)
 393                 p->table_data[i].a =
 394                         req->pipeline_enable.table[i].a;
 395
 396         p->n_tables = req->pipeline_enable.n_tables;
 397
 398         p->msgq_req = req->pipeline_enable.msgq_req;
 399         p->msgq_rsp = req->pipeline_enable.msgq_rsp;
 400         p->timer_period =
 401                 (rte_get_tsc_hz() * req->pipeline_enable.timer_period_ms) / 1000;
 402         p->time_next = rte_get_tsc_cycles() + p->timer_period;
 403
 404         t->n_pipelines++;
 405
 406         /* Response */
 407         rsp->status = 0;
 408         return rsp;
 409 }
 410
 411 static struct thread_msg_rsp *
 412 thread_msg_handle_pipeline_disable(struct thread_data *t,
 413         struct thread_msg_req *req)
 414 {
 415         struct thread_msg_rsp *rsp = (struct thread_msg_rsp *) req;
 416         uint32_t n_pipelines = t->n_pipelines;
 417         struct rte_pipeline *pipeline = req->pipeline_disable.p;
 418         uint32_t i;
 419
 420         /* find pipeline */
 421         for (i = 0; i < n_pipelines; i++) {
 422                 struct pipeline_data *p = &t->pipeline_data[i];
 423
 424                 if (p->p != pipeline)
 425                         continue;
 426
 427                 if (i < n_pipelines - 1) {
 428                         struct rte_pipeline *pipeline_last =
 429                                 t->p[n_pipelines - 1];
 430                         struct pipeline_data *p_last =
 431                                 &t->pipeline_data[n_pipelines - 1];
 432
 433                         t->p[i] = pipeline_last;
 434                         memcpy(p, p_last, sizeof(*p));
 435                 }
 436
 437                 t->n_pipelines--;
 438
 439                 rsp->status = 0;
 440                 return rsp;
 441         }
 442
 443         /* should not get here */
 444         rsp->status = 0;
 445         return rsp;
 446 }
 447
 448 static void
 449 thread_msg_handle(struct thread_data *t)
 450 {
 451         for ( ; ; ) {
 452                 struct thread_msg_req *req;
 453                 struct thread_msg_rsp *rsp;
 454
 455                 req = thread_msg_recv(t->msgq_req);
 456                 if (req == NULL)
 457                         break;
 458
 459                 switch (req->type) {
 460                 case THREAD_REQ_PIPELINE_ENABLE:
 461                         rsp = thread_msg_handle_pipeline_enable(t, req);
 462                         break;
 463
 464                 case THREAD_REQ_PIPELINE_DISABLE:
 465                         rsp = thread_msg_handle_pipeline_disable(t, req);
 466                         break;
 467
 468                 default:
 469                         rsp = (struct thread_msg_rsp *) req;
 470                         rsp->status = -1;
 471                 }
 472
 473                 thread_msg_send(t->msgq_rsp, rsp);
 474         }
 475 }
 476
 477 /**
 478  * Master thread & data plane threads: message passing
 479  */
 480 enum pipeline_req_type {
 481         /* Port IN */
 482         PIPELINE_REQ_PORT_IN_ENABLE,
 483         PIPELINE_REQ_PORT_IN_DISABLE,
 484
 485         PIPELINE_REQ_MAX
 486 };
 487
 488 struct pipeline_msg_req {
 489         enum pipeline_req_type type;
 490         uint32_t id; /* Port IN, port OUT or table ID */
 491 };
 492
 493 struct pipeline_msg_rsp {
 494         int status;
 495 };
 496
 497 /**
 498  * Master thread
 499  */
 500 static struct pipeline_msg_req *
 501 pipeline_msg_alloc(void)
 502 {
 503         size_t size = RTE_MAX(sizeof(struct pipeline_msg_req),
 504                 sizeof(struct pipeline_msg_rsp));
 505
 506         return calloc(1, size);
 507 }
 508
 509 static void
 510 pipeline_msg_free(struct pipeline_msg_rsp *rsp)
 511 {
 512         free(rsp);
 513 }
 514
 515 static struct pipeline_msg_rsp *
 516 pipeline_msg_send_recv(struct pipeline *p,
 517         struct pipeline_msg_req *req)
 518 {
 519         struct rte_ring *msgq_req = p->msgq_req;
 520         struct rte_ring *msgq_rsp = p->msgq_rsp;
 521         struct pipeline_msg_rsp *rsp;
 522         int status;
 523
 524         /* send */
 525         do {
 526                 status = rte_ring_sp_enqueue(msgq_req, req);
 527         } while (status == -ENOBUFS);
 528
 529         /* recv */
 530         do {
 531                 status = rte_ring_sc_dequeue(msgq_rsp, (void **) &rsp);
 532         } while (status != 0);
 533
 534         return rsp;
 535 }
 536
 537 int
 538 pipeline_port_in_enable(const char *pipeline_name,
 539         uint32_t port_id)
 540 {
 541         struct pipeline *p;
 542         struct pipeline_msg_req *req;
 543         struct pipeline_msg_rsp *rsp;
 544         int status;
 545
 546         /* Check input params */
 547         if (pipeline_name == NULL)
 548                 return -1;
 549
 550         p = pipeline_find(pipeline_name);
 551         if ((p == NULL) ||
 552                 (p->enabled == 0) ||
 553                 (port_id >= p->n_ports_in))
 554                 return -1;
 555
 556         /* Allocate request */
 557         req = pipeline_msg_alloc();
 558         if (req == NULL)
 559                 return -1;
 560
 561         /* Write request */
 562         req->type = PIPELINE_REQ_PORT_IN_ENABLE;
 563         req->id = port_id;
 564
 565         /* Send request and wait for response */
 566         rsp = pipeline_msg_send_recv(p, req);
 567         if (rsp == NULL)
 568                 return -1;
 569
 570         /* Read response */
 571         status = rsp->status;
 572
 573         /* Free response */
 574         pipeline_msg_free(rsp);
 575
 576         return status;
 577 }
 578
 579 int
 580 pipeline_port_in_disable(const char *pipeline_name,
 581         uint32_t port_id)
 582 {
 583         struct pipeline *p;
 584         struct pipeline_msg_req *req;
 585         struct pipeline_msg_rsp *rsp;
 586         int status;
 587
 588         /* Check input params */
 589         if (pipeline_name == NULL)
 590                 return -1;
 591
 592         p = pipeline_find(pipeline_name);
 593         if ((p == NULL) ||
 594                 (p->enabled == 0) ||
 595                 (port_id >= p->n_ports_in))
 596                 return -1;
 597
 598         /* Allocate request */
 599         req = pipeline_msg_alloc();
 600         if (req == NULL)
 601                 return -1;
 602
 603         /* Write request */
 604         req->type = PIPELINE_REQ_PORT_IN_DISABLE;
 605         req->id = port_id;
 606
 607         /* Send request and wait for response */
 608         rsp = pipeline_msg_send_recv(p, req);
 609         if (rsp == NULL)
 610                 return -1;
 611
 612         /* Read response */
 613         status = rsp->status;
 614
 615         /* Free response */
 616         pipeline_msg_free(rsp);
 617
 618         return status;
 619 }
 620
 621
 622 /**
 623  * Data plane threads: message handling
 624  */
 625 static inline struct pipeline_msg_req *
 626 pipeline_msg_recv(struct rte_ring *msgq_req)
 627 {
 628         struct pipeline_msg_req *req;
 629
 630         int status = rte_ring_sc_dequeue(msgq_req, (void **) &req);
 631
 632         if (status != 0)
 633                 return NULL;
 634
 635         return req;
 636 }
 637
 638 static inline void
 639 pipeline_msg_send(struct rte_ring *msgq_rsp,
 640         struct pipeline_msg_rsp *rsp)
 641 {
 642         int status;
 643
 644         do {
 645                 status = rte_ring_sp_enqueue(msgq_rsp, rsp);
 646         } while (status == -ENOBUFS);
 647 }
 648
 649 static struct pipeline_msg_rsp *
 650 pipeline_msg_handle_port_in_enable(struct pipeline_data *p,
 651         struct pipeline_msg_req *req)
 652 {
 653         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
 654         uint32_t port_id = req->id;
 655
 656         rsp->status = rte_pipeline_port_in_enable(p->p,
 657                 port_id);
 658
 659         return rsp;
 660 }
 661
 662 static struct pipeline_msg_rsp *
 663 pipeline_msg_handle_port_in_disable(struct pipeline_data *p,
 664         struct pipeline_msg_req *req)
 665 {
 666         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
 667         uint32_t port_id = req->id;
 668
 669         rsp->status = rte_pipeline_port_in_disable(p->p,
 670                 port_id);
 671
 672         return rsp;
 673 }
 674
 675 static void
 676 pipeline_msg_handle(struct pipeline_data *p)
 677 {
 678         for ( ; ; ) {
 679                 struct pipeline_msg_req *req;
 680                 struct pipeline_msg_rsp *rsp;
 681
 682                 req = pipeline_msg_recv(p->msgq_req);
 683                 if (req == NULL)
 684                         break;
 685
 686                 switch (req->type) {
 687                 case PIPELINE_REQ_PORT_IN_ENABLE:
 688                         rsp = pipeline_msg_handle_port_in_enable(p, req);
 689                         break;
 690
 691                 case PIPELINE_REQ_PORT_IN_DISABLE:
 692                         rsp = pipeline_msg_handle_port_in_disable(p, req);
 693                         break;
 694
 695                 default:
 696                         rsp = (struct pipeline_msg_rsp *) req;
 697                         rsp->status = -1;
 698                 }
 699
 700                 pipeline_msg_send(p->msgq_rsp, rsp);
 701         }
 702 }
 703
 704 /**
 705  * Data plane threads: main
 706  */
 707 int
 708 thread_main(void *arg __rte_unused)
 709 {
 710         struct thread_data *t;
 711         uint32_t thread_id, i;
 712
 713         thread_id = rte_lcore_id();
 714         t = &thread_data[thread_id];
 715
 716         /* Dispatch loop */
 717         for (i = 0; ; i++) {
 718                 uint32_t j;
 719
 720                 /* Data Plane */
 721                 for (j = 0; j < t->n_pipelines; j++)
 722                         rte_pipeline_run(t->p[j]);
 723
 724                 /* Control Plane */
 725                 if ((i & 0xF) == 0) {
 726                         uint64_t time = rte_get_tsc_cycles();
 727                         uint64_t time_next_min = UINT64_MAX;
 728
 729                         if (time < t->time_next_min)
 730                                 continue;
 731
 732                         /* Pipeline message queues */
 733                         for (j = 0; j < t->n_pipelines; j++) {
 734                                 struct pipeline_data *p =
 735                                         &t->pipeline_data[j];
 736                                 uint64_t time_next = p->time_next;
 737
 738                                 if (time_next <= time) {
 739                                         pipeline_msg_handle(p);
 740                                         rte_pipeline_flush(p->p);
 741                                         time_next = time + p->timer_period;
 742                                         p->time_next = time_next;
 743                                 }
 744
 745                                 if (time_next < time_next_min)
 746                                         time_next_min = time_next;
 747                         }
 748
 749                         /* Thread message queues */
 750                         {
 751                                 uint64_t time_next = t->time_next;
 752
 753                                 if (time_next <= time) {
 754                                         thread_msg_handle(t);
 755                                         time_next = time + t->timer_period;
 756                                         t->time_next = time_next;
 757                                 }
 758
 759                                 if (time_next < time_next_min)
 760                                         time_next_min = time_next;
 761                         }
 762
 763                         t->time_next_min = time_next_min;
 764                 }
 765         }
 766
 767         return 0;
 768 }