1/* 2 * Copyright (C) 2017 Corelight, Inc. and Universita` di Pisa. All rights reserved. 3 * 4 * Redistribution and use in source and binary forms, with or without 5 * modification, are permitted provided that the following conditions 6 * are met: 7 * 1. Redistributions of source code must retain the above copyright 8 * notice, this list of conditions and the following disclaimer. 9 * 2. Redistributions in binary form must reproduce the above copyright 10 * notice, this list of conditions and the following disclaimer in the 11 * documentation and/or other materials provided with the distribution. 12 * 13 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND 14 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 15 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 16 * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE 17 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 18 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS 19 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) 20 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 21 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY 22 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF 23 * SUCH DAMAGE. 24 */ 25/* $FreeBSD: stable/11/tools/tools/netmap/lb.c 341434 2018-12-03 17:51:22Z vmaffione $ */ 26#include <stdio.h> 27#include <string.h> 28#include <ctype.h> 29#include <stdbool.h> 30#include <inttypes.h> 31#include <syslog.h> 32 33#define NETMAP_WITH_LIBS 34#include <net/netmap_user.h> 35#include <sys/poll.h> 36 37#include <netinet/in.h> /* htonl */ 38 39#include <pthread.h> 40 41#include "pkt_hash.h" 42#include "ctrs.h" 43 44 45/* 46 * use our version of header structs, rather than bringing in a ton 47 * of platform specific ones 48 */ 49#ifndef ETH_ALEN 50#define ETH_ALEN 6 51#endif 52 53struct compact_eth_hdr { 54 unsigned char h_dest[ETH_ALEN]; 55 unsigned char h_source[ETH_ALEN]; 56 u_int16_t h_proto; 57}; 58 59struct compact_ip_hdr { 60 u_int8_t ihl:4, version:4; 61 u_int8_t tos; 62 u_int16_t tot_len; 63 u_int16_t id; 64 u_int16_t frag_off; 65 u_int8_t ttl; 66 u_int8_t protocol; 67 u_int16_t check; 68 u_int32_t saddr; 69 u_int32_t daddr; 70}; 71 72struct compact_ipv6_hdr { 73 u_int8_t priority:4, version:4; 74 u_int8_t flow_lbl[3]; 75 u_int16_t payload_len; 76 u_int8_t nexthdr; 77 u_int8_t hop_limit; 78 struct in6_addr saddr; 79 struct in6_addr daddr; 80}; 81 82#define MAX_IFNAMELEN 64 83#define MAX_PORTNAMELEN (MAX_IFNAMELEN + 40) 84#define DEF_OUT_PIPES 2 85#define DEF_EXTRA_BUFS 0 86#define DEF_BATCH 2048 87#define DEF_WAIT_LINK 2 88#define DEF_STATS_INT 600 89#define BUF_REVOKE 100 90#define STAT_MSG_MAXSIZE 1024 91 92struct { 93 char ifname[MAX_IFNAMELEN]; 94 char base_name[MAX_IFNAMELEN]; 95 int netmap_fd; 96 uint16_t output_rings; 97 uint16_t num_groups; 98 uint32_t extra_bufs; 99 uint16_t batch; 100 int stdout_interval; 101 int syslog_interval; 102 int wait_link; 103 bool busy_wait; 104} glob_arg; 105 106/* 107 * the overflow queue is a circular queue of buffers 108 */ 109struct overflow_queue { 110 char name[MAX_IFNAMELEN + 16]; 111 struct netmap_slot *slots; 112 uint32_t head; 113 uint32_t tail; 114 uint32_t n; 115 uint32_t size; 116}; 117 118struct overflow_queue *freeq; 119 120static inline int 121oq_full(struct overflow_queue *q) 122{ 123 return q->n >= q->size; 124} 125 126static inline int 127oq_empty(struct overflow_queue *q) 128{ 129 return q->n <= 0; 130} 131 132static inline void 133oq_enq(struct overflow_queue *q, const struct netmap_slot *s) 134{ 135 if (unlikely(oq_full(q))) { 136 D("%s: queue full!", q->name); 137 abort(); 138 } 139 q->slots[q->tail] = *s; 140 q->n++; 141 q->tail++; 142 if (q->tail >= q->size) 143 q->tail = 0; 144} 145 146static inline struct netmap_slot 147oq_deq(struct overflow_queue *q) 148{ 149 struct netmap_slot s = q->slots[q->head]; 150 if (unlikely(oq_empty(q))) { 151 D("%s: queue empty!", q->name); 152 abort(); 153 } 154 q->n--; 155 q->head++; 156 if (q->head >= q->size) 157 q->head = 0; 158 return s; 159} 160 161static volatile int do_abort = 0; 162 163uint64_t dropped = 0; 164uint64_t forwarded = 0; 165uint64_t received_bytes = 0; 166uint64_t received_pkts = 0; 167uint64_t non_ip = 0; 168uint32_t freeq_n = 0; 169 170struct port_des { 171 char interface[MAX_PORTNAMELEN]; 172 struct my_ctrs ctr; 173 unsigned int last_sync; 174 uint32_t last_tail; 175 struct overflow_queue *oq; 176 struct nm_desc *nmd; 177 struct netmap_ring *ring; 178 struct group_des *group; 179}; 180 181struct port_des *ports; 182 183/* each group of pipes receives all the packets */ 184struct group_des { 185 char pipename[MAX_IFNAMELEN]; 186 struct port_des *ports; 187 int first_id; 188 int nports; 189 int last; 190 int custom_port; 191}; 192 193struct group_des *groups; 194 195/* statistcs */ 196struct counters { 197 struct timeval ts; 198 struct my_ctrs *ctrs; 199 uint64_t received_pkts; 200 uint64_t received_bytes; 201 uint64_t non_ip; 202 uint32_t freeq_n; 203 int status __attribute__((aligned(64))); 204#define COUNTERS_EMPTY 0 205#define COUNTERS_FULL 1 206}; 207 208struct counters counters_buf; 209 210static void * 211print_stats(void *arg) 212{ 213 int npipes = glob_arg.output_rings; 214 int sys_int = 0; 215 (void)arg; 216 struct my_ctrs cur, prev; 217 struct my_ctrs *pipe_prev; 218 219 pipe_prev = calloc(npipes, sizeof(struct my_ctrs)); 220 if (pipe_prev == NULL) { 221 D("out of memory"); 222 exit(1); 223 } 224 225 char stat_msg[STAT_MSG_MAXSIZE] = ""; 226 227 memset(&prev, 0, sizeof(prev)); 228 while (!do_abort) { 229 int j, dosyslog = 0, dostdout = 0, newdata; 230 uint64_t pps = 0, dps = 0, bps = 0, dbps = 0, usec = 0; 231 struct my_ctrs x; 232 233 counters_buf.status = COUNTERS_EMPTY; 234 newdata = 0; 235 memset(&cur, 0, sizeof(cur)); 236 sleep(1); 237 if (counters_buf.status == COUNTERS_FULL) { 238 __sync_synchronize(); 239 newdata = 1; 240 cur.t = counters_buf.ts; 241 if (prev.t.tv_sec || prev.t.tv_usec) { 242 usec = (cur.t.tv_sec - prev.t.tv_sec) * 1000000 + 243 cur.t.tv_usec - prev.t.tv_usec; 244 } 245 } 246 247 ++sys_int; 248 if (glob_arg.stdout_interval && sys_int % glob_arg.stdout_interval == 0) 249 dostdout = 1; 250 if (glob_arg.syslog_interval && sys_int % glob_arg.syslog_interval == 0) 251 dosyslog = 1; 252 253 for (j = 0; j < npipes; ++j) { 254 struct my_ctrs *c = &counters_buf.ctrs[j]; 255 cur.pkts += c->pkts; 256 cur.drop += c->drop; 257 cur.drop_bytes += c->drop_bytes; 258 cur.bytes += c->bytes; 259 260 if (usec) { 261 x.pkts = c->pkts - pipe_prev[j].pkts; 262 x.drop = c->drop - pipe_prev[j].drop; 263 x.bytes = c->bytes - pipe_prev[j].bytes; 264 x.drop_bytes = c->drop_bytes - pipe_prev[j].drop_bytes; 265 pps = (x.pkts*1000000 + usec/2) / usec; 266 dps = (x.drop*1000000 + usec/2) / usec; 267 bps = ((x.bytes*1000000 + usec/2) / usec) * 8; 268 dbps = ((x.drop_bytes*1000000 + usec/2) / usec) * 8; 269 } 270 pipe_prev[j] = *c; 271 272 if ( (dosyslog || dostdout) && newdata ) 273 snprintf(stat_msg, STAT_MSG_MAXSIZE, 274 "{" 275 "\"ts\":%.6f," 276 "\"interface\":\"%s\"," 277 "\"output_ring\":%" PRIu16 "," 278 "\"packets_forwarded\":%" PRIu64 "," 279 "\"packets_dropped\":%" PRIu64 "," 280 "\"data_forward_rate_Mbps\":%.4f," 281 "\"data_drop_rate_Mbps\":%.4f," 282 "\"packet_forward_rate_kpps\":%.4f," 283 "\"packet_drop_rate_kpps\":%.4f," 284 "\"overflow_queue_size\":%" PRIu32 285 "}", cur.t.tv_sec + (cur.t.tv_usec / 1000000.0), 286 ports[j].interface, 287 j, 288 c->pkts, 289 c->drop, 290 (double)bps / 1024 / 1024, 291 (double)dbps / 1024 / 1024, 292 (double)pps / 1000, 293 (double)dps / 1000, 294 c->oq_n); 295 296 if (dosyslog && stat_msg[0]) 297 syslog(LOG_INFO, "%s", stat_msg); 298 if (dostdout && stat_msg[0]) 299 printf("%s\n", stat_msg); 300 } 301 if (usec) { 302 x.pkts = cur.pkts - prev.pkts; 303 x.drop = cur.drop - prev.drop; 304 x.bytes = cur.bytes - prev.bytes; 305 x.drop_bytes = cur.drop_bytes - prev.drop_bytes; 306 pps = (x.pkts*1000000 + usec/2) / usec; 307 dps = (x.drop*1000000 + usec/2) / usec; 308 bps = ((x.bytes*1000000 + usec/2) / usec) * 8; 309 dbps = ((x.drop_bytes*1000000 + usec/2) / usec) * 8; 310 } 311 312 if ( (dosyslog || dostdout) && newdata ) 313 snprintf(stat_msg, STAT_MSG_MAXSIZE, 314 "{" 315 "\"ts\":%.6f," 316 "\"interface\":\"%s\"," 317 "\"output_ring\":null," 318 "\"packets_received\":%" PRIu64 "," 319 "\"packets_forwarded\":%" PRIu64 "," 320 "\"packets_dropped\":%" PRIu64 "," 321 "\"non_ip_packets\":%" PRIu64 "," 322 "\"data_forward_rate_Mbps\":%.4f," 323 "\"data_drop_rate_Mbps\":%.4f," 324 "\"packet_forward_rate_kpps\":%.4f," 325 "\"packet_drop_rate_kpps\":%.4f," 326 "\"free_buffer_slots\":%" PRIu32 327 "}", cur.t.tv_sec + (cur.t.tv_usec / 1000000.0), 328 glob_arg.ifname, 329 received_pkts, 330 cur.pkts, 331 cur.drop, 332 counters_buf.non_ip, 333 (double)bps / 1024 / 1024, 334 (double)dbps / 1024 / 1024, 335 (double)pps / 1000, 336 (double)dps / 1000, 337 counters_buf.freeq_n); 338 339 if (dosyslog && stat_msg[0]) 340 syslog(LOG_INFO, "%s", stat_msg); 341 if (dostdout && stat_msg[0]) 342 printf("%s\n", stat_msg); 343 344 prev = cur; 345 } 346 347 free(pipe_prev); 348 349 return NULL; 350} 351 352static void 353free_buffers(void) 354{ 355 int i, tot = 0; 356 struct port_des *rxport = &ports[glob_arg.output_rings]; 357 358 /* build a netmap free list with the buffers in all the overflow queues */ 359 for (i = 0; i < glob_arg.output_rings + 1; i++) { 360 struct port_des *cp = &ports[i]; 361 struct overflow_queue *q = cp->oq; 362 363 if (!q) 364 continue; 365 366 while (q->n) { 367 struct netmap_slot s = oq_deq(q); 368 uint32_t *b = (uint32_t *)NETMAP_BUF(cp->ring, s.buf_idx); 369 370 *b = rxport->nmd->nifp->ni_bufs_head; 371 rxport->nmd->nifp->ni_bufs_head = s.buf_idx; 372 tot++; 373 } 374 } 375 D("added %d buffers to netmap free list", tot); 376 377 for (i = 0; i < glob_arg.output_rings + 1; ++i) { 378 nm_close(ports[i].nmd); 379 } 380} 381 382 383static void sigint_h(int sig) 384{ 385 (void)sig; /* UNUSED */ 386 do_abort = 1; 387 signal(SIGINT, SIG_DFL); 388} 389 390void usage() 391{ 392 printf("usage: lb [options]\n"); 393 printf("where options are:\n"); 394 printf(" -h view help text\n"); 395 printf(" -i iface interface name (required)\n"); 396 printf(" -p [prefix:]npipes add a new group of output pipes\n"); 397 printf(" -B nbufs number of extra buffers (default: %d)\n", DEF_EXTRA_BUFS); 398 printf(" -b batch batch size (default: %d)\n", DEF_BATCH); 399 printf(" -w seconds wait for link up (default: %d)\n", DEF_WAIT_LINK); 400 printf(" -W enable busy waiting. this will run your CPU at 100%%\n"); 401 printf(" -s seconds seconds between syslog stats messages (default: 0)\n"); 402 printf(" -o seconds seconds between stdout stats messages (default: 0)\n"); 403 exit(0); 404} 405 406static int 407parse_pipes(char *spec) 408{ 409 char *end = index(spec, ':'); 410 static int max_groups = 0; 411 struct group_des *g; 412 413 ND("spec %s num_groups %d", spec, glob_arg.num_groups); 414 if (max_groups < glob_arg.num_groups + 1) { 415 size_t size = sizeof(*g) * (glob_arg.num_groups + 1); 416 groups = realloc(groups, size); 417 if (groups == NULL) { 418 D("out of memory"); 419 return 1; 420 } 421 } 422 g = &groups[glob_arg.num_groups]; 423 memset(g, 0, sizeof(*g)); 424 425 if (end != NULL) { 426 if (end - spec > MAX_IFNAMELEN - 8) { 427 D("name '%s' too long", spec); 428 return 1; 429 } 430 if (end == spec) { 431 D("missing prefix before ':' in '%s'", spec); 432 return 1; 433 } 434 strncpy(g->pipename, spec, end - spec); 435 g->custom_port = 1; 436 end++; 437 } else { 438 /* no prefix, this group will use the 439 * name of the input port. 440 * This will be set in init_groups(), 441 * since here the input port may still 442 * be uninitialized 443 */ 444 end = spec; 445 } 446 if (*end == '\0') { 447 g->nports = DEF_OUT_PIPES; 448 } else { 449 g->nports = atoi(end); 450 if (g->nports < 1) { 451 D("invalid number of pipes '%s' (must be at least 1)", end); 452 return 1; 453 } 454 } 455 glob_arg.output_rings += g->nports; 456 glob_arg.num_groups++; 457 return 0; 458} 459 460/* complete the initialization of the groups data structure */ 461void init_groups(void) 462{ 463 int i, j, t = 0; 464 struct group_des *g = NULL; 465 for (i = 0; i < glob_arg.num_groups; i++) { 466 g = &groups[i]; 467 g->ports = &ports[t]; 468 for (j = 0; j < g->nports; j++) 469 g->ports[j].group = g; 470 t += g->nports; 471 if (!g->custom_port) 472 strcpy(g->pipename, glob_arg.base_name); 473 for (j = 0; j < i; j++) { 474 struct group_des *h = &groups[j]; 475 if (!strcmp(h->pipename, g->pipename)) 476 g->first_id += h->nports; 477 } 478 } 479 g->last = 1; 480} 481 482/* push the packet described by slot rs to the group g. 483 * This may cause other buffers to be pushed down the 484 * chain headed by g. 485 * Return a free buffer. 486 */ 487uint32_t forward_packet(struct group_des *g, struct netmap_slot *rs) 488{ 489 uint32_t hash = rs->ptr; 490 uint32_t output_port = hash % g->nports; 491 struct port_des *port = &g->ports[output_port]; 492 struct netmap_ring *ring = port->ring; 493 struct overflow_queue *q = port->oq; 494 495 /* Move the packet to the output pipe, unless there is 496 * either no space left on the ring, or there is some 497 * packet still in the overflow queue (since those must 498 * take precedence over the new one) 499 */ 500 if (ring->head != ring->tail && (q == NULL || oq_empty(q))) { 501 struct netmap_slot *ts = &ring->slot[ring->head]; 502 struct netmap_slot old_slot = *ts; 503 504 ts->buf_idx = rs->buf_idx; 505 ts->len = rs->len; 506 ts->flags |= NS_BUF_CHANGED; 507 ts->ptr = rs->ptr; 508 ring->head = nm_ring_next(ring, ring->head); 509 port->ctr.bytes += rs->len; 510 port->ctr.pkts++; 511 forwarded++; 512 return old_slot.buf_idx; 513 } 514 515 /* use the overflow queue, if available */ 516 if (q == NULL || oq_full(q)) { 517 /* no space left on the ring and no overflow queue 518 * available: we are forced to drop the packet 519 */ 520 dropped++; 521 port->ctr.drop++; 522 port->ctr.drop_bytes += rs->len; 523 return rs->buf_idx; 524 } 525 526 oq_enq(q, rs); 527 528 /* 529 * we cannot continue down the chain and we need to 530 * return a free buffer now. We take it from the free queue. 531 */ 532 if (oq_empty(freeq)) { 533 /* the free queue is empty. Revoke some buffers 534 * from the longest overflow queue 535 */ 536 uint32_t j; 537 struct port_des *lp = &ports[0]; 538 uint32_t max = lp->oq->n; 539 540 /* let lp point to the port with the longest queue */ 541 for (j = 1; j < glob_arg.output_rings; j++) { 542 struct port_des *cp = &ports[j]; 543 if (cp->oq->n > max) { 544 lp = cp; 545 max = cp->oq->n; 546 } 547 } 548 549 /* move the oldest BUF_REVOKE buffers from the 550 * lp queue to the free queue 551 */ 552 // XXX optimize this cycle 553 for (j = 0; lp->oq->n && j < BUF_REVOKE; j++) { 554 struct netmap_slot tmp = oq_deq(lp->oq); 555 556 dropped++; 557 lp->ctr.drop++; 558 lp->ctr.drop_bytes += tmp.len; 559 560 oq_enq(freeq, &tmp); 561 } 562 563 ND(1, "revoked %d buffers from %s", j, lq->name); 564 } 565 566 return oq_deq(freeq).buf_idx; 567} 568 569int main(int argc, char **argv) 570{ 571 int ch; 572 uint32_t i; 573 int rv; 574 unsigned int iter = 0; 575 int poll_timeout = 10; /* default */ 576 577 glob_arg.ifname[0] = '\0'; 578 glob_arg.output_rings = 0; 579 glob_arg.batch = DEF_BATCH; 580 glob_arg.wait_link = DEF_WAIT_LINK; 581 glob_arg.busy_wait = false; 582 glob_arg.syslog_interval = 0; 583 glob_arg.stdout_interval = 0; 584 585 while ( (ch = getopt(argc, argv, "hi:p:b:B:s:o:w:W")) != -1) { 586 switch (ch) { 587 case 'i': 588 D("interface is %s", optarg); 589 if (strlen(optarg) > MAX_IFNAMELEN - 8) { 590 D("ifname too long %s", optarg); 591 return 1; 592 } 593 if (strncmp(optarg, "netmap:", 7) && strncmp(optarg, "vale", 4)) { 594 sprintf(glob_arg.ifname, "netmap:%s", optarg); 595 } else { 596 strcpy(glob_arg.ifname, optarg); 597 } 598 break; 599 600 case 'p': 601 if (parse_pipes(optarg)) { 602 usage(); 603 return 1; 604 } 605 break; 606 607 case 'B': 608 glob_arg.extra_bufs = atoi(optarg); 609 D("requested %d extra buffers", glob_arg.extra_bufs); 610 break; 611 612 case 'b': 613 glob_arg.batch = atoi(optarg); 614 D("batch is %d", glob_arg.batch); 615 break; 616 617 case 'w': 618 glob_arg.wait_link = atoi(optarg); 619 D("link wait for up time is %d", glob_arg.wait_link); 620 break; 621 622 case 'W': 623 glob_arg.busy_wait = true; 624 break; 625 626 case 'o': 627 glob_arg.stdout_interval = atoi(optarg); 628 break; 629 630 case 's': 631 glob_arg.syslog_interval = atoi(optarg); 632 break; 633 634 case 'h': 635 usage(); 636 return 0; 637 break; 638 639 default: 640 D("bad option %c %s", ch, optarg); 641 usage(); 642 return 1; 643 } 644 } 645 646 if (glob_arg.ifname[0] == '\0') { 647 D("missing interface name"); 648 usage(); 649 return 1; 650 } 651 652 /* extract the base name */ 653 char *nscan = strncmp(glob_arg.ifname, "netmap:", 7) ? 654 glob_arg.ifname : glob_arg.ifname + 7; 655 strncpy(glob_arg.base_name, nscan, MAX_IFNAMELEN-1); 656 for (nscan = glob_arg.base_name; *nscan && !index("-*^{}/@", *nscan); nscan++) 657 ; 658 *nscan = '\0'; 659 660 if (glob_arg.num_groups == 0) 661 parse_pipes(""); 662 663 if (glob_arg.syslog_interval) { 664 setlogmask(LOG_UPTO(LOG_INFO)); 665 openlog("lb", LOG_CONS | LOG_PID | LOG_NDELAY, LOG_LOCAL1); 666 } 667 668 uint32_t npipes = glob_arg.output_rings; 669 670 671 pthread_t stat_thread; 672 673 ports = calloc(npipes + 1, sizeof(struct port_des)); 674 if (!ports) { 675 D("failed to allocate the stats array"); 676 return 1; 677 } 678 struct port_des *rxport = &ports[npipes]; 679 init_groups(); 680 681 memset(&counters_buf, 0, sizeof(counters_buf)); 682 counters_buf.ctrs = calloc(npipes, sizeof(struct my_ctrs)); 683 if (!counters_buf.ctrs) { 684 D("failed to allocate the counters snapshot buffer"); 685 return 1; 686 } 687 688 /* we need base_req to specify pipes and extra bufs */ 689 struct nmreq base_req; 690 memset(&base_req, 0, sizeof(base_req)); 691 692 base_req.nr_arg1 = npipes; 693 base_req.nr_arg3 = glob_arg.extra_bufs; 694 695 rxport->nmd = nm_open(glob_arg.ifname, &base_req, 0, NULL); 696 697 if (rxport->nmd == NULL) { 698 D("cannot open %s", glob_arg.ifname); 699 return (1); 700 } else { 701 D("successfully opened %s (tx rings: %u)", glob_arg.ifname, 702 rxport->nmd->req.nr_tx_slots); 703 } 704 705 uint32_t extra_bufs = rxport->nmd->req.nr_arg3; 706 struct overflow_queue *oq = NULL; 707 /* reference ring to access the buffers */ 708 rxport->ring = NETMAP_RXRING(rxport->nmd->nifp, 0); 709 710 if (!glob_arg.extra_bufs) 711 goto run; 712 713 D("obtained %d extra buffers", extra_bufs); 714 if (!extra_bufs) 715 goto run; 716 717 /* one overflow queue for each output pipe, plus one for the 718 * free extra buffers 719 */ 720 oq = calloc(npipes + 1, sizeof(struct overflow_queue)); 721 if (!oq) { 722 D("failed to allocated overflow queues descriptors"); 723 goto run; 724 } 725 726 freeq = &oq[npipes]; 727 rxport->oq = freeq; 728 729 freeq->slots = calloc(extra_bufs, sizeof(struct netmap_slot)); 730 if (!freeq->slots) { 731 D("failed to allocate the free list"); 732 } 733 freeq->size = extra_bufs; 734 snprintf(freeq->name, MAX_IFNAMELEN, "free queue"); 735 736 /* 737 * the list of buffers uses the first uint32_t in each buffer 738 * as the index of the next buffer. 739 */ 740 uint32_t scan; 741 for (scan = rxport->nmd->nifp->ni_bufs_head; 742 scan; 743 scan = *(uint32_t *)NETMAP_BUF(rxport->ring, scan)) 744 { 745 struct netmap_slot s; 746 s.len = s.flags = 0; 747 s.ptr = 0; 748 s.buf_idx = scan; 749 ND("freeq <- %d", s.buf_idx); 750 oq_enq(freeq, &s); 751 } 752 753 754 if (freeq->n != extra_bufs) { 755 D("something went wrong: netmap reported %d extra_bufs, but the free list contained %d", 756 extra_bufs, freeq->n); 757 return 1; 758 } 759 rxport->nmd->nifp->ni_bufs_head = 0; 760 761run: 762 atexit(free_buffers); 763 764 int j, t = 0; 765 for (j = 0; j < glob_arg.num_groups; j++) { 766 struct group_des *g = &groups[j]; 767 int k; 768 for (k = 0; k < g->nports; ++k) { 769 struct port_des *p = &g->ports[k]; 770 snprintf(p->interface, MAX_PORTNAMELEN, "%s%s{%d/xT@%d", 771 (strncmp(g->pipename, "vale", 4) ? "netmap:" : ""), 772 g->pipename, g->first_id + k, 773 rxport->nmd->req.nr_arg2); 774 D("opening pipe named %s", p->interface); 775 776 p->nmd = nm_open(p->interface, NULL, 0, rxport->nmd); 777 778 if (p->nmd == NULL) { 779 D("cannot open %s", p->interface); 780 return (1); 781 } else if (p->nmd->req.nr_arg2 != rxport->nmd->req.nr_arg2) { 782 D("failed to open pipe #%d in zero-copy mode, " 783 "please close any application that uses either pipe %s}%d, " 784 "or %s{%d, and retry", 785 k + 1, g->pipename, g->first_id + k, g->pipename, g->first_id + k); 786 return (1); 787 } else { 788 D("successfully opened pipe #%d %s (tx slots: %d)", 789 k + 1, p->interface, p->nmd->req.nr_tx_slots); 790 p->ring = NETMAP_TXRING(p->nmd->nifp, 0); 791 p->last_tail = nm_ring_next(p->ring, p->ring->tail); 792 } 793 D("zerocopy %s", 794 (rxport->nmd->mem == p->nmd->mem) ? "enabled" : "disabled"); 795 796 if (extra_bufs) { 797 struct overflow_queue *q = &oq[t + k]; 798 q->slots = calloc(extra_bufs, sizeof(struct netmap_slot)); 799 if (!q->slots) { 800 D("failed to allocate overflow queue for pipe %d", k); 801 /* make all overflow queue management fail */ 802 extra_bufs = 0; 803 } 804 q->size = extra_bufs; 805 snprintf(q->name, sizeof(q->name), "oq %s{%4d", g->pipename, k); 806 p->oq = q; 807 } 808 } 809 t += g->nports; 810 } 811 812 if (glob_arg.extra_bufs && !extra_bufs) { 813 if (oq) { 814 for (i = 0; i < npipes + 1; i++) { 815 free(oq[i].slots); 816 oq[i].slots = NULL; 817 } 818 free(oq); 819 oq = NULL; 820 } 821 D("*** overflow queues disabled ***"); 822 } 823 824 sleep(glob_arg.wait_link); 825 826 /* start stats thread after wait_link */ 827 if (pthread_create(&stat_thread, NULL, print_stats, NULL) == -1) { 828 D("unable to create the stats thread: %s", strerror(errno)); 829 return 1; 830 } 831 832 struct pollfd pollfd[npipes + 1]; 833 memset(&pollfd, 0, sizeof(pollfd)); 834 signal(SIGINT, sigint_h); 835 836 /* make sure we wake up as often as needed, even when there are no 837 * packets coming in 838 */ 839 if (glob_arg.syslog_interval > 0 && glob_arg.syslog_interval < poll_timeout) 840 poll_timeout = glob_arg.syslog_interval; 841 if (glob_arg.stdout_interval > 0 && glob_arg.stdout_interval < poll_timeout) 842 poll_timeout = glob_arg.stdout_interval; 843 844 while (!do_abort) { 845 u_int polli = 0; 846 iter++; 847 848 for (i = 0; i < npipes; ++i) { 849 struct netmap_ring *ring = ports[i].ring; 850 int pending = nm_tx_pending(ring); 851 852 /* if there are packets pending, we want to be notified when 853 * tail moves, so we let cur=tail 854 */ 855 ring->cur = pending ? ring->tail : ring->head; 856 857 if (!glob_arg.busy_wait && !pending) { 858 /* no need to poll, there are no packets pending */ 859 continue; 860 } 861 pollfd[polli].fd = ports[i].nmd->fd; 862 pollfd[polli].events = POLLOUT; 863 pollfd[polli].revents = 0; 864 ++polli; 865 } 866 867 pollfd[polli].fd = rxport->nmd->fd; 868 pollfd[polli].events = POLLIN; 869 pollfd[polli].revents = 0; 870 ++polli; 871 872 //RD(5, "polling %d file descriptors", polli+1); 873 rv = poll(pollfd, polli, poll_timeout); 874 if (rv <= 0) { 875 if (rv < 0 && errno != EAGAIN && errno != EINTR) 876 RD(1, "poll error %s", strerror(errno)); 877 goto send_stats; 878 } 879 880 /* if there are several groups, try pushing released packets from 881 * upstream groups to the downstream ones. 882 * 883 * It is important to do this before returned slots are reused 884 * for new transmissions. For the same reason, this must be 885 * done starting from the last group going backwards. 886 */ 887 for (i = glob_arg.num_groups - 1U; i > 0; i--) { 888 struct group_des *g = &groups[i - 1]; 889 int j; 890 891 for (j = 0; j < g->nports; j++) { 892 struct port_des *p = &g->ports[j]; 893 struct netmap_ring *ring = p->ring; 894 uint32_t last = p->last_tail, 895 stop = nm_ring_next(ring, ring->tail); 896 897 /* slight abuse of the API here: we touch the slot 898 * pointed to by tail 899 */ 900 for ( ; last != stop; last = nm_ring_next(ring, last)) { 901 struct netmap_slot *rs = &ring->slot[last]; 902 // XXX less aggressive? 903 rs->buf_idx = forward_packet(g + 1, rs); 904 rs->flags |= NS_BUF_CHANGED; 905 rs->ptr = 0; 906 } 907 p->last_tail = last; 908 } 909 } 910 911 912 913 if (oq) { 914 /* try to push packets from the overflow queues 915 * to the corresponding pipes 916 */ 917 for (i = 0; i < npipes; i++) { 918 struct port_des *p = &ports[i]; 919 struct overflow_queue *q = p->oq; 920 uint32_t j, lim; 921 struct netmap_ring *ring; 922 struct netmap_slot *slot; 923 924 if (oq_empty(q)) 925 continue; 926 ring = p->ring; 927 lim = nm_ring_space(ring); 928 if (!lim) 929 continue; 930 if (q->n < lim) 931 lim = q->n; 932 for (j = 0; j < lim; j++) { 933 struct netmap_slot s = oq_deq(q), tmp; 934 tmp.ptr = 0; 935 slot = &ring->slot[ring->head]; 936 tmp.buf_idx = slot->buf_idx; 937 oq_enq(freeq, &tmp); 938 *slot = s; 939 slot->flags |= NS_BUF_CHANGED; 940 ring->head = nm_ring_next(ring, ring->head); 941 } 942 } 943 } 944 945 /* push any new packets from the input port to the first group */ 946 int batch = 0; 947 for (i = rxport->nmd->first_rx_ring; i <= rxport->nmd->last_rx_ring; i++) { 948 struct netmap_ring *rxring = NETMAP_RXRING(rxport->nmd->nifp, i); 949 950 //D("prepare to scan rings"); 951 int next_cur = rxring->cur; 952 struct netmap_slot *next_slot = &rxring->slot[next_cur]; 953 const char *next_buf = NETMAP_BUF(rxring, next_slot->buf_idx); 954 while (!nm_ring_empty(rxring)) { 955 struct netmap_slot *rs = next_slot; 956 struct group_des *g = &groups[0]; 957 ++received_pkts; 958 received_bytes += rs->len; 959 960 // CHOOSE THE CORRECT OUTPUT PIPE 961 rs->ptr = pkt_hdr_hash((const unsigned char *)next_buf, 4, 'B'); 962 if (rs->ptr == 0) { 963 non_ip++; // XXX ?? 964 } 965 // prefetch the buffer for the next round 966 next_cur = nm_ring_next(rxring, next_cur); 967 next_slot = &rxring->slot[next_cur]; 968 next_buf = NETMAP_BUF(rxring, next_slot->buf_idx); 969 __builtin_prefetch(next_buf); 970 // 'B' is just a hashing seed 971 rs->buf_idx = forward_packet(g, rs); 972 rs->flags |= NS_BUF_CHANGED; 973 rxring->head = rxring->cur = next_cur; 974 975 batch++; 976 if (unlikely(batch >= glob_arg.batch)) { 977 ioctl(rxport->nmd->fd, NIOCRXSYNC, NULL); 978 batch = 0; 979 } 980 ND(1, 981 "Forwarded Packets: %"PRIu64" Dropped packets: %"PRIu64" Percent: %.2f", 982 forwarded, dropped, 983 ((float)dropped / (float)forwarded * 100)); 984 } 985 986 } 987 988 send_stats: 989 if (counters_buf.status == COUNTERS_FULL) 990 continue; 991 /* take a new snapshot of the counters */ 992 gettimeofday(&counters_buf.ts, NULL); 993 for (i = 0; i < npipes; i++) { 994 struct my_ctrs *c = &counters_buf.ctrs[i]; 995 *c = ports[i].ctr; 996 /* 997 * If there are overflow queues, copy the number of them for each 998 * port to the ctrs.oq_n variable for each port. 999 */ 1000 if (ports[i].oq != NULL) 1001 c->oq_n = ports[i].oq->n; 1002 } 1003 counters_buf.received_pkts = received_pkts; 1004 counters_buf.received_bytes = received_bytes; 1005 counters_buf.non_ip = non_ip; 1006 if (freeq != NULL) 1007 counters_buf.freeq_n = freeq->n; 1008 __sync_synchronize(); 1009 counters_buf.status = COUNTERS_FULL; 1010 } 1011 1012 /* 1013 * If freeq exists, copy the number to the freeq_n member of the 1014 * message struct, otherwise set it to 0. 1015 */ 1016 if (freeq != NULL) { 1017 freeq_n = freeq->n; 1018 } else { 1019 freeq_n = 0; 1020 } 1021 1022 pthread_join(stat_thread, NULL); 1023 1024 printf("%"PRIu64" packets forwarded. %"PRIu64" packets dropped. Total %"PRIu64"\n", forwarded, 1025 dropped, forwarded + dropped); 1026 return 0; 1027} 1028