- Timestamp:
- 02/11/15 18:47:20 (6 years ago)
- Branches:
- 4.0.1-hotfixes, cachetimestamps, develop, dpdk-ndag, etsilive, libtrace4, master, ndag_format, pfring, rc-4.0.1, rc-4.0.2, rc-4.0.3, rc-4.0.4, ringdecrementfix, ringperformance, ringtimestampfixes
- Children:
- 6cf3ca0
- Parents:
- 1871afc
- Location:
- lib
- Files:
-
- 3 edited
Legend:
- Unmodified
- Added
- Removed
-
lib/Makefile.am
r035f8a7 r771ab22 29 29 NATIVEFORMATS+= format_dpdk.c 30 30 # So we also make libtrace.mk in dpdk otherwise automake tries to expand 31 # it to early which I cannot seem to stop unless we use a path that31 # it too early which I cannot seem to stop unless we use a path that 32 32 # doesn't exist currently 33 33 export RTE_SDK=@RTE_SDK@ -
lib/format_linux.c
r035f8a7 r771ab22 72 72 static int linuxnative_init_input(libtrace_t *libtrace) 73 73 { 74 struct linux_per_stream_t stream_data ;74 struct linux_per_stream_t stream_data= ZERO_LINUX_STREAM; 75 75 76 76 libtrace->format_data = (struct linux_format_data_t *) … … 82 82 assert(FORMAT_DATA->per_stream != NULL); 83 83 84 /* We'll start with just one instance of stream_data, and we'll85 * add more later if we need them */86 memset(&stream_data, 0, sizeof(stream_data));87 84 libtrace_list_push_back(FORMAT_DATA->per_stream, &stream_data); 88 85 89 FORMAT_DATA_FIRST->fd = -1;90 86 FORMAT_DATA->promisc = -1; 91 87 FORMAT_DATA->snaplen = LIBTRACE_PACKET_BUFSIZE; … … 103 99 static int linuxnative_init_output(libtrace_out_t *libtrace) 104 100 { 105 libtrace->format_data = (struct linux_ output_format_data_t*)106 malloc(sizeof(struct linux_ output_format_data_t));101 libtrace->format_data = (struct linux_format_data_out_t*) 102 malloc(sizeof(struct linux_format_data_out_t)); 107 103 assert(libtrace->format_data != NULL); 108 104 … … 117 113 } 118 114 119 static int linuxnative_start_input(libtrace_t *libtrace) 115 /* Close an input stream, this is safe to be called part way through 116 * initilisation as a cleanup function assuming streams were set to 117 * ZERO_LINUX_STREAM to begin with. 118 */ 119 static inline void linuxnative_close_input_stream(libtrace_t *libtrace, 120 struct linux_per_stream_t *stream) { 121 if (stream->fd != -1) 122 close(stream->fd); 123 stream->fd = -1; 124 /* TODO maybe store size against stream XXX */ 125 if (stream->rx_ring) 126 munmap(stream->rx_ring, 127 FORMAT_DATA->req.tp_block_size * 128 FORMAT_DATA->req.tp_block_nr); 129 stream->rx_ring = NULL; 130 } 131 132 static inline int linuxnative_start_input_stream(libtrace_t *libtrace, 133 struct linux_per_stream_t *stream) 120 134 { 121 135 struct sockaddr_ll addr; 122 int one = 1;136 const int one = 1; 123 137 memset(&addr,0,sizeof(addr)); 124 138 libtrace_filter_t *filter = FORMAT_DATA->filter; 125 139 126 140 /* Create a raw socket for reading packets on */ 127 FORMAT_DATA_FIRST->fd = 128 socket(PF_PACKET, SOCK_RAW, htons(ETH_P_ALL)); 129 if (FORMAT_DATA_FIRST->fd==-1) { 141 stream->fd = socket(PF_PACKET, SOCK_RAW, htons(ETH_P_ALL)); 142 if (stream->fd==-1) { 130 143 trace_set_err(libtrace, errno, "Could not create raw socket"); 131 free(libtrace->format_data);132 libtrace->format_data = NULL;133 144 return -1; 134 145 } … … 140 151 addr.sll_ifindex = if_nametoindex(libtrace->uridata); 141 152 if (addr.sll_ifindex == 0) { 142 close(FORMAT_DATA_FIRST->fd);153 linuxnative_close_input_stream(libtrace, stream); 143 154 trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, 144 155 "Failed to find interface %s", 145 156 libtrace->uridata); 146 free(libtrace->format_data);147 libtrace->format_data = NULL;148 157 return -1; 149 158 } … … 151 160 addr.sll_ifindex = 0; 152 161 } 153 if (bind(FORMAT_DATA_FIRST->fd, 154 (struct sockaddr*)&addr, 155 (socklen_t)sizeof(addr))==-1) { 156 free(libtrace->format_data); 157 libtrace->format_data = NULL; 162 if (bind(stream->fd, 163 (struct sockaddr*)&addr, 164 (socklen_t)sizeof(addr))==-1) { 165 linuxnative_close_input_stream(libtrace, stream); 158 166 trace_set_err(libtrace, errno, 159 167 "Failed to bind to interface %s", … … 180 188 mreq.mr_ifindex = addr.sll_ifindex; 181 189 mreq.mr_type = PACKET_MR_PROMISC; 182 if (setsockopt( FORMAT_DATA_FIRST->fd,190 if (setsockopt(stream->fd, 183 191 SOL_PACKET, 184 192 PACKET_ADD_MEMBERSHIP, … … 192 200 * clock resolution possible */ 193 201 #ifdef SO_TIMESTAMPNS 194 if (setsockopt( FORMAT_DATA_FIRST->fd,202 if (setsockopt(stream->fd, 195 203 SOL_SOCKET, 196 204 SO_TIMESTAMPNS, … … 203 211 * if we fail the first! */ 204 212 #endif 205 if (setsockopt( FORMAT_DATA_FIRST->fd,213 if (setsockopt(stream->fd, 206 214 SOL_SOCKET, 207 215 SO_TIMESTAMP, … … 218 226 */ 219 227 if (filter != NULL) { 220 221 222 228 /* Check if the filter was successfully compiled. If not, 229 * it is probably a bad filter and we should return an error 230 * before the caller tries to read any packets */ 223 231 if (filter->flag == 0) { 224 return -1; 225 } 226 227 if (setsockopt(FORMAT_DATA_FIRST->fd, 232 linuxnative_close_input_stream(libtrace, stream); 233 trace_set_err(libtrace, TRACE_ERR_BAD_FILTER, 234 "Cannot attach a bad filter to %s", 235 libtrace->uridata); 236 return -1; 237 } 238 239 if (setsockopt(stream->fd, 228 240 SOL_SOCKET, 229 241 SO_ATTACH_FILTER, … … 237 249 */ 238 250 void *buf = malloc((size_t)LIBTRACE_PACKET_BUFSIZE); 239 while(recv( FORMAT_DATA_FIRST->fd,251 while(recv(stream->fd, 240 252 buf, 241 253 (size_t)LIBTRACE_PACKET_BUFSIZE, … … 248 260 249 261 return 0; 262 } 263 264 static int linuxnative_start_input(libtrace_t *libtrace) 265 { 266 int ret = linuxnative_start_input_stream(libtrace, FORMAT_DATA_FIRST); 267 if (ret != 0) { 268 libtrace_list_deinit(FORMAT_DATA->per_stream); 269 free(libtrace->format_data); 270 libtrace->format_data = NULL; 271 } 272 return ret; 250 273 } 251 274 … … 258 281 * @return 0 success, -1 error 259 282 */ 260 static inline int socket_to_packet_fanout(int fd, 261 uint16_t fanout_flags, 262 uint16_t fanout_group) 263 { 264 int fanout_opt = ((int)fanout_flags << 16) | (int)fanout_group; 265 if (setsockopt(fd, SOL_PACKET, PACKET_FANOUT, 283 static inline int linuxnative_socket_to_packet_fanout(libtrace_t *libtrace, 284 struct linux_per_stream_t *stream) 285 { 286 int fanout_opt = ((int)FORMAT_DATA->fanout_flags << 16) | (int)FORMAT_DATA->fanout_group; 287 if (setsockopt(stream->fd, SOL_PACKET, PACKET_FANOUT, 266 288 &fanout_opt, sizeof(fanout_opt)) == -1) { 289 trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, 290 "Converting the fd to a socket fanout failed %s", 291 libtrace->uridata); 267 292 return -1; 268 293 } … … 275 300 int iserror = 0; 276 301 // We store this here otherwise it will be leaked if the memory doesn't know 277 struct linux_per_thread_t *per_thread = NULL; 278 279 if (!FORMAT(libtrace->format_data)->per_thread) { 280 //per_thread = calloc(tot, sizeof(struct linux_per_thread_t)); 281 posix_memalign((void **)&per_thread, CACHE_LINE_SIZE, tot*sizeof(struct linux_per_thread_t)); 282 FORMAT(libtrace->format_data)->per_thread = per_thread; 283 } else { 284 // Whats going on this might not work 100% 285 // We assume all sockets have been closed ;) 286 printf("Pause and then start called again lets hope that perpkt_thread_count hasn't changed\n"); 287 } 288 302 struct linux_per_stream_t empty_stream = ZERO_LINUX_STREAM; 303 289 304 printf("Calling native pstart packet\n"); 290 305 for (i = 0; i < tot; ++i) 291 306 { 292 if (FORMAT(libtrace->format_data)->format == TRACE_RT_DATA_LINUX_NATIVE) { 293 if (linuxnative_start_input(libtrace) != 0) { 307 struct linux_per_stream_t *stream; 308 /* Add storage for another stream */ 309 if (libtrace_list_get_size(FORMAT_DATA->per_stream) <= (size_t) i) 310 libtrace_list_push_back(FORMAT_DATA->per_stream, &empty_stream); 311 312 stream = libtrace_list_get_index(FORMAT_DATA->per_stream, i)->data; 313 if (FORMAT_DATA->format == TRACE_RT_DATA_LINUX_NATIVE) { 314 if (linuxnative_start_input_stream(libtrace, stream) != 0) { 294 315 iserror = 1; 295 316 break; 296 317 } 297 318 } else { 319 perror("BAD CODE XXX TODO PUT CODE HERE!!"); 298 320 // This must be ring 321 /* 299 322 if (linuxring_start_input(libtrace) != 0) { 300 323 iserror = 1; 301 324 break; 302 } 303 } 304 if ( socket_to_packet_fanout(FORMAT(libtrace->format_data)->fd, FORMAT(libtrace->format_data)->fanout_flags, FORMAT(libtrace->format_data)->fanout_group) != 0)325 }*/ 326 } 327 if (linuxnative_socket_to_packet_fanout(libtrace, stream) != 0) 305 328 { 306 329 iserror = 1; 307 // Clean up here to keep consistent with every one else 308 trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "Converting the fd to a socket fanout failed"); 309 close(FORMAT(libtrace->format_data)->fd); 310 free(libtrace->format_data); 311 libtrace->format_data = NULL; 330 close(stream->fd); 331 stream->fd = -1; 312 332 break; 313 333 } 314 per_thread[i].fd = FORMAT(libtrace->format_data)->fd; 315 if (FORMAT(libtrace->format_data)->format == TRACE_RT_DATA_LINUX_RING) { 316 per_thread[i].rxring_offset = FORMAT(libtrace->format_data)->rxring_offset; 317 per_thread[i].rx_ring = FORMAT(libtrace->format_data)->rx_ring; 318 } 319 } 320 321 // Roll back those that failed - by this point in time the format_data 322 // has been freed 334 } 335 336 // Roll back those that failed 323 337 if (iserror) { 324 338 for (i = i - 1; i >= 0; i--) { 325 close(per_thread[i].fd); 326 } 327 free(per_thread); 328 per_thread = NULL; 339 struct linux_per_stream_t *stream; 340 stream = libtrace_list_get_index(FORMAT_DATA->per_stream, i)->data; 341 linuxnative_close_input_stream(libtrace, stream); 342 } 343 libtrace_list_deinit(FORMAT_DATA->per_stream); 344 free(libtrace->format_data); 345 libtrace->format_data = NULL; 329 346 return -1; 330 347 } … … 335 352 static int linux_pregister_thread(libtrace_t *libtrace, libtrace_thread_t *t, bool reading) { 336 353 fprintf(stderr, "registering thread %d!!\n", t->perpkt_num); 337 if (reading) { 338 if(t->type == THREAD_PERPKT) { 339 t->format_data = &FORMAT(libtrace->format_data)->per_thread[t->perpkt_num]; 340 } else { 341 t->format_data = &FORMAT(libtrace->format_data)->per_thread[0]; 342 } 343 } 344 return 0; 354 if (reading) { 355 /* XXX TODO remove this oneday make sure hasher thread still works */ 356 struct linux_per_stream_t *stream; 357 stream = libtrace_list_get_index(FORMAT_DATA->per_stream, 358 t->perpkt_num)->data; 359 t->format_data = stream; 360 if (!stream) { 361 /* This should never happen and indicates an 362 * internal libtrace bug */ 363 trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, 364 "Failed to attached thread %d to a stream", 365 t->perpkt_num); 366 return -1; 367 } 368 } 369 return 0; 345 370 } 346 371 347 372 static int linuxnative_start_output(libtrace_out_t *libtrace) 348 373 { 349 DATAOUT(libtrace)->fd =socket(PF_PACKET, SOCK_RAW, 0);350 if ( DATAOUT(libtrace)->fd==-1) {351 free( DATAOUT(libtrace));374 FORMAT_DATA_OUT->fd = socket(PF_PACKET, SOCK_RAW, 0); 375 if (FORMAT_DATA_OUT->fd==-1) { 376 free(FORMAT_DATA_OUT); 352 377 return -1; 353 378 } … … 359 384 static int linuxnative_pause_input(libtrace_t *libtrace) 360 385 { 361 libtrace_list_node_t *tmp = FORMAT_DATA_HEAD;386 size_t i; 362 387 363 388 /* Stop and detach each stream */ 364 while (tmp != NULL) { 365 close(STREAM_DATA(tmp)->fd); 366 tmp = tmp->next; 389 for (i = 0; i < libtrace_list_get_size(FORMAT_DATA->per_stream); ++i) { 390 struct linux_per_stream_t *stream; 391 stream = libtrace_list_get_index(FORMAT_DATA->per_stream, i)->data; 392 linuxnative_close_input_stream(libtrace, stream); 367 393 } 368 394 … … 387 413 static int linuxnative_fin_output(libtrace_out_t *libtrace) 388 414 { 389 close( DATAOUT(libtrace)->fd);390 DATAOUT(libtrace)->fd=-1;415 close(FORMAT_DATA_OUT->fd); 416 FORMAT_DATA_OUT->fd=-1; 391 417 free(libtrace->format_data); 392 418 return 0; … … 431 457 432 458 pcap = pcap_open_dead(dlt, 433 FORMAT (libtrace->format_data)->snaplen);459 FORMAT_DATA->snaplen); 434 460 435 461 if (pcap_compile(pcap, &f->filter, f->filterstring, 0, 0) == -1) { … … 453 479 } 454 480 455 if (FORMAT (libtrace->format_data)->filter != NULL)456 free(FORMAT (libtrace->format_data)->filter);457 458 FORMAT (libtrace->format_data)->filter = f;481 if (FORMAT_DATA->filter != NULL) 482 free(FORMAT_DATA->filter); 483 484 FORMAT_DATA->filter = f; 459 485 460 486 return 0; … … 469 495 switch(option) { 470 496 case TRACE_OPTION_SNAPLEN: 471 FORMAT (libtrace->format_data)->snaplen=*(int*)data;497 FORMAT_DATA->snaplen=*(int*)data; 472 498 return 0; 473 499 case TRACE_OPTION_PROMISC: 474 FORMAT (libtrace->format_data)->promisc=*(int*)data;500 FORMAT_DATA->promisc=*(int*)data; 475 501 return 0; 476 502 case TRACE_OPTION_FILTER: … … 505 531 case HASHER_BALANCE: 506 532 // Do fanout 507 FORMAT (libtrace->format_data)->fanout_flags = PACKET_FANOUT_LB;533 FORMAT_DATA->fanout_flags = PACKET_FANOUT_LB; 508 534 // Or we could balance to the CPU 509 535 return 0; 510 536 case HASHER_BIDIRECTIONAL: 511 537 case HASHER_UNIDIRECTIONAL: 512 FORMAT (libtrace->format_data)->fanout_flags = PACKET_FANOUT_HASH;538 FORMAT_DATA->fanout_flags = PACKET_FANOUT_HASH; 513 539 return 0; 514 540 case HASHER_CUSTOM: … … 533 559 libtrace_rt_types_t rt_type, uint32_t flags) { 534 560 535 536 537 538 539 540 541 542 543 544 545 546 547 561 if (packet->buffer != buffer && 562 packet->buf_control == TRACE_CTRL_PACKET) { 563 free(packet->buffer); 564 } 565 566 if ((flags & TRACE_PREP_OWN_BUFFER) == TRACE_PREP_OWN_BUFFER) { 567 packet->buf_control = TRACE_CTRL_PACKET; 568 } else 569 packet->buf_control = TRACE_CTRL_EXTERNAL; 570 571 572 packet->buffer = buffer; 573 packet->header = buffer; 548 574 packet->payload = (char *)buffer + 549 575 sizeof(struct libtrace_linuxnative_header); … … 567 593 568 594 #ifdef HAVE_NETPACKET_PACKET_H 569 libtrace_thread_t * get_thread_table(libtrace_t *libtrace) ; 570 inline static int linuxnative_read_packet_fd(libtrace_t *libtrace, libtrace_packet_t *packet, int fd, const int check_queue) 595 inline static int linuxnative_read_stream(libtrace_t *libtrace, 596 libtrace_packet_t *packet, 597 struct linux_per_stream_t *stream, 598 libtrace_message_queue_t *queue) 571 599 { 572 600 struct libtrace_linuxnative_header *hdr; … … 578 606 579 607 uint32_t flags = 0; 580 581 582 608 fd_set readfds; 609 struct timeval tout; 610 int ret; 583 611 584 612 if (!packet->buffer || packet->buf_control == TRACE_CTRL_EXTERNAL) { … … 596 624 snaplen=LIBTRACE_MIN( 597 625 (int)LIBTRACE_PACKET_BUFSIZE-(int)sizeof(*hdr), 598 (int)FORMAT (libtrace->format_data)->snaplen);626 (int)FORMAT_DATA->snaplen); 599 627 /* Prepare the msghdr and iovec for the kernel to write the 600 628 * captured packet into. The msghdr will point to the part of our … … 614 642 iovec.iov_base = (void*)(packet->buffer+sizeof(*hdr)); 615 643 iovec.iov_len = snaplen; 616 617 if (check_queue) { 618 // Check for a packet - TODO only Linux has MSG_DONTWAIT should use fctl O_NONBLOCK 619 hdr->wirelen = recvmsg(fd, &msghdr, MSG_DONTWAIT | MSG_TRUNC); 620 if ((int) hdr->wirelen == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { 621 // Do message queue check or select 622 int ret; 623 fd_set rfds; 624 FD_ZERO(&rfds); 625 FD_SET(fd, &rfds); 626 FD_SET(get_thread_table(libtrace)->messages.pipefd[0], &rfds); 627 int largestfd = fd > get_thread_table(libtrace)->messages.pipefd[0] ? fd : get_thread_table(libtrace)->messages.pipefd[0]; 628 do { 629 ret = select(largestfd+1, &rfds, NULL, NULL, NULL); 630 if (ret == -1 && errno != EINTR) 631 perror("Select() failed"); 644 645 // Check for a packet - TODO only Linux has MSG_DONTWAIT should use fctl O_NONBLOCK 646 /* Try check ahead this should be fast if something is waiting */ 647 hdr->wirelen = recvmsg(stream->fd, &msghdr, MSG_DONTWAIT | MSG_TRUNC); 648 649 /* No data was waiting */ 650 if ((int) hdr->wirelen == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { 651 /* Do message queue check or select */ 652 int message_fd; 653 int largestfd = stream->fd; 654 655 /* Also check the message queue */ 656 if (queue) { 657 message_fd = libtrace_message_queue_get_fd(queue); 658 if (message_fd > largestfd) 659 largestfd = message_fd; 660 } 661 do { 662 /* Use select to allow us to time out occasionally to check if someone 663 * has hit Ctrl-C or otherwise wants us to stop reading and return 664 * so they can exit their program. 665 */ 666 tout.tv_sec = 0; 667 tout.tv_usec = 500000; 668 /* Make sure we reset these each loop */ 669 FD_ZERO(&readfds); 670 FD_SET(stream->fd, &readfds); 671 if (queue) 672 FD_SET(message_fd, &readfds); 673 674 ret = select(largestfd+1, &readfds, NULL, NULL, &tout); 675 if (ret >= 1) { 676 /* A file descriptor triggered */ 677 break; 678 } else if (ret < 0 && errno != EINTR) { 679 trace_set_err(libtrace, errno, "select"); 680 return -1; 681 } else { 682 if (libtrace_halt) 683 return READ_EOF; 632 684 } 633 while (ret == -1); 634 635 assert (ret == 1 || ret == 2); // No timeout 0 is not an option 636 637 if (FD_ISSET(get_thread_table(libtrace)->messages.pipefd[0], &rfds)) { 638 // Not an error but check the message queue we have something 639 return -2; 640 } 641 // Otherwise we must have a packet 642 hdr->wirelen = recvmsg(fd, &msghdr, MSG_TRUNC); 643 } 644 } else { 645 /* Use select to allow us to time out occasionally to check if someone 646 * has hit Ctrl-C or otherwise wants us to stop reading and return 647 * so they can exit their program. 648 */ 649 650 while (1) { 651 tout.tv_sec = 0; 652 tout.tv_usec = 500000; 653 FD_ZERO(&readfds); 654 FD_SET(FORMAT(libtrace->format_data)->fd, &readfds); 655 656 ret = select(FORMAT(libtrace->format_data)->fd + 1, &readfds, 657 NULL, NULL, &tout); 658 if (ret < 0 && errno != EINTR) { 659 trace_set_err(libtrace, errno, "select"); 660 return -1; 661 } else if (ret < 0) { 662 continue; 663 } 664 665 if (FD_ISSET(FORMAT(libtrace->format_data)->fd, &readfds)) { 666 /* There's something available for us to read */ 667 break; 668 } 669 670 671 /* If we get here, we timed out -- check if we should halt */ 672 if (libtrace_halt) 673 return 0; 674 } 675 hdr->wirelen = recvmsg(FORMAT(libtrace->format_data)->fd, &msghdr, MSG_TRUNC); 676 } 677 678 685 } 686 while (ret <= 0); 687 688 /* Message waiting? */ 689 if (queue && FD_ISSET(message_fd, &readfds)) 690 return READ_MESSAGE; 691 692 /* We must have a packet */ 693 hdr->wirelen = recvmsg(stream->fd, &msghdr, MSG_TRUNC); 694 } 695 679 696 if (hdr->wirelen==~0U) { 680 697 trace_set_err(libtrace,errno,"recvmsg"); … … 724 741 if (cmsg == NULL) { 725 742 struct timeval tv; 726 if (ioctl( fd, SIOCGSTAMP,&tv)==0) {743 if (ioctl(stream->fd, SIOCGSTAMP,&tv)==0) { 727 744 hdr->tv.tv_sec = tv.tv_sec; 728 745 hdr->tv.tv_usec = tv.tv_usec; … … 747 764 static int linuxnative_read_packet(libtrace_t *libtrace, libtrace_packet_t *packet) 748 765 { 749 int fd = FORMAT(libtrace->format_data)->fd; 750 return linuxnative_read_packet_fd(libtrace, packet, fd, 0); 766 return linuxnative_read_stream(libtrace, packet, FORMAT_DATA_FIRST, NULL); 751 767 } 752 768 … … 756 772 UNUSED size_t nb_packets) { 757 773 /* For now just read one packet */ 758 int fd = PERPKT_FORMAT(t)->fd; 759 packets[0]->error = linuxnative_read_packet_fd(libtrace, packets[0], 760 fd, 1); 774 packets[0]->error = linuxnative_read_stream(libtrace, packets[0], 775 t->format_data, &t->messages); 761 776 if (packets[0]->error >= 1) 762 777 return 1; … … 765 780 } 766 781 767 static int linuxnative_write_packet(libtrace_out_t * trace,782 static int linuxnative_write_packet(libtrace_out_t *libtrace, 768 783 libtrace_packet_t *packet) 769 784 { … … 776 791 hdr.sll_family = AF_PACKET; 777 792 hdr.sll_protocol = 0; 778 hdr.sll_ifindex = if_nametoindex( trace->uridata);793 hdr.sll_ifindex = if_nametoindex(libtrace->uridata); 779 794 hdr.sll_hatype = 0; 780 795 hdr.sll_pkttype = 0; … … 784 799 /* This is pretty easy, just send the payload using sendto() (after 785 800 * setting up the sll header properly, of course) */ 786 ret = sendto( DATAOUT(trace)->fd,801 ret = sendto(FORMAT_DATA_OUT->fd, 787 802 packet->payload, 788 803 trace_get_capture_length(packet), … … 791 806 792 807 if (ret < 0) { 793 trace_set_err_out( trace, errno, "sendto failed");808 trace_set_err_out(libtrace, errno, "sendto failed"); 794 809 } 795 810 … … 896 911 } 897 912 898 static int linuxnative_get_fd(const libtrace_t * trace) {899 if ( trace->format_data == NULL)913 static int linuxnative_get_fd(const libtrace_t *libtrace) { 914 if (libtrace->format_data == NULL) 900 915 return -1; 901 return FORMAT (trace->format_data)->fd;916 return FORMAT_DATA_FIRST->fd; 902 917 } 903 918 … … 913 928 } 914 929 930 #ifdef HAVE_NETPACKET_PACKET_H 931 static void linuxnative_update_statistics(libtrace_t *libtrace) { 932 struct tpacket_stats stats; 933 size_t i; 934 socklen_t len = sizeof(stats); 935 936 for (i = 0; i < libtrace_list_get_size(FORMAT_DATA->per_stream); ++i) { 937 struct linux_per_stream_t *stream; 938 stream = libtrace_list_get_index(FORMAT_DATA->per_stream, i)->data; 939 if (stream->fd != -1) { 940 if (getsockopt(stream->fd, 941 SOL_PACKET, 942 PACKET_STATISTICS, 943 &stats, 944 &len) == 0) { 945 if (FORMAT_DATA->stats_valid==0) { 946 FORMAT_DATA->stats.tp_drops = stats.tp_drops; 947 FORMAT_DATA->stats.tp_packets = stats.tp_packets; 948 FORMAT_DATA->stats_valid = 1; 949 } else { 950 FORMAT_DATA->stats.tp_drops += stats.tp_drops; 951 FORMAT_DATA->stats.tp_drops += stats.tp_packets; 952 } 953 } else { 954 perror("getsockopt PACKET_STATISTICS failed"); 955 } 956 } 957 } 958 } 959 #endif 960 915 961 /* Number of packets that passed filtering */ 916 static uint64_t linuxnative_get_captured_packets(libtrace_t *trace) { 917 struct tpacket_stats stats; 918 919 if (trace->format_data == NULL) 962 static uint64_t linuxnative_get_captured_packets(libtrace_t *libtrace) { 963 if (libtrace->format_data == NULL) 920 964 return UINT64_MAX; 921 if (FORMAT (trace->format_data)->fd == -1) {965 if (FORMAT_DATA_FIRST->fd == -1) { 922 966 /* This is probably a 'dead' trace so obviously we can't query 923 967 * the socket for capture counts, can we? */ … … 926 970 927 971 #ifdef HAVE_NETPACKET_PACKET_H 928 929 if ((FORMAT(trace->format_data)->stats_valid & 1) 930 || FORMAT(trace->format_data)->stats_valid == 0) { 931 if (FORMAT(trace->format_data)->per_thread) { 932 int i; 933 FORMAT(trace->format_data)->stats.tp_drops = 0; 934 FORMAT(trace->format_data)->stats.tp_packets = 0; 935 for (i = 0; i < trace->perpkt_thread_count; ++i) { 936 socklen_t len = sizeof(stats); 937 getsockopt(FORMAT(trace->format_data)->per_thread[i].fd, 938 SOL_PACKET, 939 PACKET_STATISTICS, 940 &stats, 941 &len); 942 FORMAT(trace->format_data)->stats.tp_drops += stats.tp_drops; 943 FORMAT(trace->format_data)->stats.tp_packets += stats.tp_packets; 944 } 945 FORMAT(trace->format_data)->stats_valid |= 1; 946 } else { 947 socklen_t len = sizeof(FORMAT(trace->format_data)->stats); 948 getsockopt(FORMAT(trace->format_data)->fd, 949 SOL_PACKET, 950 PACKET_STATISTICS, 951 &FORMAT(trace->format_data)->stats, 952 &len); 953 FORMAT(trace->format_data)->stats_valid |= 1; 954 } 955 } 956 957 return FORMAT(trace->format_data)->stats.tp_packets; 972 linuxnative_update_statistics(libtrace); 973 if (FORMAT_DATA->stats_valid) 974 return FORMAT_DATA->stats.tp_packets; 975 else 976 return UINT64_MAX; 958 977 #else 959 978 return UINT64_MAX; … … 961 980 } 962 981 982 963 983 /* Number of packets that got past filtering and were then dropped because 964 * of lack of space 984 * of lack of space. 985 * 986 * We could also try read from /sys/class/net/ethX/statistics/ to get 987 * real drop counters and stuff. 965 988 */ 966 static uint64_t linuxnative_get_dropped_packets(libtrace_t *trace) { 967 struct tpacket_stats stats; 968 if (trace->format_data == NULL) 989 static uint64_t linuxnative_get_dropped_packets(libtrace_t *libtrace) { 990 if (libtrace->format_data == NULL) 969 991 return UINT64_MAX; 970 if (FORMAT (trace->format_data)->fd == -1) {992 if (FORMAT_DATA_FIRST->fd == -1) { 971 993 /* This is probably a 'dead' trace so obviously we can't query 972 994 * the socket for drop counts, can we? */ … … 974 996 } 975 997 976 #ifdef HAVE_NETPACKET_PACKET_H 977 if ((FORMAT(trace->format_data)->stats_valid & 2) 978 || (FORMAT(trace->format_data)->stats_valid==0)) { 979 if (FORMAT(trace->format_data)->per_thread) { 980 int i; 981 FORMAT(trace->format_data)->stats.tp_drops = 0; 982 FORMAT(trace->format_data)->stats.tp_packets = 0; 983 for (i = 0; i < trace->perpkt_thread_count; ++i) { 984 socklen_t len = sizeof(stats); 985 getsockopt(FORMAT(trace->format_data)->per_thread[i].fd, 986 SOL_PACKET, 987 PACKET_STATISTICS, 988 &stats, 989 &len); 990 FORMAT(trace->format_data)->stats.tp_drops += stats.tp_drops; 991 FORMAT(trace->format_data)->stats.tp_packets += stats.tp_packets; 992 } 993 FORMAT(trace->format_data)->stats_valid |= 2; 994 } else { 995 socklen_t len = sizeof(FORMAT(trace->format_data)->stats); 996 getsockopt(FORMAT(trace->format_data)->fd, 997 SOL_PACKET, 998 PACKET_STATISTICS, 999 &FORMAT(trace->format_data)->stats, 1000 &len); 1001 FORMAT(trace->format_data)->stats_valid |= 2; 1002 } 1003 } 1004 1005 return FORMAT(trace->format_data)->stats.tp_drops; 998 #ifdef HAVE_NETPACKET_PACKET_H 999 linuxnative_update_statistics(libtrace); 1000 if (FORMAT_DATA->stats_valid) 1001 return FORMAT_DATA->stats.tp_drops; 1002 else 1003 return UINT64_MAX; 1006 1004 #else 1007 1005 return UINT64_MAX; … … 1065 1063 linuxnative_pstart_input, /* pstart_input */ 1066 1064 linuxnative_pread_packets, /* pread_packets */ 1067 linuxnative_p pause_input, /* ppause */1065 linuxnative_pause_input, /* ppause */ 1068 1066 linuxnative_fin_input, /* p_fin */ 1069 1067 linuxnative_pconfig_input, /* pconfig input */ -
lib/format_linux.h
r1871afc r771ab22 244 244 } ALIGN_STRUCT(CACHE_LINE_SIZE); 245 245 246 #define ZERO_LINUX_STREAM {-1, NULL, 0} 247 248 246 249 /* Format header for encapsulating packets captured using linux native */ 247 250 struct libtrace_linuxnative_header {
Note: See TracChangeset
for help on using the changeset viewer.