source: lib/format_rt.c @ e9f9ab3

4.0.1-hotfixescachetimestampsdevelopdpdk-ndagetsilivegetfragoffhelplibtrace4ndag_formatpfringrc-4.0.1rc-4.0.2rc-4.0.3rc-4.0.4ringdecrementfixringperformanceringtimestampfixes
Last change on this file since e9f9ab3 was e9f9ab3, checked in by Shane Alcock <salcock@…>, 16 years ago

Shifted setting packet->trace and packet->payload into separate functions

  • Property mode set to 100644
File size: 12.4 KB
Line 
1/*
2 * This file is part of libtrace
3 *
4 * Copyright (c) 2004 The University of Waikato, Hamilton, New Zealand.
5 * Authors: Daniel Lawson
6 *          Perry Lorier
7 *          Shane Alcock
8 *
9 * All rights reserved.
10 *
11 * This code has been developed by the University of Waikato WAND
12 * research group. For further information please see http://www.wand.net.nz/
13 *
14 * libtrace is free software; you can redistribute it and/or modify
15 * it under the terms of the GNU General Public License as published by
16 * the Free Software Foundation; either version 2 of the License, or
17 * (at your option) any later version.
18 *
19 * libtrace is distributed in the hope that it will be useful,
20 * but WITHOUT ANY WARRANTY; without even the implied warranty of
21 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
22 * GNU General Public License for more details.
23 *
24 * You should have received a copy of the GNU General Public License
25 * along with libtrace; if not, write to the Free Software
26 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
27 *
28 * $Id$
29 *
30 */
31
32#define _GNU_SOURCE
33
34#include "config.h"
35#include "common.h"
36#include "libtrace.h"
37#include "libtrace_int.h"
38#include "format_helper.h"
39#include "parse_cmd.h"
40#include "rt_protocol.h"
41
42#ifdef HAVE_INTTYPES_H
43#  include <inttypes.h>
44#else
45#  error "Can't find inttypes.h - this needs to be fixed"
46#endif
47
48#ifdef HAVE_STDDEF_H
49#  include <stddef.h>
50#else
51# error "Can't find stddef.h - do you define ptrdiff_t elsewhere?"
52#endif
53#include <sys/types.h>
54#include <sys/socket.h>
55#include <sys/un.h>
56#include <sys/mman.h>
57#include <sys/stat.h>
58#include <unistd.h>
59#include <assert.h>
60#include <errno.h>
61#include <netdb.h>
62#include <fcntl.h>
63#include <getopt.h>
64#include <stdio.h>
65#include <string.h>
66#include <stdlib.h>
67
68#define RT_INFO libtrace->format_data
69
70struct libtrace_format_data_t {
71        char *hostname;
72        int port;
73        int input_fd;
74        int reliable;
75
76        struct libtrace_t *dummy_erf;
77        struct libtrace_t *dummy_pcap;
78        struct libtrace_t *dummy_wag;
79};
80
81static struct libtrace_format_t rt;
82
83static int rt_connect(struct libtrace_t *libtrace) {
84        struct hostent *he;
85        struct sockaddr_in remote;
86        rt_header_t connect_msg;
87        uint8_t reason; 
88       
89        if ((he=gethostbyname(RT_INFO->hostname)) == NULL) {
90                perror("gethostbyname");
91                return 0;
92        }
93        if ((RT_INFO->input_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
94                perror("socket");
95                return 0;
96        }
97
98        remote.sin_family = AF_INET;
99        remote.sin_port = htons(RT_INFO->port);
100        remote.sin_addr = *((struct in_addr *)he->h_addr);
101        bzero(&(remote.sin_zero), 8);
102
103        if (connect(RT_INFO->input_fd, (struct sockaddr *)&remote,
104                                sizeof(struct sockaddr)) == -1) {
105                perror("connect (inet)");
106                return 0;
107        }
108       
109        // We are connected, now receive message from server
110       
111        if (recv(RT_INFO->input_fd, &connect_msg, sizeof(rt_header_t), 0) != sizeof(rt_header_t) ) {
112                printf("An error occured while connecting to %s\n", RT_INFO->hostname);
113                return -1;
114        }
115
116        switch (connect_msg.type) {
117                case RT_DENY_CONN:
118                       
119                        if (recv(RT_INFO->input_fd, &reason, 1, 0) != 1) {
120                                reason = 0;
121                        }       
122                        printf("Connection attempt is denied by the server: %s\n",
123                                        rt_deny_reason(reason));
124                        return -1;
125                case RT_HELLO:
126                        // do something with options
127                       
128                       
129                        return 1;
130                case RT_DATA:
131                        printf("Server needs to send RT_HELLO before sending data to clients\n");
132                        return -1;
133                default:
134                        printf("Unexpected message type: %d\n", connect_msg.type);
135                        return -1;
136        }
137       
138        return -1;
139}
140
141
142static int rt_init_input(struct libtrace_t *libtrace) {
143        char *scan;
144        char *uridata = libtrace->uridata;
145        libtrace->format_data = (struct libtrace_format_data_t *)
146                malloc(sizeof(struct libtrace_format_data_t));
147
148        //libtrace->sourcetype = RT;
149
150        RT_INFO->dummy_erf = NULL;
151        RT_INFO->dummy_pcap = NULL;
152        RT_INFO->dummy_wag = NULL;
153       
154        if (strlen(uridata) == 0) {
155                RT_INFO->hostname =
156                        strdup("localhost");
157                RT_INFO->port =
158                        COLLECTOR_PORT;
159        } else {
160                if ((scan = strchr(uridata,':')) == NULL) {
161                        RT_INFO->hostname =
162                                strdup(uridata);
163                        RT_INFO->port =
164                                COLLECTOR_PORT;
165                } else {
166                        RT_INFO->hostname =
167                                (char *)strndup(uridata,
168                                                (scan - uridata));
169                        RT_INFO->port =
170                                atoi(++scan);
171                }
172        }
173
174        return rt_connect(libtrace);
175}
176       
177static int rt_start_input(struct libtrace_t *libtrace) {
178        rt_header_t start_msg;
179
180        start_msg.type = RT_START;
181        start_msg.length = sizeof(rt_header_t);
182       
183        // Need to send start message to server
184        if (send(RT_INFO->input_fd, &start_msg, sizeof(rt_header_t), 0) != sizeof(rt_header_t)) {
185                printf("Failed to send start message to server\n");
186                return -1;
187        }
188
189        return 1;
190}
191
192static int rt_fin_input(struct libtrace_t *libtrace) {
193        rt_header_t close_msg;
194
195        close_msg.type = RT_CLOSE;
196        close_msg.length = sizeof(rt_header_t);
197       
198        // Send a close message to the server
199        if (send(RT_INFO->input_fd, &close_msg, sizeof(rt_header_t), 0) != sizeof(rt_header_t)) {
200                printf("Failed to send close message to server\n");
201       
202        }
203        if (RT_INFO->dummy_erf) 
204                trace_destroy_dead(RT_INFO->dummy_erf);
205               
206        if (RT_INFO->dummy_pcap)
207                trace_destroy_dead(RT_INFO->dummy_pcap);
208
209        if (RT_INFO->dummy_wag)
210                trace_destroy_dead(RT_INFO->dummy_wag);
211        close(RT_INFO->input_fd);
212        return 0;
213}
214
215static int rt_read(struct libtrace_t *libtrace, void *buffer, size_t len) {
216        int numbytes;
217
218        if (buffer == 0)
219                buffer = malloc(len);
220        while(1) {
221#ifndef MSG_NOSIGNAL
222#  define MSG_NOSIGNAL 0
223#endif
224                if ((numbytes = recv(RT_INFO->input_fd,
225                                                buffer,
226                                                len,
227                                                MSG_NOSIGNAL)) == -1) {
228                        if (errno == EINTR) {
229                                //ignore EINTR in case
230                                // a caller is using signals
231                                continue;
232                        }
233                        perror("recv");
234                        return -1;
235                }
236                break;
237
238        }
239        return numbytes;
240}
241
242
243static int rt_set_format(struct libtrace_t *libtrace, 
244                struct libtrace_packet_t *packet, uint16_t format) {
245        switch (format) {
246                case RT_FORMAT_ERF:
247                        if (!RT_INFO->dummy_erf) {
248                                RT_INFO->dummy_erf = trace_create_dead("erf:-");
249                        }
250                        packet->trace = RT_INFO->dummy_erf;
251                        break;
252                case RT_FORMAT_PCAP:
253                        if (!RT_INFO->dummy_pcap) {
254                                RT_INFO->dummy_pcap = trace_create_dead("pcap:-");
255                        }
256                        packet->trace = RT_INFO->dummy_pcap;
257                        break;
258                case RT_FORMAT_WAG:
259                        if (!RT_INFO->dummy_wag) {
260                                RT_INFO->dummy_wag = trace_create_dead("wtf:-");
261                        }
262                        packet->trace = RT_INFO->dummy_wag;
263                        break;
264                default:
265                        printf("Unrecognised format: %d\n", format);
266                        return -1;
267        }
268        return 1;
269}               
270
271static void rt_set_payload(struct libtrace_packet_t *packet, uint16_t format) {
272        dag_record_t *erfptr;
273       
274        switch (format) {
275                case RT_FORMAT_ERF:
276                        erfptr = (dag_record_t *)packet->header;
277                       
278                        if (erfptr->flags.rxerror == 1) {
279                                packet->payload = NULL;
280                        } else {
281                                packet->payload = (char *)packet->buffer
282                                        + trace_get_framing_length(packet);
283                        }
284                        break;
285                default:
286                        packet->payload = (char *)packet->buffer +
287                                trace_get_framing_length(packet);
288                        break;
289        }
290}
291
292static int rt_read_packet(struct libtrace_t *libtrace, struct libtrace_packet_t *packet) {
293       
294        int numbytes = 0;
295        char buf[RP_BUFSIZE];
296        int read_required = 0;
297        rt_header_t pkt_hdr;
298        uint16_t format;
299        char msg_buf[RP_BUFSIZE];
300
301       
302        void *buffer = 0;
303
304        packet->trace = libtrace;
305
306        if (packet->buf_control == EXTERNAL) {
307                packet->buf_control = PACKET;
308                packet->buffer = malloc(LIBTRACE_PACKET_BUFSIZE);
309        }
310
311        buffer = packet->buffer;
312        packet->header = packet->buffer;
313
314
315        do {
316                if (tracefifo_out_available(libtrace->fifo) == 0 || read_required) {
317                        if ((numbytes = rt_read(
318                                        libtrace,buf,RP_BUFSIZE))<=0) {
319                                return numbytes;
320                        }
321                        tracefifo_write(libtrace->fifo,buf,numbytes);
322                        read_required = 0;
323                }
324                // Read rt header
325                if (tracefifo_out_read(libtrace->fifo,
326                                &pkt_hdr, sizeof(rt_header_t)) == 0) {
327                        read_required = 1;
328                        continue;
329                }
330                tracefifo_out_update(libtrace->fifo, sizeof(rt_header_t));
331               
332                packet->size = pkt_hdr.length;
333                packet->type = pkt_hdr.type;
334
335                switch (packet->type) {
336                        case RT_DATA:
337                                if (tracefifo_out_read(libtrace->fifo, 
338                                                        &format, 
339                                                        sizeof(uint16_t)) == 0) 
340                                {
341                                        tracefifo_out_reset(libtrace->fifo);
342                                        read_required = 1;
343                                        break;
344                                }
345                                if (tracefifo_out_read(libtrace->fifo, buffer, 
346                                                        packet->size - 
347                                                        sizeof(uint16_t)) == 0)
348                                {
349                                        tracefifo_out_reset(libtrace->fifo);
350                                        read_required = 1;
351                                        break;
352                                }
353                                // set packet->trace
354                                if (rt_set_format(libtrace, packet, format) < 0) {
355                                        return -1;
356                                }
357                                // set packet->payload
358                                rt_set_payload(packet, format);
359                                // send ack
360                                break;
361                        case RT_STATUS:
362                                if (tracefifo_out_read(libtrace->fifo, buffer,
363                                                sizeof(rt_status_t)) == 0)
364                                {
365                                        tracefifo_out_reset(libtrace->fifo);
366                                        read_required = 1;
367                                        break;
368                                }
369                                break;
370                        case RT_DUCK:
371                                if (tracefifo_out_read(libtrace->fifo, buffer,
372                                                sizeof(rt_duck_t)) == 0)
373                                {
374                                        tracefifo_out_reset(libtrace->fifo);
375                                        read_required = 1;
376                                        break;
377                                }
378                                break;
379
380                        case RT_END_DATA:
381                                // need to do something sensible here
382                               
383                                break;
384
385                        case RT_PAUSE_ACK:
386                                // Check if we asked for a pause
387                               
388                               
389                                break;
390
391                        case RT_OPTION:
392                                // Server is requesting some option?
393
394                                break;
395
396                        default:
397                                printf("Bad rt type: %d\n", packet->type);
398                                return -1;
399                               
400                }
401                if (read_required)
402                        continue;
403                               
404               
405                // got in our whole packet, so...
406                tracefifo_out_update(libtrace->fifo,packet->size);
407
408                tracefifo_ack_update(libtrace->fifo,packet->size +
409                                sizeof(rt_header_t));
410                return 1;
411        } while(1);
412       
413}
414
415static int rt_get_fd(const struct libtrace_packet_t *packet) {
416        return packet->trace->format_data->input_fd;
417}
418
419
420
421static void rt_help() {
422        printf("rt format module\n");
423        printf("Supported input URIs:\n");
424        printf("\trt:hostname:port\n");
425        printf("\trt:hostname (connects on default port)\n");
426        printf("\n");
427        printf("\te.g.: rt:localhost\n");
428        printf("\te.g.: rt:localhost:32500\n");
429        printf("\n");
430
431}
432
433
434static struct libtrace_format_t rt = {
435        "rt",
436        "$Id$",
437        "rt",
438        rt_init_input,            /* init_input */
439        NULL,                           /* config_input */
440        rt_start_input,           /* start_input */
441        NULL,                           /* init_output */
442        NULL,                           /* config_output */
443        NULL,                           /* start_output */
444        rt_fin_input,             /* fin_input */
445        NULL,                           /* fin_output */
446        rt_read_packet,           /* read_packet */
447        NULL,                           /* write_packet */
448        NULL,                           /* get_link_type */
449        NULL,                           /* get_direction */
450        NULL,                           /* set_direction */
451        NULL,                           /* get_erf_timestamp */
452        NULL,                           /* get_timeval */
453        NULL,                           /* get_seconds */
454        NULL,                           /* get_capture_length */
455        NULL,                           /* get_wire_length */
456        NULL,                           /* get_framing_length */
457        NULL,                           /* set_capture_length */
458        rt_get_fd,                /* get_fd */
459        trace_event_device,             /* trace_event */
460        rt_help                   /* help */
461};
462
463void __attribute__((constructor)) rt_constructor() {
464        register_format(&rt);
465}
Note: See TracBrowser for help on using the repository browser.