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_STATS_READ,
 483         PIPELINE_REQ_PORT_IN_ENABLE,
 484         PIPELINE_REQ_PORT_IN_DISABLE,
 485
 486         /* Port OUT */
 487         PIPELINE_REQ_PORT_OUT_STATS_READ,
 488
 489         /* Table */
 490         PIPELINE_REQ_TABLE_STATS_READ,
 491         PIPELINE_REQ_TABLE_RULE_ADD,
 492         PIPELINE_REQ_TABLE_RULE_ADD_DEFAULT,
 493         PIPELINE_REQ_TABLE_RULE_ADD_BULK,
 494         PIPELINE_REQ_TABLE_RULE_DELETE,
 495         PIPELINE_REQ_TABLE_RULE_DELETE_DEFAULT,
 496         PIPELINE_REQ_TABLE_RULE_STATS_READ,
 497         PIPELINE_REQ_MAX
 498 };
 499
 500 struct pipeline_msg_req_port_in_stats_read {
 501         int clear;
 502 };
 503
 504 struct pipeline_msg_req_port_out_stats_read {
 505         int clear;
 506 };
 507
 508 struct pipeline_msg_req_table_stats_read {
 509         int clear;
 510 };
 511
 512 struct pipeline_msg_req_table_rule_add {
 513         struct table_rule_match match;
 514         struct table_rule_action action;
 515 };
 516
 517 struct pipeline_msg_req_table_rule_add_default {
 518         struct table_rule_action action;
 519 };
 520
 521 struct pipeline_msg_req_table_rule_add_bulk {
 522         struct table_rule_match *match;
 523         struct table_rule_action *action;
 524         void **data;
 525         uint32_t n_rules;
 526         int bulk;
 527 };
 528
 529 struct pipeline_msg_req_table_rule_delete {
 530         struct table_rule_match match;
 531 };
 532
 533 struct pipeline_msg_req_table_rule_stats_read {
 534         void *data;
 535         int clear;
 536 };
 537
 538 struct pipeline_msg_req {
 539         enum pipeline_req_type type;
 540         uint32_t id; /* Port IN, port OUT or table ID */
 541
 542         RTE_STD_C11
 543         union {
 544                 struct pipeline_msg_req_port_in_stats_read port_in_stats_read;
 545                 struct pipeline_msg_req_port_out_stats_read port_out_stats_read;
 546                 struct pipeline_msg_req_table_stats_read table_stats_read;
 547                 struct pipeline_msg_req_table_rule_add table_rule_add;
 548                 struct pipeline_msg_req_table_rule_add_default table_rule_add_default;
 549                 struct pipeline_msg_req_table_rule_add_bulk table_rule_add_bulk;
 550                 struct pipeline_msg_req_table_rule_delete table_rule_delete;
 551                 struct pipeline_msg_req_table_rule_stats_read table_rule_stats_read;
 552         };
 553 };
 554
 555 struct pipeline_msg_rsp_port_in_stats_read {
 556         struct rte_pipeline_port_in_stats stats;
 557 };
 558
 559 struct pipeline_msg_rsp_port_out_stats_read {
 560         struct rte_pipeline_port_out_stats stats;
 561 };
 562
 563 struct pipeline_msg_rsp_table_stats_read {
 564         struct rte_pipeline_table_stats stats;
 565 };
 566
 567 struct pipeline_msg_rsp_table_rule_add {
 568         void *data;
 569 };
 570
 571 struct pipeline_msg_rsp_table_rule_add_default {
 572         void *data;
 573 };
 574
 575 struct pipeline_msg_rsp_table_rule_add_bulk {
 576         uint32_t n_rules;
 577 };
 578
 579 struct pipeline_msg_rsp_table_rule_stats_read {
 580         struct rte_table_action_stats_counters stats;
 581 };
 582
 583 struct pipeline_msg_rsp {
 584         int status;
 585
 586         RTE_STD_C11
 587         union {
 588                 struct pipeline_msg_rsp_port_in_stats_read port_in_stats_read;
 589                 struct pipeline_msg_rsp_port_out_stats_read port_out_stats_read;
 590                 struct pipeline_msg_rsp_table_stats_read table_stats_read;
 591                 struct pipeline_msg_rsp_table_rule_add table_rule_add;
 592                 struct pipeline_msg_rsp_table_rule_add_default table_rule_add_default;
 593                 struct pipeline_msg_rsp_table_rule_add_bulk table_rule_add_bulk;
 594                 struct pipeline_msg_rsp_table_rule_stats_read table_rule_stats_read;
 595         };
 596 };
 597
 598 /**
 599  * Master thread
 600  */
 601 static struct pipeline_msg_req *
 602 pipeline_msg_alloc(void)
 603 {
 604         size_t size = RTE_MAX(sizeof(struct pipeline_msg_req),
 605                 sizeof(struct pipeline_msg_rsp));
 606
 607         return calloc(1, size);
 608 }
 609
 610 static void
 611 pipeline_msg_free(struct pipeline_msg_rsp *rsp)
 612 {
 613         free(rsp);
 614 }
 615
 616 static struct pipeline_msg_rsp *
 617 pipeline_msg_send_recv(struct pipeline *p,
 618         struct pipeline_msg_req *req)
 619 {
 620         struct rte_ring *msgq_req = p->msgq_req;
 621         struct rte_ring *msgq_rsp = p->msgq_rsp;
 622         struct pipeline_msg_rsp *rsp;
 623         int status;
 624
 625         /* send */
 626         do {
 627                 status = rte_ring_sp_enqueue(msgq_req, req);
 628         } while (status == -ENOBUFS);
 629
 630         /* recv */
 631         do {
 632                 status = rte_ring_sc_dequeue(msgq_rsp, (void **) &rsp);
 633         } while (status != 0);
 634
 635         return rsp;
 636 }
 637
 638 int
 639 pipeline_port_in_stats_read(const char *pipeline_name,
 640         uint32_t port_id,
 641         struct rte_pipeline_port_in_stats *stats,
 642         int clear)
 643 {
 644         struct pipeline *p;
 645         struct pipeline_msg_req *req;
 646         struct pipeline_msg_rsp *rsp;
 647         int status;
 648
 649         /* Check input params */
 650         if ((pipeline_name == NULL) ||
 651                 (stats == NULL))
 652                 return -1;
 653
 654         p = pipeline_find(pipeline_name);
 655         if ((p == NULL) ||
 656                 (p->enabled == 0) ||
 657                 (port_id >= p->n_ports_in))
 658                 return -1;
 659
 660         /* Allocate request */
 661         req = pipeline_msg_alloc();
 662         if (req == NULL)
 663                 return -1;
 664
 665         /* Write request */
 666         req->type = PIPELINE_REQ_PORT_IN_STATS_READ;
 667         req->id = port_id;
 668         req->port_in_stats_read.clear = clear;
 669
 670         /* Send request and wait for response */
 671         rsp = pipeline_msg_send_recv(p, req);
 672         if (rsp == NULL)
 673                 return -1;
 674
 675         /* Read response */
 676         status = rsp->status;
 677         if (status)
 678                 memcpy(stats, &rsp->port_in_stats_read.stats, sizeof(*stats));
 679
 680         /* Free response */
 681         pipeline_msg_free(rsp);
 682
 683         return status;
 684 }
 685
 686 int
 687 pipeline_port_in_enable(const char *pipeline_name,
 688         uint32_t port_id)
 689 {
 690         struct pipeline *p;
 691         struct pipeline_msg_req *req;
 692         struct pipeline_msg_rsp *rsp;
 693         int status;
 694
 695         /* Check input params */
 696         if (pipeline_name == NULL)
 697                 return -1;
 698
 699         p = pipeline_find(pipeline_name);
 700         if ((p == NULL) ||
 701                 (p->enabled == 0) ||
 702                 (port_id >= p->n_ports_in))
 703                 return -1;
 704
 705         /* Allocate request */
 706         req = pipeline_msg_alloc();
 707         if (req == NULL)
 708                 return -1;
 709
 710         /* Write request */
 711         req->type = PIPELINE_REQ_PORT_IN_ENABLE;
 712         req->id = port_id;
 713
 714         /* Send request and wait for response */
 715         rsp = pipeline_msg_send_recv(p, req);
 716         if (rsp == NULL)
 717                 return -1;
 718
 719         /* Read response */
 720         status = rsp->status;
 721
 722         /* Free response */
 723         pipeline_msg_free(rsp);
 724
 725         return status;
 726 }
 727
 728 int
 729 pipeline_port_in_disable(const char *pipeline_name,
 730         uint32_t port_id)
 731 {
 732         struct pipeline *p;
 733         struct pipeline_msg_req *req;
 734         struct pipeline_msg_rsp *rsp;
 735         int status;
 736
 737         /* Check input params */
 738         if (pipeline_name == NULL)
 739                 return -1;
 740
 741         p = pipeline_find(pipeline_name);
 742         if ((p == NULL) ||
 743                 (p->enabled == 0) ||
 744                 (port_id >= p->n_ports_in))
 745                 return -1;
 746
 747         /* Allocate request */
 748         req = pipeline_msg_alloc();
 749         if (req == NULL)
 750                 return -1;
 751
 752         /* Write request */
 753         req->type = PIPELINE_REQ_PORT_IN_DISABLE;
 754         req->id = port_id;
 755
 756         /* Send request and wait for response */
 757         rsp = pipeline_msg_send_recv(p, req);
 758         if (rsp == NULL)
 759                 return -1;
 760
 761         /* Read response */
 762         status = rsp->status;
 763
 764         /* Free response */
 765         pipeline_msg_free(rsp);
 766
 767         return status;
 768 }
 769
 770 int
 771 pipeline_port_out_stats_read(const char *pipeline_name,
 772         uint32_t port_id,
 773         struct rte_pipeline_port_out_stats *stats,
 774         int clear)
 775 {
 776         struct pipeline *p;
 777         struct pipeline_msg_req *req;
 778         struct pipeline_msg_rsp *rsp;
 779         int status;
 780
 781         /* Check input params */
 782         if ((pipeline_name == NULL) ||
 783                 (stats == NULL))
 784                 return -1;
 785
 786         p = pipeline_find(pipeline_name);
 787         if ((p == NULL) ||
 788                 (p->enabled == 0) ||
 789                 (port_id >= p->n_ports_out))
 790                 return -1;
 791
 792         /* Allocate request */
 793         req = pipeline_msg_alloc();
 794         if (req == NULL)
 795                 return -1;
 796
 797         /* Write request */
 798         req->type = PIPELINE_REQ_PORT_OUT_STATS_READ;
 799         req->id = port_id;
 800         req->port_out_stats_read.clear = clear;
 801
 802         /* Send request and wait for response */
 803         rsp = pipeline_msg_send_recv(p, req);
 804         if (rsp == NULL)
 805                 return -1;
 806
 807         /* Read response */
 808         status = rsp->status;
 809         if (status)
 810                 memcpy(stats, &rsp->port_out_stats_read.stats, sizeof(*stats));
 811
 812         /* Free response */
 813         pipeline_msg_free(rsp);
 814
 815         return status;
 816 }
 817
 818 int
 819 pipeline_table_stats_read(const char *pipeline_name,
 820         uint32_t table_id,
 821         struct rte_pipeline_table_stats *stats,
 822         int clear)
 823 {
 824         struct pipeline *p;
 825         struct pipeline_msg_req *req;
 826         struct pipeline_msg_rsp *rsp;
 827         int status;
 828
 829         /* Check input params */
 830         if ((pipeline_name == NULL) ||
 831                 (stats == NULL))
 832                 return -1;
 833
 834         p = pipeline_find(pipeline_name);
 835         if ((p == NULL) ||
 836                 (p->enabled == 0) ||
 837                 (table_id >= p->n_tables))
 838                 return -1;
 839
 840         /* Allocate request */
 841         req = pipeline_msg_alloc();
 842         if (req == NULL)
 843                 return -1;
 844
 845         /* Write request */
 846         req->type = PIPELINE_REQ_TABLE_STATS_READ;
 847         req->id = table_id;
 848         req->table_stats_read.clear = clear;
 849
 850         /* Send request and wait for response */
 851         rsp = pipeline_msg_send_recv(p, req);
 852         if (rsp == NULL)
 853                 return -1;
 854
 855         /* Read response */
 856         status = rsp->status;
 857         if (status)
 858                 memcpy(stats, &rsp->table_stats_read.stats, sizeof(*stats));
 859
 860         /* Free response */
 861         pipeline_msg_free(rsp);
 862
 863         return status;
 864 }
 865
 866 static int
 867 match_check(struct table_rule_match *match,
 868         struct pipeline *p,
 869         uint32_t table_id)
 870 {
 871         struct table *table;
 872
 873         if ((match == NULL) ||
 874                 (p == NULL) ||
 875                 (table_id >= p->n_tables))
 876                 return -1;
 877
 878         table = &p->table[table_id];
 879         if (match->match_type != table->params.match_type)
 880                 return -1;
 881
 882         switch (match->match_type) {
 883         case TABLE_ACL:
 884         {
 885                 struct table_acl_params *t = &table->params.match.acl;
 886                 struct table_rule_match_acl *r = &match->match.acl;
 887
 888                 if ((r->ip_version && (t->ip_version == 0)) ||
 889                         ((r->ip_version == 0) && t->ip_version))
 890                         return -1;
 891
 892                 if (r->ip_version) {
 893                         if ((r->sa_depth > 32) ||
 894                                 (r->da_depth > 32))
 895                                 return -1;
 896                 } else {
 897                         if ((r->sa_depth > 128) ||
 898                                 (r->da_depth > 128))
 899                                 return -1;
 900                 }
 901                 return 0;
 902         }
 903
 904         case TABLE_ARRAY:
 905                 return 0;
 906
 907         case TABLE_HASH:
 908                 return 0;
 909
 910         case TABLE_LPM:
 911         {
 912                 struct table_lpm_params *t = &table->params.match.lpm;
 913                 struct table_rule_match_lpm *r = &match->match.lpm;
 914
 915                 if ((r->ip_version && (t->key_size != 4)) ||
 916                         ((r->ip_version == 0) && (t->key_size != 16)))
 917                         return -1;
 918
 919                 if (r->ip_version) {
 920                         if (r->depth > 32)
 921                                 return -1;
 922                 } else {
 923                         if (r->depth > 128)
 924                                 return -1;
 925                 }
 926                 return 0;
 927         }
 928
 929         case TABLE_STUB:
 930                 return -1;
 931
 932         default:
 933                 return -1;
 934         }
 935 }
 936
 937 static int
 938 action_check(struct table_rule_action *action,
 939         struct pipeline *p,
 940         uint32_t table_id)
 941 {
 942         struct table_action_profile *ap;
 943
 944         if ((action == NULL) ||
 945                 (p == NULL) ||
 946                 (table_id >= p->n_tables))
 947                 return -1;
 948
 949         ap = p->table[table_id].ap;
 950         if (action->action_mask != ap->params.action_mask)
 951                 return -1;
 952
 953         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_FWD)) {
 954                 if ((action->fwd.action == RTE_PIPELINE_ACTION_PORT) &&
 955                         (action->fwd.id >= p->n_ports_out))
 956                         return -1;
 957
 958                 if ((action->fwd.action == RTE_PIPELINE_ACTION_TABLE) &&
 959                         (action->fwd.id >= p->n_tables))
 960                         return -1;
 961         }
 962
 963         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_MTR)) {
 964                 uint32_t tc_mask0 = (1 << ap->params.mtr.n_tc) - 1;
 965                 uint32_t tc_mask1 = action->mtr.tc_mask;
 966
 967                 if (tc_mask1 != tc_mask0)
 968                         return -1;
 969         }
 970
 971         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_TM)) {
 972                 uint32_t n_subports_per_port =
 973                         ap->params.tm.n_subports_per_port;
 974                 uint32_t n_pipes_per_subport =
 975                         ap->params.tm.n_pipes_per_subport;
 976                 uint32_t subport_id = action->tm.subport_id;
 977                 uint32_t pipe_id = action->tm.pipe_id;
 978
 979                 if ((subport_id >= n_subports_per_port) ||
 980                         (pipe_id >= n_pipes_per_subport))
 981                         return -1;
 982         }
 983
 984         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_ENCAP)) {
 985                 uint64_t encap_mask = ap->params.encap.encap_mask;
 986                 enum rte_table_action_encap_type type = action->encap.type;
 987
 988                 if ((encap_mask & (1LLU << type)) == 0)
 989                         return -1;
 990         }
 991
 992         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_NAT)) {
 993                 int ip_version0 = ap->params.common.ip_version;
 994                 int ip_version1 = action->nat.ip_version;
 995
 996                 if ((ip_version1 && (ip_version0 == 0)) ||
 997                         ((ip_version1 == 0) && ip_version0))
 998                         return -1;
 999         }
