Changeset 9c4b5e3

Timestamp:

03/20/06 18:29:00 (15 years ago)

Author:

Perry Lorier <perry@…>

Branches:

4.0.1-hotfixes, cachetimestamps, develop, dpdk-ndag, etsilive, getfragoff, help, 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:

c219603

Parents:

25efeac

Message:

Added support for nonblocking events for format_rt

Files:

• lib/format_erf.c (modified) (2 diffs)
• lib/format_helper.c (modified) (1 diff)
• lib/format_rt.c (modified) (17 diffs)
• lib/libtrace.h (modified) (1 diff)
• tools/tracertstats/tracertstats.c (modified) (2 diffs)

lib/format_erf.c

@@ -206 +206 @@
 {
         libtrace->format_data = malloc(sizeof(struct erf_format_data_t));
-
+        
         INPUT.file = NULL;
 
@@ -552 +552 @@
         }
         if (numbytes == 0) {
+                printf("eof\n");
                 return 0;
         }

lib/format_helper.c

@@ -65 +65 @@
                 event.type = TRACE_EVENT_TERMINATE;
                 return event;
-        }
-        if (data>0) {
+        } 
+                
+        if (data>=0) {
                 event.size = trace_read_packet(trace,packet);
                 event.type = TRACE_EVENT_PACKET;

lib/format_rt.c

@@ -115 +115 @@
         rt_hello_t hello_opts;
         uint8_t reason;
+        int oldflags;
+
         
         if ((he=gethostbyname(RT_INFO->hostname)) == NULL) {
@@ -137 +139 @@
                 return -1;
         }
+
+        
+#if 0
+        oldflags = fcntl(RT_INFO->input_fd, F_GETFL, 0);
+        if (oldflags == -1) {
+                trace_set_err(libtrace, errno,
+                                "Could not get fd flags from fd %d\n",
+                                RT_INFO->input_fd);
+                return -1;
+        }
+        oldflags |= O_NONBLOCK;
+        if (fcntl(RT_INFO->input_fd, F_SETFL, oldflags) == -1) {
+                trace_set_err(libtrace, errno,
+                                "Could not set fd flags for fd %d\n",
+                                RT_INFO->input_fd);
+                return -1;
+        }
+#endif
+        
         
         /* We are connected, now receive message from server */
@@ -146 +167 @@
                 return -1;
         }
-
+        
         switch (connect_msg.type) {
                 case RT_DENY_CONN:
@@ -264 +285 @@
 #define RT_BUF_SIZE 4000
 
-static int rt_read(struct libtrace_t *libtrace, void **buffer, size_t len) {
+static int rt_read(struct libtrace_t *libtrace, void **buffer, size_t len, int block) {
         int numbytes;
+        int i;
+        char *buf_ptr;
 
         assert(len <= RT_BUF_SIZE);
@@ -274 +297 @@
                 RT_INFO->buf_left = 0;
         }
+
+        if (block)
+                block=0;
+        else
+                block=MSG_DONTWAIT;
 
         
@@ -289 +317 @@
                                                 RT_INFO->buf_left,
                                                 RT_BUF_SIZE-RT_INFO->buf_left,
-                                                MSG_NOSIGNAL)) == -1) {
+                                                MSG_NOSIGNAL|block)) <= 0) {
+                                if (numbytes == 0) {
+                                        trace_set_err(libtrace, TRACE_ERR_BAD_PACKET, 
+                                                        "No data received");
+                                        return -1;
+                                }
+                                
                                 if (errno == EINTR) {
                                         /* ignore EINTR in case
@@ -296 +330 @@
                                         continue;
                                 }
+                                if (errno == EAGAIN) {
+                                        trace_set_err(libtrace,
+                                                        EAGAIN,
+                                                        "EAGAIN");
+                                        return -1;
+                                }
+                                
                                 perror("recv");
+                                trace_set_err(libtrace, TRACE_ERR_RECV_FAILED,
+                                                "Failed to read data into rt recv buffer");
                                 return -1;
                         }
+                        /*
+                        buf_ptr = RT_INFO->pkt_buffer;
+                        for (i = 0; i < RT_BUF_SIZE ; i++) {
+                                        
+                                printf("%02x", (unsigned char)*buf_ptr);
+                                buf_ptr ++;
+                        }
+                        printf("\n");
+                        */
                         RT_INFO->buf_left+=numbytes;
                 }
@@ -339 +391 @@
                 case RT_DATA_LEGACY_POS:
                         printf("Sending legacy over RT is currently not supported\n");
+                        trace_set_err(libtrace, TRACE_ERR_BAD_PACKET, "Legacy packet cannot be sent over rt");
                         return -1;
                 default:
                         printf("Unrecognised format: %d\n", packet->type);
+                        trace_set_err(libtrace, TRACE_ERR_BAD_PACKET, "Unrecognised packet format");
                         return -1;
         }
@@ -400 +454 @@
                         else {
                                 printf("Error sending ack\n");
+                                trace_set_err(libtrace, TRACE_ERR_BAD_PACKET, 
+                                                "Error sending ack");
                                 return -1;
                         }
@@ -410 +466 @@
         return 1;
 }
