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