Changeset 32751a2 for lib

Timestamp:

09/04/15 15:23:24 (6 years ago)

Author:

Shane Alcock <salcock@…>

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:

0f6bc3f

Parents:

bf986a1

Message:

Make RT safe for use as a parallel input

Previously, the RT format used to assume that only one read packet would
be in use at any given time, so subsequent read_packet calls could overwrite
the buffer space that previous packets had occupied. With parallel libtrace,
this is no longer the case.

For now, I've gone with the safe but slow solution of copying every
complete RT packet out of the receive buffer into memory space that has
been allocated to the libtrace_packet_t. It'd be nice to do this without
the copy, but we'd need to start tracking which packets have been returned
to us to be able to do this properly.

File:

• lib/format_rt.c (modified) (7 diffs)

lib/format_rt.c

@@ -338 +338 @@
 
 /* Receives data from an RT server */
-static int rt_read(libtrace_t *libtrace, void **buffer, size_t len, int block) 
+static int rt_read(libtrace_t *libtrace, void *buffer, size_t len, int block) 
 {
         int numbytes;
@@ -407 +407 @@
 
         }
-        *buffer = RT_INFO->buf_current;
+        memcpy(buffer, RT_INFO->buf_current, len);
         RT_INFO->buf_current += len;
         RT_INFO->buf_filled -= len;
@@ -585 +585 @@
         uint32_t prep_flags = 0;
 
-        prep_flags |= TRACE_PREP_DO_NOT_OWN_BUFFER;
+        prep_flags |= TRACE_PREP_OWN_BUFFER;
 
         /* The stored RT header will tell us how much data we need to read */
-        if (rt_read(libtrace, &packet->buffer, (size_t)RT_INFO->rt_hdr.length, 
+        if (rt_read(libtrace, packet->buffer, (size_t)RT_INFO->rt_hdr.length, 
                                 blocking) != RT_INFO->rt_hdr.length) {
                 return -1;
@@ -623 +623 @@
 static int rt_read_packet_versatile(libtrace_t *libtrace,
                 libtrace_packet_t *packet,int blocking) {
+        rt_header_t hdr;
         rt_header_t *pkt_hdr = NULL;
         void *void_hdr;
         libtrace_rt_types_t switch_type;
         
-        if (packet->buf_control == TRACE_CTRL_PACKET) {
-                packet->buf_control = TRACE_CTRL_EXTERNAL;
-                free(packet->buffer);
-                packet->buffer = NULL;
-        }
-
         /* RT_LAST indicates that we need to read the RT header for the next
          * packet. This is a touch hax, I admit */
         if (RT_INFO->rt_hdr.type == TRACE_RT_LAST) {
-                void_hdr = (void *)pkt_hdr;
                 /* FIXME: Better error handling required */
-                if (rt_read(libtrace, &void_hdr, 
+                if (rt_read(libtrace, (void *)&hdr, 
                                 sizeof(rt_header_t),blocking) !=
                                 sizeof(rt_header_t)) {
                         return -1;
                 }
-                pkt_hdr = (rt_header_t *)void_hdr;
-                
                 /* Need to store these in case the next rt_read overwrites 
                  * the buffer they came from! */
-                RT_INFO->rt_hdr.type = ntohl(pkt_hdr->type);
-                RT_INFO->rt_hdr.length = ntohs(pkt_hdr->length);
-                RT_INFO->rt_hdr.sequence = ntohl(pkt_hdr->sequence);
-        }
-        packet->type = RT_INFO->rt_hdr.type;
+                RT_INFO->rt_hdr.type = ntohl(hdr.type);
+                RT_INFO->rt_hdr.length = ntohs(hdr.length);
+                RT_INFO->rt_hdr.sequence = ntohl(hdr.sequence);
+        }
+
+        if (packet->buf_control == TRACE_CTRL_PACKET) {
+                if (packet->buffer == NULL) {
+                        packet->buffer = malloc(RT_INFO->rt_hdr.length + sizeof(rt_header_t));
+
+                } else if (RT_INFO->rt_hdr.length > sizeof(packet->buffer)) {
+                        packet->buffer = realloc(packet->buffer, RT_INFO->rt_hdr.length + sizeof(rt_header_t));
+                }
+        }
+        
+        packet->type = RT_INFO->rt_hdr.type;
+        packet->payload = packet->buffer;
+        
         
         /* All data-bearing packets (as opposed to RT internal messages) 
@@ -668 +672 @@
                 case TRACE_RT_STATUS:
                 case TRACE_RT_METADATA:
-                        if (rt_read_data_packet(libtrace, packet, blocking))
+                        if (rt_read_data_packet(libtrace, packet, blocking))
                                 return -1;
                         break;
@@ -677 +681 @@
                 case TRACE_RT_SERVERSTART:
                         /* All these have no payload */
+                        packet->header = packet->buffer;
+                        packet->payload = ((char *)packet->buffer + sizeof(rt_header_t));
+                        pkt_hdr = (rt_header_t *)packet->header;
+                        pkt_hdr->type = ntohl(RT_INFO->rt_hdr.type);
+                        pkt_hdr->length = ntohs(RT_INFO->rt_hdr.length);
+                        pkt_hdr->sequence = ntohl(RT_INFO->rt_hdr.sequence);
+
+                        /* XXX Do we need to save the other crap? */
                         break;
                 case TRACE_RT_PAUSE_ACK:
@@ -689 +701 @@
                         return -1;
         }
-                                
-                        
-                
-        /* Return the number of bytes read from the stream */
+        
+        /* Return the number of bytes read from the stream */
         RT_INFO->rt_hdr.type = TRACE_RT_LAST;
         return RT_INFO->rt_hdr.length + sizeof(rt_header_t);