-        
-static int rt_read_packet(libtrace_t *libtrace, libtrace_packet_t *packet) {
+
+        
+static int rt_read_packet_versatile(libtrace_t *libtrace,
+                libtrace_packet_t *packet,int blocking) {
         rt_header_t rt_hdr;
         rt_header_t *pkt_hdr = &rt_hdr;
@@ -422 +480 @@
         } 
 
-        packet->header = packet->buffer;
 
         /* FIXME: Better error handling required */
-        if (rt_read(libtrace, (void **)&pkt_hdr, sizeof(rt_header_t)) !=
+        if (rt_read(libtrace, (void **)&pkt_hdr, sizeof(rt_header_t),blocking) !=
                         sizeof(rt_header_t)) {
-                printf("Error receiving rt header\n");
                 return -1;
         }
@@ -436 +492 @@
 
         if (packet->type >= RT_DATA_SIMPLE) {
-                if (rt_read(libtrace, &packet->buffer, pkt_size) != pkt_size) {
+                if (rt_read(libtrace, &packet->buffer, pkt_size,1) != pkt_size) {
                         printf("Error receiving packet\n");
                         return -1;
                 }
+                packet->header = packet->buffer;
                 
                 if (rt_set_format(libtrace, packet) < 0) {
@@ -459 +516 @@
                         case RT_DUCK:
                                 if (rt_read(libtrace, &packet->buffer, 
-                                                        pkt_size) !=
+                                                        pkt_size,1) !=
                                                 pkt_size) {
                                         printf("Error receiving status packet\n");
@@ -483 +540 @@
         }
         /* Return the number of bytes read from the stream */
-        return sizeof(rt_header_t) + packet->size; 
-}
+        return packet->size; 
+}
+
+static int rt_read_packet(libtrace_t *libtrace,
+                libtrace_packet_t *packet) {
+        rt_read_packet_versatile(libtrace,packet,1);
+}
+
 
 static int rt_get_capture_length(const struct libtrace_packet_t *packet) {
@@ -516 +579 @@
         return 0;
 }
+
+static int rt_get_wire_length(const libtrace_packet_t *packet) {
+        return 0;
+}
                         
 static int rt_get_framing_length(const libtrace_packet_t *packet) {
@@ -521 +588 @@
 }
 
-
 static int rt_get_fd(const libtrace_t *trace) {
         return ((struct rt_format_data_t *)trace->format_data)->input_fd;
 }
 
-
+struct libtrace_eventobj_t trace_event_rt(struct libtrace_t *trace, struct libtrace_packet_t *packet) {
+        struct libtrace_eventobj_t event = {0,0,0.0,0};
+        libtrace_err_t read_err;
+        int data;
+
+        assert(trace);
+        assert(packet);
+        
+        if (trace->format->get_fd) {
+                event.fd = trace->format->get_fd(trace);
+        } else {
+                event.fd = 0;
+        }
+
+        event.size = rt_read_packet_versatile(trace, packet, 0);
+        if (event.size == -1) {
+                read_err = trace_get_err(trace);
+                if (read_err.err_num == EAGAIN) {
+                        event.type = TRACE_EVENT_IOWAIT;
+                }
+                else {
+                        printf("packet error\n");
+                        event.type = TRACE_EVENT_PACKET;
+                }
+        } else if (event.size == 0) {
+                event.type = TRACE_EVENT_TERMINATE;
+                
+        }       
+        else {
+                event.type = TRACE_EVENT_PACKET;
+        }
+        
+        return event;
+}
 
 static void rt_help() {
@@ -567 +666 @@
         NULL,                           /* seek_seconds */
         rt_get_capture_length,          /* get_capture_length */
-        NULL,                           /* get_wire_length */
+        rt_get_wire_length,                     /* get_wire_length */
         rt_get_framing_length,          /* get_framing_length */
         NULL,                           /* set_capture_length */
         rt_get_fd,                      /* get_fd */
-        trace_event_device,             /* trace_event */
+        trace_event_rt,             /* trace_event */
         rt_help,                        /* help */
         NULL                            /* next pointer */

lib/libtrace.h

@@ -133 +133 @@
         TRACE_ERR_NO_CONVERSION = -8,
         TRACE_ERR_BAD_PACKET    = -9,
-        TRACE_ERR_OPTION_UNAVAIL= -10
+        TRACE_ERR_OPTION_UNAVAIL= -10,
+        TRACE_ERR_RECV_FAILED   = -11
 };
 

tools/tracertstats/tracertstats.c

@@ -54 +54 @@
 #include "libtrace.h"
 #include "output.h"
+#include "rt_protocol.h"
+#include "dagformat.h"
 
 struct libtrace_t *trace;
@@ -128 +130 @@
         for (;;) {
                 int psize;
+                dag_record_t *erf_hdr;
                 if ((psize = trace_read_packet(trace, packet)) <1) {
                         break;
                 }
+                erf_hdr = (dag_record_t *)packet->header;
+                
+                if (trace_get_link(packet) == NULL) {
+                        continue;
+                }
+                
                 ts = trace_get_seconds(packet);
-
                 for(i=0;i<filter_count;++i) {
                         if(trace_bpf_filter(filters[i].filter,packet)) {