source: lib/format_rt.c @ eeab9832

4.0.1-hotfixescachetimestampsdevelopdpdk-ndagetsilivegetfragoffhelplibtrace4ndag_formatpfringrc-4.0.1rc-4.0.2rc-4.0.3rc-4.0.4ringdecrementfixringperformanceringtimestampfixes
Last change on this file since eeab9832 was eeab9832, checked in by Perry Lorier <perry@…>, 16 years ago

more error handling support
dag card work-around
removed annoying issue with mallocing memory during constructor that we

never free'd

added the ability to catch finalised packets
  • Property mode set to 100644
File size: 13.9 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 ((struct rt_format_data_t*)libtrace->format_data)
69
70int reliability = 0;
71
72char *rt_deny_reason(uint8_t reason) {
73        char *string = 0;
74
75        switch(reason) {
76                case RT_DENY_WRAPPER:
77                        string = "Rejected by TCP Wrappers";
78                        break;
79                case RT_DENY_FULL:
80                        string = "Max connections reached on server";
81                        break;
82                case RT_DENY_AUTH:
83                        string = "Authentication failed";
84                        break;
85                default:
86                        string = "Unknown reason";
87        }
88
89        return string;
90}
91
92
93struct rt_format_data_t {
94        char *hostname;
95        int port;
96        int input_fd;
97        int reliable;
98
99        struct libtrace_t *dummy_erf;
100        struct libtrace_t *dummy_pcap;
101        struct libtrace_t *dummy_wag;
102};
103
104static struct libtrace_format_t rt;
105
106static int rt_connect(struct libtrace_t *libtrace) {
107        struct hostent *he;
108        struct sockaddr_in remote;
109        rt_header_t connect_msg;
110        rt_deny_conn_t deny_hdr;       
111        rt_hello_t hello_opts;
112        uint8_t reason;
113       
114        if ((he=gethostbyname(RT_INFO->hostname)) == NULL) {
115                perror("gethostbyname");
116                return -1;
117        }
118        if ((RT_INFO->input_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
119                perror("socket");
120                return -1;
121        }
122
123        remote.sin_family = AF_INET;
124        remote.sin_port = htons(RT_INFO->port);
125        remote.sin_addr = *((struct in_addr *)he->h_addr);
126        bzero(&(remote.sin_zero), 8);
127
128        if (connect(RT_INFO->input_fd, (struct sockaddr *)&remote,
129                                sizeof(struct sockaddr)) == -1) {
130                perror("connect (inet)");
131                return -1;
132        }
133       
134        /* We are connected, now receive message from server */
135       
136        if (recv(RT_INFO->input_fd, &connect_msg, sizeof(rt_header_t), 0) != sizeof(rt_header_t) ) {
137                printf("An error occured while connecting to %s\n", RT_INFO->hostname);
138                return -1;
139        }
140
141        switch (connect_msg.type) {
142                case RT_DENY_CONN:
143                       
144                        if (recv(RT_INFO->input_fd, &deny_hdr, 
145                                                sizeof(rt_deny_conn_t),
146                                                0) != sizeof(rt_deny_conn_t)) {
147                                reason = 0;
148                        }       
149                        reason = deny_hdr.reason;
150                        printf("Connection attempt is denied by the server: %s\n",
151                                        rt_deny_reason(reason));
152                        return -1;
153                case RT_HELLO:
154                        /* do something with options */
155                        printf("Hello\n");     
156                        if (recv(RT_INFO->input_fd, &hello_opts, 
157                                                sizeof(rt_hello_t), 0)
158                                        != sizeof(rt_hello_t)) {
159                                printf("Failed to read hello options\n");
160                                return 0;
161                        }
162                        reliability = hello_opts.reliable;
163                       
164                        return 0;
165                default:
166                        printf("Unexpected message type: %d\n", connect_msg.type);
167                        return -1;
168        }
169       
170        return -1;
171}
172
173
174static int rt_init_input(struct libtrace_t *libtrace) {
175        char *scan;
176        char *uridata = libtrace->uridata;
177        libtrace->format_data = malloc(sizeof(struct rt_format_data_t));
178
179        RT_INFO->dummy_erf = NULL;
180        RT_INFO->dummy_pcap = NULL;
181        RT_INFO->dummy_wag = NULL;
182       
183        if (strlen(uridata) == 0) {
184                RT_INFO->hostname =
185                        strdup("localhost");
186                RT_INFO->port =
187                        COLLECTOR_PORT;
188        } else {
189                if ((scan = strchr(uridata,':')) == NULL) {
190                        RT_INFO->hostname =
191                                strdup(uridata);
192                        RT_INFO->port =
193                                COLLECTOR_PORT;
194                } else {
195                        RT_INFO->hostname =
196                                (char *)strndup(uridata,
197                                                (scan - uridata));
198                        RT_INFO->port =
199                                atoi(++scan);
200                }
201        }
202
203        return rt_connect(libtrace);
204}
205       
206static int rt_start_input(struct libtrace_t *libtrace) {
207        rt_header_t start_msg;
208
209        start_msg.type = RT_START;
210        start_msg.length = sizeof(rt_start_t);
211
212        printf("Sending start - len: %d\n", start_msg.length);
213       
214        /* Need to send start message to server */
215        if (send(RT_INFO->input_fd, &start_msg, sizeof(rt_header_t) +
216                                start_msg.length, 0) != sizeof(rt_header_t)) {
217                printf("Failed to send start message to server\n");
218                return -1;
219        }
220
221        return 0;
222}
223
224static int rt_fin_input(struct libtrace_t *libtrace) {
225        rt_header_t close_msg;
226
227        close_msg.type = RT_CLOSE;
228        close_msg.length = sizeof(rt_close_t);
229       
230        /* Send a close message to the server */
231        if (send(RT_INFO->input_fd, &close_msg, sizeof(rt_header_t) + 
232                                close_msg.length, 0) != sizeof(rt_header_t)) {
233                printf("Failed to send close message to server\n");
234       
235        }
236        if (RT_INFO->dummy_erf) 
237                trace_destroy_dead(RT_INFO->dummy_erf);
238               
239        if (RT_INFO->dummy_pcap)
240                trace_destroy_dead(RT_INFO->dummy_pcap);
241
242        if (RT_INFO->dummy_wag)
243                trace_destroy_dead(RT_INFO->dummy_wag);
244        close(RT_INFO->input_fd);
245        free(libtrace->format_data);
246        return 0;
247}
248
249static int rt_read(struct libtrace_t *libtrace, void *buffer, size_t len) {
250        int numbytes;
251
252        while(1) {
253#ifndef MSG_NOSIGNAL
254#  define MSG_NOSIGNAL 0
255#endif
256                if ((numbytes = recv(RT_INFO->input_fd,
257                                                buffer,
258                                                len,
259                                                MSG_NOSIGNAL)) == -1) {
260                        if (errno == EINTR) {
261                                /* ignore EINTR in case
262                                 * a caller is using signals
263                                 */
264                                continue;
265                        }
266                        perror("recv");
267                        return -1;
268                }
269                break;
270
271        }
272        return numbytes;
273}
274
275
276static int rt_set_format(libtrace_t *libtrace, libtrace_packet_t *packet) 
277{
278        switch (packet->type) {
279                case RT_DATA_ERF:
280                        if (!RT_INFO->dummy_erf) {
281                                RT_INFO->dummy_erf = trace_create_dead("erf:-");
282                        }
283                        packet->trace = RT_INFO->dummy_erf;
284                        break;
285                case RT_DATA_PCAP:
286                        if (!RT_INFO->dummy_pcap) {
287                                RT_INFO->dummy_pcap = trace_create_dead("pcap:-");
288                        }
289                        packet->trace = RT_INFO->dummy_pcap;
290                        break;
291                case RT_DATA_WAG:
292                        if (!RT_INFO->dummy_wag) {
293                                RT_INFO->dummy_wag = trace_create_dead("wtf:-");
294                        }
295                        packet->trace = RT_INFO->dummy_wag;
296                        break;
297                case RT_DATA_LEGACY_ETH:
298                case RT_DATA_LEGACY_ATM:
299                case RT_DATA_LEGACY_POS:
300                        printf("Sending legacy over RT is currently not supported\n");
301                        return -1;
302                default:
303                        printf("Unrecognised format: %d\n", packet->type);
304                        return -1;
305        }
306        return 0; /* success */
307}               
308
309static void rt_set_payload(struct libtrace_packet_t *packet) {
310        dag_record_t *erfptr;
311       
312        switch (packet->type) {
313                case RT_DATA_ERF:
314                        erfptr = (dag_record_t *)packet->header;
315                       
316                        if (erfptr->flags.rxerror == 1) {
317                                packet->payload = NULL;
318                                break;
319                        }
320                        /* else drop into the default case */
321                default:
322                        packet->payload = (char *)packet->buffer +
323                                trace_get_framing_length(packet);
324                        break;
325        }
326}
327
328static int rt_send_ack(struct libtrace_t *libtrace, 
329                uint32_t seqno)  {
330       
331        static char *ack_buffer = 0;
332        char *buf_ptr;
333        int numbytes = 0;
334        int to_write = 0;
335        rt_header_t *hdr;
336        rt_ack_t *ack_hdr;
337       
338        if (!ack_buffer) {
339                ack_buffer = malloc(sizeof(rt_header_t) + sizeof(rt_ack_t));
340        }
341       
342        hdr = (rt_header_t *) ack_buffer;
343        ack_hdr = (rt_ack_t *) (ack_buffer + sizeof(rt_header_t));
344       
345        hdr->type = RT_ACK;
346        hdr->length = sizeof(rt_ack_t);
347
348        ack_hdr->sequence = seqno;
349       
350        to_write = hdr->length + sizeof(rt_header_t);
351        buf_ptr = ack_buffer;
352
353       
354        while (to_write > 0) {
355                numbytes = send(RT_INFO->input_fd, buf_ptr, to_write, 0); 
356                if (numbytes == -1) {
357                        if (errno == EINTR || errno == EAGAIN) {
358                                continue;
359                        }
360                        else {
361                                printf("Error sending ack\n");
362                                return -1;
363                        }
364                }
365                to_write = to_write - numbytes;
366                buf_ptr = buf_ptr + to_write;
367               
368        }
369
370        return 1;
371}
372       
373static int rt_read_packet(libtrace_t *libtrace, libtrace_packet_t *packet) {
374       
375        rt_header_t pkt_hdr;
376        int pkt_size = 0;
377       
378        void *buffer = 0;
379
380        if (packet->buf_control == TRACE_CTRL_EXTERNAL || !packet->buffer) {
381                packet->buf_control = TRACE_CTRL_PACKET;
382                packet->buffer = malloc(LIBTRACE_PACKET_BUFSIZE);
383        } 
384
385        buffer = packet->buffer;
386        packet->header = packet->buffer;
387
388        /* FIXME: Better error handling required */
389        if (rt_read(libtrace, &pkt_hdr, sizeof(rt_header_t)) !=
390                        sizeof(rt_header_t)) {
391                printf("Error receiving rt header\n");
392                return -1;
393        }
394
395        packet->type = pkt_hdr.type;
396        pkt_size = pkt_hdr.length;
397
398        switch(packet->type) {
399                case RT_DATA_ERF:
400                case RT_DATA_PCAP:
401                case RT_DATA_WAG:
402                case RT_DATA_LEGACY_ETH:
403                case RT_DATA_LEGACY_POS:
404                case RT_DATA_LEGACY_ATM:
405                        if (rt_read(libtrace, buffer, pkt_size) != pkt_size) {
406                                printf("Error receiving packet\n");
407                                return -1;
408                        }
409                       
410                        if (rt_set_format(libtrace, packet) < 0) {
411                                return -1;
412                        }
413                        rt_set_payload(packet);
414
415                        if (reliability > 0) {
416                               
417                                if (rt_send_ack(libtrace, pkt_hdr.sequence) 
418                                                == -1)
419                                {
420                                        return -1;
421                                }
422                        }
423                        break;
424                case RT_STATUS:
425                case RT_DUCK:
426                        if (rt_read(libtrace, buffer, pkt_size) !=
427                                        pkt_size) {
428                                printf("Error receiving status packet\n");
429                                return -1;
430                        }
431                        packet->header = 0;
432                        packet->payload = buffer;
433                        break;
434                case RT_END_DATA:
435                        return 0;
436                case RT_PAUSE_ACK:
437                        /* FIXME: Do something useful */
438                        break;
439                case RT_OPTION:
440                        /* FIXME: Do something useful here as well */
441                        break;
442                default:
443                        printf("Bad rt type for client receipt: %d\n",
444                                        pkt_hdr.type);
445        }
446        return trace_get_capture_length(packet)+trace_get_framing_length(packet);
447}
448
449static int rt_get_capture_length(const struct libtrace_packet_t *packet) {
450        switch (packet->type) {
451                case RT_DUCK:
452                        return sizeof(rt_duck_t);
453                case RT_STATUS:
454                        return sizeof(rt_status_t);
455                case RT_HELLO:
456                        return sizeof(rt_hello_t);
457                case RT_START:
458                        return sizeof(rt_start_t);
459                case RT_ACK:
460                        return sizeof(rt_ack_t);
461                case RT_END_DATA:
462                        return sizeof(rt_end_data_t);
463                case RT_CLOSE:
464                        return sizeof(rt_close_t);
465                case RT_DENY_CONN:
466                        return sizeof(rt_deny_conn_t);
467                case RT_PAUSE:
468                        return sizeof(rt_pause_t);
469                case RT_PAUSE_ACK:
470                        return sizeof(rt_pause_ack_t);
471                case RT_OPTION:
472                        return sizeof(rt_option_t);
473        }
474        printf("Unknown type: %d\n", packet->type);
475        return 0;
476}
477                       
478static int rt_get_framing_length(const libtrace_packet_t *packet) {
479        return 0;
480}
481
482
483static int rt_get_fd(const libtrace_t *trace) {
484        return ((struct rt_format_data_t *)trace->format_data)->input_fd;
485}
486
487
488
489static void rt_help() {
490        printf("rt format module\n");
491        printf("Supported input URIs:\n");
492        printf("\trt:hostname:port\n");
493        printf("\trt:hostname (connects on default port)\n");
494        printf("\n");
495        printf("\te.g.: rt:localhost\n");
496        printf("\te.g.: rt:localhost:32500\n");
497        printf("\n");
498
499}
500
501
502static struct libtrace_format_t rt = {
503        "rt",
504        "$Id$",
505        TRACE_FORMAT_RT,
506        rt_init_input,                  /* init_input */
507        NULL,                           /* config_input */
508        rt_start_input,                 /* start_input */
509        NULL,                           /* init_output */
510        NULL,                           /* config_output */
511        NULL,                           /* start_output */
512        NULL,                           /* pause_output */
513        rt_fin_input,                   /* fin_input */
514        NULL,                           /* fin_output */
515        rt_read_packet,                 /* read_packet */
516        NULL,                           /* fin_packet */
517        NULL,                           /* write_packet */
518        NULL,                           /* get_link_type */
519        NULL,                           /* get_direction */
520        NULL,                           /* set_direction */
521        NULL,                           /* get_erf_timestamp */
522        NULL,                           /* get_timeval */
523        NULL,                           /* get_seconds */
524        NULL,                           /* seek_erf */
525        NULL,                           /* seek_timeval */
526        NULL,                           /* seek_seconds */
527        rt_get_capture_length,          /* get_capture_length */
528        NULL,                           /* get_wire_length */
529        rt_get_framing_length,          /* get_framing_length */
530        NULL,                           /* set_capture_length */
531        rt_get_fd,                      /* get_fd */
532        trace_event_device,             /* trace_event */
533        rt_help,                        /* help */
534        NULL                            /* next pointer */
535};
536
537void __attribute__((constructor)) rt_constructor() {
538        register_format(&rt);
539}
Note: See TracBrowser for help on using the repository browser.