1000
1001         return 0;
1002 }
1003
1004 static int
1005 action_default_check(struct table_rule_action *action,
1006         struct pipeline *p,
1007         uint32_t table_id)
1008 {
1009         if ((action == NULL) ||
1010                 (action->action_mask != (1LLU << RTE_TABLE_ACTION_FWD)) ||
1011                 (p == NULL) ||
1012                 (table_id >= p->n_tables))
1013                 return -1;
1014
1015         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_FWD)) {
1016                 if ((action->fwd.action == RTE_PIPELINE_ACTION_PORT) &&
1017                         (action->fwd.id >= p->n_ports_out))
1018                         return -1;
1019
1020                 if ((action->fwd.action == RTE_PIPELINE_ACTION_TABLE) &&
1021                         (action->fwd.id >= p->n_tables))
1022                         return -1;
1023         }
1024
1025         return 0;
1026 }
1027
1028 int
1029 pipeline_table_rule_add(const char *pipeline_name,
1030         uint32_t table_id,
1031         struct table_rule_match *match,
1032         struct table_rule_action *action,
1033         void **data)
1034 {
1035         struct pipeline *p;
1036         struct pipeline_msg_req *req;
1037         struct pipeline_msg_rsp *rsp;
1038         int status;
1039
1040         /* Check input params */
1041         if ((pipeline_name == NULL) ||
1042                 (match == NULL) ||
1043                 (action == NULL) ||
1044                 (data == NULL))
1045                 return -1;
1046
1047         p = pipeline_find(pipeline_name);
1048         if ((p == NULL) ||
1049                 (p->enabled == 0) ||
1050                 (table_id >= p->n_tables) ||
1051                 match_check(match, p, table_id) ||
1052                 action_check(action, p, table_id))
1053                 return -1;
1054
1055         /* Allocate request */
1056         req = pipeline_msg_alloc();
1057         if (req == NULL)
1058                 return -1;
1059
1060         /* Write request */
1061         req->type = PIPELINE_REQ_TABLE_RULE_ADD;
1062         req->id = table_id;
1063         memcpy(&req->table_rule_add.match, match, sizeof(*match));
1064         memcpy(&req->table_rule_add.action, action, sizeof(*action));
1065
1066         /* Send request and wait for response */
1067         rsp = pipeline_msg_send_recv(p, req);
1068         if (rsp == NULL)
1069                 return -1;
1070
1071         /* Read response */
1072         status = rsp->status;
1073         if (status == 0)
1074                 *data = rsp->table_rule_add.data;
1075
1076         /* Free response */
1077         pipeline_msg_free(rsp);
1078
1079         return status;
1080 }
1081
1082 int
1083 pipeline_table_rule_add_default(const char *pipeline_name,
1084         uint32_t table_id,
1085         struct table_rule_action *action,
1086         void **data)
1087 {
1088         struct pipeline *p;
1089         struct pipeline_msg_req *req;
1090         struct pipeline_msg_rsp *rsp;
1091         int status;
1092
1093         /* Check input params */
1094         if ((pipeline_name == NULL) ||
1095                 (action == NULL) ||
1096                 (data == NULL))
1097                 return -1;
1098
1099         p = pipeline_find(pipeline_name);
1100         if ((p == NULL) ||
1101                 (p->enabled == 0) ||
1102                 (table_id >= p->n_tables) ||
1103                 action_default_check(action, p, table_id))
1104                 return -1;
1105
1106         /* Allocate request */
1107         req = pipeline_msg_alloc();
1108         if (req == NULL)
1109                 return -1;
1110
1111         /* Write request */
1112         req->type = PIPELINE_REQ_TABLE_RULE_ADD_DEFAULT;
1113         req->id = table_id;
1114         memcpy(&req->table_rule_add_default.action, action, sizeof(*action));
1115
1116         /* Send request and wait for response */
1117         rsp = pipeline_msg_send_recv(p, req);
1118         if (rsp == NULL)
1119                 return -1;
1120
1121         /* Read response */
1122         status = rsp->status;
1123         if (status == 0)
1124                 *data = rsp->table_rule_add_default.data;
1125
1126         /* Free response */
1127         pipeline_msg_free(rsp);
1128
1129         return status;
1130 }
1131
1132 int
1133 pipeline_table_rule_add_bulk(const char *pipeline_name,
1134         uint32_t table_id,
1135         struct table_rule_match *match,
1136         struct table_rule_action *action,
1137         void **data,
1138         uint32_t *n_rules)
1139 {
1140         struct pipeline *p;
1141         struct pipeline_msg_req *req;
1142         struct pipeline_msg_rsp *rsp;
1143         uint32_t i;
1144         int status;
1145
1146         /* Check input params */
1147         if ((pipeline_name == NULL) ||
1148                 (match == NULL) ||
1149                 (action == NULL) ||
1150                 (data == NULL) ||
1151                 (n_rules == NULL) ||
1152                 (*n_rules == 0))
1153                 return -1;
1154
1155         p = pipeline_find(pipeline_name);
1156         if ((p == NULL) ||
1157                 (p->enabled == 0) ||
1158                 (table_id >= p->n_tables))
1159                 return -1;
1160
1161         for (i = 0; i < *n_rules; i++)
1162                 if (match_check(match, p, table_id) ||
1163                         action_check(action, p, table_id))
1164                         return -1;
1165
1166         /* Allocate request */
1167         req = pipeline_msg_alloc();
1168         if (req == NULL)
1169                 return -1;
1170
1171         /* Write request */
1172         req->type = PIPELINE_REQ_TABLE_RULE_ADD_BULK;
1173         req->id = table_id;
1174         req->table_rule_add_bulk.match = match;
1175         req->table_rule_add_bulk.action = action;
1176         req->table_rule_add_bulk.data = data;
1177         req->table_rule_add_bulk.n_rules = *n_rules;
1178         req->table_rule_add_bulk.bulk =
1179                 (p->table[table_id].params.match_type == TABLE_ACL) ? 1 : 0;
1180
1181         /* Send request and wait for response */
1182         rsp = pipeline_msg_send_recv(p, req);
1183         if (rsp == NULL)
1184                 return -1;
1185
1186         /* Read response */
1187         status = rsp->status;
1188         if (status == 0)
1189                 *n_rules = rsp->table_rule_add_bulk.n_rules;
1190
1191         /* Free response */
1192         pipeline_msg_free(rsp);
1193
1194         return status;
1195 }
1196
1197 int
1198 pipeline_table_rule_delete(const char *pipeline_name,
1199         uint32_t table_id,
1200         struct table_rule_match *match)
1201 {
1202         struct pipeline *p;
1203         struct pipeline_msg_req *req;
1204         struct pipeline_msg_rsp *rsp;
1205         int status;
1206
1207         /* Check input params */
1208         if ((pipeline_name == NULL) ||
1209                 (match == NULL))
1210                 return -1;
1211
1212         p = pipeline_find(pipeline_name);
1213         if ((p == NULL) ||
1214                 (p->enabled == 0) ||
1215                 (table_id >= p->n_tables) ||
1216                 match_check(match, p, table_id))
1217                 return -1;
1218
1219         /* Allocate request */
1220         req = pipeline_msg_alloc();
1221         if (req == NULL)
1222                 return -1;
1223
1224         /* Write request */
1225         req->type = PIPELINE_REQ_TABLE_RULE_DELETE;
1226         req->id = table_id;
1227         memcpy(&req->table_rule_delete.match, match, sizeof(*match));
1228
1229         /* Send request and wait for response */
1230         rsp = pipeline_msg_send_recv(p, req);
1231         if (rsp == NULL)
1232                 return -1;
1233
1234         /* Read response */
1235         status = rsp->status;
1236
1237         /* Free response */
1238         pipeline_msg_free(rsp);
1239
1240         return status;
1241 }
1242
1243 int
1244 pipeline_table_rule_delete_default(const char *pipeline_name,
1245         uint32_t table_id)
1246 {
1247         struct pipeline *p;
1248         struct pipeline_msg_req *req;
1249         struct pipeline_msg_rsp *rsp;
1250         int status;
1251
1252         /* Check input params */
1253         if (pipeline_name == NULL)
1254                 return -1;
1255
1256         p = pipeline_find(pipeline_name);
1257         if ((p == NULL) ||
1258                 (p->enabled == 0) ||
1259                 (table_id >= p->n_tables))
1260                 return -1;
1261
1262         /* Allocate request */
1263         req = pipeline_msg_alloc();
1264         if (req == NULL)
1265                 return -1;
1266
1267         /* Write request */
1268         req->type = PIPELINE_REQ_TABLE_RULE_DELETE_DEFAULT;
1269         req->id = table_id;
1270
1271         /* Send request and wait for response */
1272         rsp = pipeline_msg_send_recv(p, req);
1273         if (rsp == NULL)
1274                 return -1;
1275
1276         /* Read response */
1277         status = rsp->status;
1278
1279         /* Free response */
1280         pipeline_msg_free(rsp);
1281
1282         return status;
1283 }
1284
1285 int
1286 pipeline_table_rule_stats_read(const char *pipeline_name,
1287         uint32_t table_id,
1288         void *data,
1289         struct rte_table_action_stats_counters *stats,
1290         int clear)
1291 {
1292         struct pipeline *p;
1293         struct pipeline_msg_req *req;
1294         struct pipeline_msg_rsp *rsp;
1295         int status;
1296
1297         /* Check input params */
1298         if ((pipeline_name == NULL) ||
1299                 (data == NULL) ||
1300                 (stats == NULL))
1301                 return -1;
1302
1303         p = pipeline_find(pipeline_name);
1304         if ((p == NULL) ||
1305                 (p->enabled == 0) ||
1306                 (table_id >= p->n_tables))
1307                 return -1;
1308
1309         /* Allocate request */
1310         req = pipeline_msg_alloc();
1311         if (req == NULL)
1312                 return -1;
1313
1314         /* Write request */
1315         req->type = PIPELINE_REQ_TABLE_RULE_STATS_READ;
1316         req->id = table_id;
1317         req->table_rule_stats_read.data = data;
1318         req->table_rule_stats_read.clear = clear;
1319
1320         /* Send request and wait for response */
1321         rsp = pipeline_msg_send_recv(p, req);
1322         if (rsp == NULL)
1323                 return -1;
1324
1325         /* Read response */
1326         status = rsp->status;
1327         if (status)
1328                 memcpy(stats, &rsp->table_rule_stats_read.stats, sizeof(*stats));
1329
1330         /* Free response */
1331         pipeline_msg_free(rsp);
1332
1333         return status;
1334 }
1335
1336 /**
1337  * Data plane threads: message handling
1338  */
1339 static inline struct pipeline_msg_req *
1340 pipeline_msg_recv(struct rte_ring *msgq_req)
1341 {
1342         struct pipeline_msg_req *req;
1343
1344         int status = rte_ring_sc_dequeue(msgq_req, (void **) &req);
1345
1346         if (status != 0)
1347                 return NULL;
1348
1349         return req;
1350 }
1351
1352 static inline void
1353 pipeline_msg_send(struct rte_ring *msgq_rsp,
1354         struct pipeline_msg_rsp *rsp)
1355 {
1356         int status;
1357
1358         do {
1359                 status = rte_ring_sp_enqueue(msgq_rsp, rsp);
1360         } while (status == -ENOBUFS);
1361 }
1362
1363 static struct pipeline_msg_rsp *
1364 pipeline_msg_handle_port_in_stats_read(struct pipeline_data *p,
1365         struct pipeline_msg_req *req)
1366 {
1367         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
1368         uint32_t port_id = req->id;
1369         int clear = req->port_in_stats_read.clear;
1370
1371         rsp->status = rte_pipeline_port_in_stats_read(p->p,
1372                 port_id,
1373                 &rsp->port_in_stats_read.stats,
1374                 clear);
1375
1376         return rsp;
1377 }
1378
1379 static struct pipeline_msg_rsp *
1380 pipeline_msg_handle_port_in_enable(struct pipeline_data *p,
1381         struct pipeline_msg_req *req)
1382 {
1383         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
1384         uint32_t port_id = req->id;
1385
1386         rsp->status = rte_pipeline_port_in_enable(p->p,
1387                 port_id);
1388
1389         return rsp;
1390 }
1391
1392 static struct pipeline_msg_rsp *
1393 pipeline_msg_handle_port_in_disable(struct pipeline_data *p,
1394         struct pipeline_msg_req *req)
1395 {
1396         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
1397         uint32_t port_id = req->id;
1398
1399         rsp->status = rte_pipeline_port_in_disable(p->p,
1400                 port_id);
1401
1402         return rsp;
1403 }
1404
1405 static struct pipeline_msg_rsp *
1406 pipeline_msg_handle_port_out_stats_read(struct pipeline_data *p,
1407         struct pipeline_msg_req *req)
1408 {
1409         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
1410         uint32_t port_id = req->id;
1411         int clear = req->port_out_stats_read.clear;
1412
1413         rsp->status = rte_pipeline_port_out_stats_read(p->p,
1414                 port_id,
1415                 &rsp->port_out_stats_read.stats,
1416                 clear);
1417
1418         return rsp;
1419 }
1420
1421 static struct pipeline_msg_rsp *
1422 pipeline_msg_handle_table_stats_read(struct pipeline_data *p,
1423         struct pipeline_msg_req *req)
1424 {
1425         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
1426         uint32_t port_id = req->id;
1427         int clear = req->table_stats_read.clear;
1428
1429         rsp->status = rte_pipeline_table_stats_read(p->p,
1430                 port_id,
1431                 &rsp->table_stats_read.stats,
1432                 clear);
1433
1434         return rsp;
1435 }
1436
1437 union table_rule_match_low_level {
1438         struct rte_table_acl_rule_add_params acl_add;
1439         struct rte_table_acl_rule_delete_params acl_delete;
1440         struct rte_table_array_key array;
1441         uint8_t hash[TABLE_RULE_MATCH_SIZE_MAX];
1442         struct rte_table_lpm_key lpm_ipv4;
1443         struct rte_table_lpm_ipv6_key lpm_ipv6;
1444 };
1445
1446 static int
1447 match_convert_ipv6_depth(uint32_t depth, uint32_t *depth32)
1448 {
1449         if (depth > 128)
1450                 return -1;
1451
1452         switch (depth / 32) {
1453         case 0:
1454                 depth32[0] = depth;
1455                 depth32[1] = 0;
1456                 depth32[2] = 0;
1457                 depth32[3] = 0;
1458                 return 0;
1459
1460         case 1:
1461                 depth32[0] = 32;
1462                 depth32[1] = depth - 32;
1463                 depth32[2] = 0;
1464                 depth32[3] = 0;
1465                 return 0;
1466
1467         case 2:
1468                 depth32[0] = 32;
1469                 depth32[1] = 32;
1470                 depth32[2] = depth - 64;
1471                 depth32[3] = 0;
1472                 return 0;
1473
1474         case 3:
1475                 depth32[0] = 32;
1476                 depth32[1] = 32;
1477                 depth32[2] = 32;
1478                 depth32[3] = depth - 96;
1479                 return 0;
1480
1481         case 4:
1482                 depth32[0] = 32;
1483                 depth32[1] = 32;
1484                 depth32[2] = 32;
1485                 depth32[3] = 32;
1486                 return 0;
1487
1488         default:
1489                 return -1;
1490         }
1491 }
1492
1493 static int
1494 match_convert(struct table_rule_match *mh,
1495         union table_rule_match_low_level *ml,
1496         int add)
1497 {
1498         memset(ml, 0, sizeof(*ml));
1499
1500         switch (mh->match_type) {
1501         case TABLE_ACL:
1502                 if (mh->match.acl.ip_version)
1503                         if (add) {
1504                                 ml->acl_add.field_value[0].value.u8 =
1505                                         mh->match.acl.proto;
1506                                 ml->acl_add.field_value[0].mask_range.u8 =
1507                                         mh->match.acl.proto_mask;
1508
1509                                 ml->acl_add.field_value[1].value.u32 =
1510                                         mh->match.acl.ipv4.sa;
1511                                 ml->acl_add.field_value[1].mask_range.u32 =
1512                                         mh->match.acl.sa_depth;
1513
1514                                 ml->acl_add.field_value[2].value.u32 =
1515                                         mh->match.acl.ipv4.da;
1516                                 ml->acl_add.field_value[2].mask_range.u32 =
1517                                         mh->match.acl.da_depth;
1518
1519                                 ml->acl_add.field_value[3].value.u16 =
1520                                         mh->match.acl.sp0;
1521                                 ml->acl_add.field_value[3].mask_range.u16 =
1522                                         mh->match.acl.sp1;
1523
1524                                 ml->acl_add.field_value[4].value.u16 =
1525                                         mh->match.acl.dp0;
1526                                 ml->acl_add.field_value[4].mask_range.u16 =
1527                                         mh->match.acl.dp1;
1528
1529                                 ml->acl_add.priority =
1530                                         (int32_t) mh->match.acl.priority;
1531                         } else {
1532                                 ml->acl_delete.field_value[0].value.u8 =
1533                                         mh->match.acl.proto;
1534                                 ml->acl_delete.field_value[0].mask_range.u8 =
1535                                         mh->match.acl.proto_mask;
1536
1537                                 ml->acl_delete.field_value[1].value.u32 =
1538                                         mh->match.acl.ipv4.sa;
1539                                 ml->acl_delete.field_value[1].mask_range.u32 =
1540                                         mh->match.acl.sa_depth;
1541
1542                                 ml->acl_delete.field_value[2].value.u32 =
1543                                         mh->match.acl.ipv4.da;
1544                                 ml->acl_delete.field_value[2].mask_range.u32 =
1545                                         mh->match.acl.da_depth;
1546
1547                                 ml->acl_delete.field_value[3].value.u16 =
1548                                         mh->match.acl.sp0;
1549                                 ml->acl_delete.field_value[3].mask_range.u16 =
1550                                         mh->match.acl.sp1;
1551
1552                                 ml->acl_delete.field_value[4].value.u16 =
1553                                         mh->match.acl.dp0;
1554                                 ml->acl_delete.field_value[4].mask_range.u16 =
1555                                         mh->match.acl.dp1;
1556                         }
1557                 else
1558                         if (add) {
1559                                 uint32_t *sa32 =
1560                                         (uint32_t *) mh->match.acl.ipv6.sa;
1561                                 uint32_t *da32 =
1562                                         (uint32_t *) mh->match.acl.ipv6.da;
1563                                 uint32_t sa32_depth[4], da32_depth[4];
1564                                 int status;
1565
1566                                 status = match_convert_ipv6_depth(
1567                                         mh->match.acl.sa_depth,
1568                                         sa32_depth);
1569                                 if (status)
1570                                         return status;
1571
1572                                 status = match_convert_ipv6_depth(
1573                                         mh->match.acl.da_depth,
1574                                         da32_depth);
1575                                 if (status)
1576                                         return status;
1577
1578                                 ml->acl_add.field_value[0].value.u8 =
1579                                         mh->match.acl.proto;
1580                                 ml->acl_add.field_value[0].mask_range.u8 =
1581                                         mh->match.acl.proto_mask;
1582
1583                                 ml->acl_add.field_value[1].value.u32 = sa32[0];
1584                                 ml->acl_add.field_value[1].mask_range.u32 =
1585                                         sa32_depth[0];
1586                                 ml->acl_add.field_value[2].value.u32 = sa32[1];
1587                                 ml->acl_add.field_value[2].mask_range.u32 =
1588                                         sa32_depth[1];
1589                                 ml->acl_add.field_value[3].value.u32 = sa32[2];
1590                                 ml->acl_add.field_value[3].mask_range.u32 =
1591                                         sa32_depth[2];
1592                                 ml->acl_add.field_value[4].value.u32 = sa32[3];
1593                                 ml->acl_add.field_value[4].mask_range.u32 =
1594                                         sa32_depth[3];
1595
1596                                 ml->acl_add.field_value[5].value.u32 = da32[0];
1597                                 ml->acl_add.field_value[5].mask_range.u32 =
1598                                         da32_depth[0];
1599                                 ml->acl_add.field_value[6].value.u32 = da32[1];
1600                                 ml->acl_add.field_value[6].mask_range.u32 =
1601                                         da32_depth[1];
1602                                 ml->acl_add.field_value[7].value.u32 = da32[2];
1603                                 ml->acl_add.field_value[7].mask_range.u32 =
1604                                         da32_depth[2];
1605                                 ml->acl_add.field_value[8].value.u32 = da32[3];
1606                                 ml->acl_add.field_value[8].mask_range.u32 =
1607                                         da32_depth[3];
1608
1609                                 ml->acl_add.field_value[9].value.u16 =
1610                                         mh->match.acl.sp0;
1611                                 ml->acl_add.field_value[9].mask_range.u16 =
1612                                         mh->match.acl.sp1;
1613
1614                                 ml->acl_add.field_value[10].value.u16 =
1615                                         mh->match.acl.dp0;
1616                                 ml->acl_add.field_value[10].mask_range.u16 =
1617                                         mh->match.acl.dp1;
1618
1619                                 ml->acl_add.priority =
1620                                         (int32_t) mh->match.acl.priority;
1621                         } else {
1622                                 uint32_t *sa32 =
1623                                         (uint32_t *) mh->match.acl.ipv6.sa;
1624                                 uint32_t *da32 =
1625                                         (uint32_t *) mh->match.acl.ipv6.da;
1626                                 uint32_t sa32_depth[4], da32_depth[4];
1627                                 int status;
1628
1629                                 status = match_convert_ipv6_depth(
1630                                         mh->match.acl.sa_depth,
1631                                         sa32_depth);
1632                                 if (status)
1633                                         return status;
1634
1635                                 status = match_convert_ipv6_depth(
1636                                         mh->match.acl.da_depth,
1637                                         da32_depth);
1638                                 if (status)
1639                                         return status;
1640
1641                                 ml->acl_delete.field_value[0].value.u8 =
1642                                         mh->match.acl.proto;
1643                                 ml->acl_delete.field_value[0].mask_range.u8 =
1644                                         mh->match.acl.proto_mask;
1645
1646                                 ml->acl_delete.field_value[1].value.u32 =
1647                                         sa32[0];
1648                                 ml->acl_delete.field_value[1].mask_range.u32 =
1649                                         sa32_depth[0];
1650                                 ml->acl_delete.field_value[2].value.u32 =
1651                                         sa32[1];
1652                                 ml->acl_delete.field_value[2].mask_range.u32 =
1653                                         sa32_depth[1];
1654                                 ml->acl_delete.field_value[3].value.u32 =
1655                                         sa32[2];
1656                                 ml->acl_delete.field_value[3].mask_range.u32 =
1657                                         sa32_depth[2];
1658                                 ml->acl_delete.field_value[4].value.u32 =
1659                                         sa32[3];
1660                                 ml->acl_delete.field_value[4].mask_range.u32 =
1661                                         sa32_depth[3];
1662
1663                                 ml->acl_delete.field_value[5].value.u32 =
1664                                         da32[0];
1665                                 ml->acl_delete.field_value[5].mask_range.u32 =
1666                                         da32_depth[0];
1667                                 ml->acl_delete.field_value[6].value.u32 =
1668                                         da32[1];
1669                                 ml->acl_delete.field_value[6].mask_range.u32 =
1670                                         da32_depth[1];
1671                                 ml->acl_delete.field_value[7].value.u32 =
1672                                         da32[2];
1673                                 ml->acl_delete.field_value[7].mask_range.u32 =
1674                                         da32_depth[2];
1675                                 ml->acl_delete.field_value[8].value.u32 =
1676                                         da32[3];
1677                                 ml->acl_delete.field_value[8].mask_range.u32 =
1678                                         da32_depth[3];
1679
1680                                 ml->acl_delete.field_value[9].value.u16 =
1681                                         mh->match.acl.sp0;
1682                                 ml->acl_delete.field_value[9].mask_range.u16 =
1683                                         mh->match.acl.sp1;
1684
1685                                 ml->acl_delete.field_value[10].value.u16 =
1686                                         mh->match.acl.dp0;
1687                                 ml->acl_delete.field_value[10].mask_range.u16 =
1688                                         mh->match.acl.dp1;
1689                         }
1690                 return 0;
1691
1692         case TABLE_ARRAY:
1693                 ml->array.pos = mh->match.array.pos;
1694                 return 0;
1695
1696         case TABLE_HASH:
1697                 memcpy(ml->hash, mh->match.hash.key, sizeof(ml->hash));
1698                 return 0;
1699
1700         case TABLE_LPM:
1701                 if (mh->match.lpm.ip_version) {
1702                         ml->lpm_ipv4.ip = mh->match.lpm.ipv4;
1703                         ml->lpm_ipv4.depth = mh->match.lpm.depth;
1704                 } else {
1705                         memcpy(ml->lpm_ipv6.ip,
1706                                 mh->match.lpm.ipv6, sizeof(ml->lpm_ipv6.ip));
1707                         ml->lpm_ipv6.depth = mh->match.lpm.depth;
1708                 }
1709
1710                 return 0;
1711
1712         default:
1713                 return -1;
1714         }
1715 }
1716
1717 static struct pipeline_msg_rsp *
1718 pipeline_msg_handle_table_rule_add(struct pipeline_data *p,
1719         struct pipeline_msg_req *req)
1720 {
1721         union table_rule_match_low_level match_ll;
1722         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
1723         struct table_rule_match *match = &req->table_rule_add.match;
1724         struct table_rule_action *action = &req->table_rule_add.action;
1725         struct rte_pipeline_table_entry *data_in, *data_out;
1726         uint32_t table_id = req->id;
1727         int key_found, status;
1728         struct rte_table_action *a = p->table_data[table_id].a;
1729
1730         /* Apply actions */
1731         memset(p->buffer, 0, sizeof(p->buffer));
1732         data_in = (struct rte_pipeline_table_entry *) p->buffer;
1733
1734         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_FWD)) {
1735                 status = rte_table_action_apply(a,
1736                         data_in,
1737                         RTE_TABLE_ACTION_FWD,
1738                         &action->fwd);
1739
1740                 if (status) {
1741                         rsp->status = -1;
1742                         return rsp;
1743                 }
1744         }
1745
1746         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_MTR)) {
1747                 status = rte_table_action_apply(a,
1748                         data_in,
1749                         RTE_TABLE_ACTION_MTR,
1750                         &action->mtr);
1751
1752                 if (status) {
1753                         rsp->status = -1;
1754                         return rsp;
1755                 }
1756         }
1757
1758         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_TM)) {
1759                 status = rte_table_action_apply(a,
1760                         data_in,
1761                         RTE_TABLE_ACTION_TM,
1762                         &action->tm);
1763
1764                 if (status) {
1765                         rsp->status = -1;
1766                         return rsp;
1767                 }
1768         }
1769
1770         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_ENCAP)) {
1771                 status = rte_table_action_apply(a,
1772                         data_in,
1773                         RTE_TABLE_ACTION_ENCAP,
1774                         &action->encap);
1775
1776                 if (status) {
1777                         rsp->status = -1;
1778                         return rsp;
1779                 }
1780         }
1781
1782         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_NAT)) {
1783                 status = rte_table_action_apply(a,
1784                         data_in,
1785                         RTE_TABLE_ACTION_NAT,
1786                         &action->nat);
1787
1788                 if (status) {
1789                         rsp->status = -1;
1790                         return rsp;
1791                 }
1792         }
1793
1794         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_TTL)) {
1795                 status = rte_table_action_apply(a,
1796                         data_in,
1797                         RTE_TABLE_ACTION_TTL,
1798                         &action->ttl);
1799
1800                 if (status) {
1801                         rsp->status = -1;
1802                         return rsp;
1803                 }
1804         }
1805
1806         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_STATS)) {
1807                 status = rte_table_action_apply(a,
1808                         data_in,
1809                         RTE_TABLE_ACTION_STATS,
1810                         &action->stats);
1811
1812                 if (status) {
1813                         rsp->status = -1;
1814                         return rsp;
1815                 }
1816         }
1817
1818         if (action->action_mask & (1LLU << RTE_TABLE_ACTION_TIME)) {
1819                 status = rte_table_action_apply(a,
1820                         data_in,
1821                         RTE_TABLE_ACTION_TIME,
1822                         &action->time);
1823
1824                 if (status) {
1825                         rsp->status = -1;
1826                         return rsp;
1827                 }
1828         }
1829
1830         /* Add rule (match, action) to table */
1831         status = match_convert(match, &match_ll, 1);
1832         if (status) {
1833                 rsp->status = -1;
1834                 return rsp;
1835         }
1836
1837         status = rte_pipeline_table_entry_add(p->p,
1838                 table_id,
1839                 &match_ll,
1840                 data_in,
1841                 &key_found,
1842                 &data_out);
1843         if (status) {
1844                 rsp->status = -1;
1845                 return rsp;
1846         }
1847
1848         /* Write response */
1849         rsp->status = 0;
1850         rsp->table_rule_add.data = data_out;
1851
1852         return rsp;
1853 }
1854
1855 static struct pipeline_msg_rsp *
1856 pipeline_msg_handle_table_rule_add_default(struct pipeline_data *p,
1857         struct pipeline_msg_req *req)
1858 {
1859         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
1860         struct table_rule_action *action = &req->table_rule_add_default.action;
1861         struct rte_pipeline_table_entry *data_in, *data_out;
1862         uint32_t table_id = req->id;
1863         int status;
1864
1865         /* Apply actions */
1866         memset(p->buffer, 0, sizeof(p->buffer));
1867         data_in = (struct rte_pipeline_table_entry *) p->buffer;
1868
1869         data_in->action = action->fwd.action;
1870         if (action->fwd.action == RTE_PIPELINE_ACTION_PORT)
1871                 data_in->port_id = action->fwd.id;
1872         if (action->fwd.action == RTE_PIPELINE_ACTION_TABLE)
1873                 data_in->table_id = action->fwd.id;
1874
1875         /* Add default rule to table */
1876         status = rte_pipeline_table_default_entry_add(p->p,
1877                 table_id,
1878                 data_in,
1879                 &data_out);
1880         if (status) {
1881                 rsp->status = -1;
1882                 return rsp;
1883         }
1884
1885         /* Write response */
1886         rsp->status = 0;
1887         rsp->table_rule_add_default.data = data_out;
1888
1889         return rsp;
1890 }
1891
1892 static struct pipeline_msg_rsp *
1893 pipeline_msg_handle_table_rule_add_bulk(struct pipeline_data *p,
1894         struct pipeline_msg_req *req)
1895 {
1896
1897         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
1898
1899         uint32_t table_id = req->id;
1900         struct table_rule_match *match = req->table_rule_add_bulk.match;
1901         struct table_rule_action *action = req->table_rule_add_bulk.action;
1902         struct rte_pipeline_table_entry **data =
1903                 (struct rte_pipeline_table_entry **)req->table_rule_add_bulk.data;
1904         uint32_t n_rules = req->table_rule_add_bulk.n_rules;
1905         uint32_t bulk = req->table_rule_add_bulk.bulk;
1906
1907         struct rte_table_action *a = p->table_data[table_id].a;
1908         union table_rule_match_low_level *match_ll;
1909         uint8_t *action_ll;
1910         void **match_ll_ptr;
1911         struct rte_pipeline_table_entry **action_ll_ptr;
1912         int *found, status;
1913         uint32_t i;
1914
1915         /* Memory allocation */
1916         match_ll = calloc(n_rules, sizeof(union table_rule_match_low_level));
1917         action_ll = calloc(n_rules, TABLE_RULE_ACTION_SIZE_MAX);
1918         match_ll_ptr = calloc(n_rules, sizeof(void *));
1919         action_ll_ptr =
1920                 calloc(n_rules, sizeof(struct rte_pipeline_table_entry *));
1921         found = calloc(n_rules, sizeof(int));
1922
1923         if ((match_ll == NULL) ||
1924                 (action_ll == NULL) ||
1925                 (match_ll_ptr == NULL) ||
1926                 (action_ll_ptr == NULL) ||
1927                 (found == NULL))
1928                 goto fail;
1929
1930         for (i = 0; i < n_rules; i++) {
1931                 match_ll_ptr[i] = (void *)&match_ll[i];
1932                 action_ll_ptr[i] =
1933                         (struct rte_pipeline_table_entry *)&action_ll[i * TABLE_RULE_ACTION_SIZE_MAX];
1934         }
1935
1936         /* Rule match conversion */
1937         for (i = 0; i < n_rules; i++) {
1938                 status = match_convert(&match[i], match_ll_ptr[i], 1);
1939                 if (status)
1940                         goto fail;
1941         }
1942
1943         /* Rule action conversion */
1944         for (i = 0; i < n_rules; i++) {
1945                 void *data_in = action_ll_ptr[i];
1946                 struct table_rule_action *act = &action[i];
1947
1948                 if (act->action_mask & (1LLU << RTE_TABLE_ACTION_FWD)) {
1949                         status = rte_table_action_apply(a,
1950                                 data_in,
1951                                 RTE_TABLE_ACTION_FWD,
1952                                 &act->fwd);
1953
1954                         if (status)
1955                                 goto fail;
1956                 }
1957
1958                 if (act->action_mask & (1LLU << RTE_TABLE_ACTION_MTR)) {
1959                         status = rte_table_action_apply(a,
1960                                 data_in,
1961                                 RTE_TABLE_ACTION_MTR,
1962                                 &act->mtr);
1963
1964                         if (status)
1965                                 goto fail;
1966                 }
1967
1968                 if (act->action_mask & (1LLU << RTE_TABLE_ACTION_TM)) {
1969                         status = rte_table_action_apply(a,
1970                                 data_in,
1971                                 RTE_TABLE_ACTION_TM,
1972                                 &act->tm);
1973
1974                         if (status)
1975                                 goto fail;
1976                 }
1977
1978                 if (act->action_mask & (1LLU << RTE_TABLE_ACTION_ENCAP)) {
1979                         status = rte_table_action_apply(a,
1980                                 data_in,
1981                                 RTE_TABLE_ACTION_ENCAP,
1982                                 &act->encap);
1983
1984                         if (status)
1985                                 goto fail;
1986                 }
1987
1988                 if (act->action_mask & (1LLU << RTE_TABLE_ACTION_NAT)) {
1989                         status = rte_table_action_apply(a,
1990                                 data_in,
1991                                 RTE_TABLE_ACTION_NAT,
1992                                 &act->nat);
1993
1994                         if (status)
1995                                 goto fail;
1996                 }
1997
1998                 if (act->action_mask & (1LLU << RTE_TABLE_ACTION_TTL)) {
1999                         status = rte_table_action_apply(a,
2000                                 data_in,
2001                                 RTE_TABLE_ACTION_TTL,
2002                                 &act->ttl);
2003
2004                         if (status)
2005                                 goto fail;
2006                 }
2007
2008                 if (act->action_mask & (1LLU << RTE_TABLE_ACTION_STATS)) {
2009                         status = rte_table_action_apply(a,
2010                                 data_in,
2011                                 RTE_TABLE_ACTION_STATS,
2012                                 &act->stats);
2013
2014                         if (status)
2015                                 goto fail;
2016                 }
2017
2018                 if (act->action_mask & (1LLU << RTE_TABLE_ACTION_TIME)) {
2019                         status = rte_table_action_apply(a,
2020                                 data_in,
2021                                 RTE_TABLE_ACTION_TIME,
2022                                 &act->time);
2023
2024                         if (status)
2025                                 goto fail;
2026                 }
2027         }
2028
2029         /* Add rule (match, action) to table */
2030         if (bulk) {
2031                 status = rte_pipeline_table_entry_add_bulk(p->p,
2032                         table_id,
2033                         match_ll_ptr,
2034                         action_ll_ptr,
2035                         n_rules,
2036                         found,
2037                         data);
2038                 if (status)
2039                         n_rules = 0;
2040         } else
2041                 for (i = 0; i < n_rules; i++) {
2042                         status = rte_pipeline_table_entry_add(p->p,
2043                                 table_id,
2044                                 match_ll_ptr[i],
2045                                 action_ll_ptr[i],
2046                                 &found[i],
2047                                 &data[i]);
2048                         if (status) {
2049                                 n_rules = i;
2050                                 break;
2051                         }
2052                 }
2053
2054         /* Write response */
2055         rsp->status = 0;
2056         rsp->table_rule_add_bulk.n_rules = n_rules;
2057
2058         /* Free */
2059         free(found);
2060         free(action_ll_ptr);
2061         free(match_ll_ptr);
2062         free(action_ll);
2063         free(match_ll);
2064
2065         return rsp;
2066
2067 fail:
2068         free(found);
2069         free(action_ll_ptr);
2070         free(match_ll_ptr);
2071         free(action_ll);
2072         free(match_ll);
2073
2074         rsp->status = -1;
2075         rsp->table_rule_add_bulk.n_rules = 0;
2076         return rsp;
2077 }
2078
2079 static struct pipeline_msg_rsp *
2080 pipeline_msg_handle_table_rule_delete(struct pipeline_data *p,
2081         struct pipeline_msg_req *req)
2082 {
2083         union table_rule_match_low_level match_ll;
2084         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
2085         struct table_rule_match *match = &req->table_rule_delete.match;
2086         uint32_t table_id = req->id;
2087         int key_found, status;
2088
2089         status = match_convert(match, &match_ll, 0);
2090         if (status) {
2091                 rsp->status = -1;
2092                 return rsp;
2093         }
2094
2095         rsp->status = rte_pipeline_table_entry_delete(p->p,
2096                 table_id,
2097                 &match_ll,
2098                 &key_found,
2099                 NULL);
2100
2101         return rsp;
2102 }
2103
2104 static struct pipeline_msg_rsp *
2105 pipeline_msg_handle_table_rule_delete_default(struct pipeline_data *p,
2106         struct pipeline_msg_req *req)
2107 {
2108         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
2109         uint32_t table_id = req->id;
2110
2111         rsp->status = rte_pipeline_table_default_entry_delete(p->p,
2112                 table_id,
2113                 NULL);
2114
2115         return rsp;
2116 }
2117
2118 static struct pipeline_msg_rsp *
2119 pipeline_msg_handle_table_rule_stats_read(struct pipeline_data *p,
2120         struct pipeline_msg_req *req)
2121 {
2122         struct pipeline_msg_rsp *rsp = (struct pipeline_msg_rsp *) req;
2123         uint32_t table_id = req->id;
2124         void *data = req->table_rule_stats_read.data;
2125         int clear = req->table_rule_stats_read.clear;
2126         struct rte_table_action *a = p->table_data[table_id].a;
2127
2128         rsp->status = rte_table_action_stats_read(a,
2129                 data,
2130                 &rsp->table_rule_stats_read.stats,
2131                 clear);
2132
2133         return rsp;
2134 }
2135
2136 static void
2137 pipeline_msg_handle(struct pipeline_data *p)
2138 {
2139         for ( ; ; ) {
2140                 struct pipeline_msg_req *req;
2141                 struct pipeline_msg_rsp *rsp;
2142
2143                 req = pipeline_msg_recv(p->msgq_req);
2144                 if (req == NULL)
2145                         break;
2146
2147                 switch (req->type) {
2148                 case PIPELINE_REQ_PORT_IN_STATS_READ:
2149                         rsp = pipeline_msg_handle_port_in_stats_read(p, req);
2150                         break;
2151
2152                 case PIPELINE_REQ_PORT_IN_ENABLE:
2153                         rsp = pipeline_msg_handle_port_in_enable(p, req);
2154                         break;
2155
2156                 case PIPELINE_REQ_PORT_IN_DISABLE:
2157                         rsp = pipeline_msg_handle_port_in_disable(p, req);
2158                         break;
2159
2160                 case PIPELINE_REQ_PORT_OUT_STATS_READ:
2161                         rsp = pipeline_msg_handle_port_out_stats_read(p, req);
2162                         break;
2163
2164                 case PIPELINE_REQ_TABLE_STATS_READ:
2165                         rsp = pipeline_msg_handle_table_stats_read(p, req);
2166                         break;
2167
2168                 case PIPELINE_REQ_TABLE_RULE_ADD:
2169                         rsp = pipeline_msg_handle_table_rule_add(p, req);
2170                         break;
2171
2172                 case PIPELINE_REQ_TABLE_RULE_ADD_DEFAULT:
2173                         rsp = pipeline_msg_handle_table_rule_add_default(p,     req);
2174                         break;
2175
2176                 case PIPELINE_REQ_TABLE_RULE_ADD_BULK:
2177                         rsp = pipeline_msg_handle_table_rule_add_bulk(p, req);
2178                         break;
2179
2180                 case PIPELINE_REQ_TABLE_RULE_DELETE:
2181                         rsp = pipeline_msg_handle_table_rule_delete(p, req);
2182                         break;
2183
2184                 case PIPELINE_REQ_TABLE_RULE_DELETE_DEFAULT:
2185                         rsp = pipeline_msg_handle_table_rule_delete_default(p, req);
2186                         break;
2187
2188                 case PIPELINE_REQ_TABLE_RULE_STATS_READ:
2189                         rsp = pipeline_msg_handle_table_rule_stats_read(p, req);
2190                         break;
2191
2192                 default:
2193                         rsp = (struct pipeline_msg_rsp *) req;
2194                         rsp->status = -1;
2195                 }
2196
2197                 pipeline_msg_send(p->msgq_rsp, rsp);
2198         }
2199 }
2200
2201 /**
2202  * Data plane threads: main
2203  */
2204 int
2205 thread_main(void *arg __rte_unused)
2206 {
2207         struct thread_data *t;
2208         uint32_t thread_id, i;
2209
2210         thread_id = rte_lcore_id();
2211         t = &thread_data[thread_id];
2212
2213         /* Dispatch loop */
2214         for (i = 0; ; i++) {
2215                 uint32_t j;
2216
2217                 /* Data Plane */
2218                 for (j = 0; j < t->n_pipelines; j++)
2219                         rte_pipeline_run(t->p[j]);
2220
2221                 /* Control Plane */
2222                 if ((i & 0xF) == 0) {
2223                         uint64_t time = rte_get_tsc_cycles();
2224                         uint64_t time_next_min = UINT64_MAX;
2225
2226                         if (time < t->time_next_min)
2227                                 continue;
2228
2229                         /* Pipeline message queues */
2230                         for (j = 0; j < t->n_pipelines; j++) {
2231                                 struct pipeline_data *p =
2232                                         &t->pipeline_data[j];
2233                                 uint64_t time_next = p->time_next;
2234
2235                                 if (time_next <= time) {
2236                                         pipeline_msg_handle(p);
2237                                         rte_pipeline_flush(p->p);
2238                                         time_next = time + p->timer_period;
2239                                         p->time_next = time_next;
2240                                 }
2241
2242                                 if (time_next < time_next_min)
2243                                         time_next_min = time_next;
2244                         }
2245
2246                         /* Thread message queues */
2247                         {
2248                                 uint64_t time_next = t->time_next;
2249
2250                                 if (time_next <= time) {
2251                                         thread_msg_handle(t);
2252                                         time_next = time + t->timer_period;
2253                                         t->time_next = time_next;
2254                                 }
2255
2256                                 if (time_next < time_next_min)
2257                                         time_next_min = time_next;
2258                         }
2259
2260                         t->time_next_min = time_next_min;
2261                 }
2262         }
2263
2264         return 0;
2265 }