source: lib/format_rt.c @ 91db67b

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

Clean up the various usages of TRACE_ERR_*, add TRACE_ERR_UNSUPPORTED and
shuffle which error was returned where. Document their new usages.

Closes: #13

Also, fix the 802.1q definitions.
  • Property mode set to 100644
File size: 16.7 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#include <sys/stat.h>
43#include <assert.h>
44#include <errno.h>
45#include <fcntl.h>
46#include <stdio.h>
47#include <string.h>
48#include <stdlib.h>
49
50#ifndef WIN32
51# include <netdb.h>
52#endif
53
54#define RT_INFO ((struct rt_format_data_t*)libtrace->format_data)
55
56int reliability = 0;
57
58char *rt_deny_reason(uint8_t reason) {
59        char *string = 0;
60
61        switch(reason) {
62                case RT_DENY_WRAPPER:
63                        string = "Rejected by TCP Wrappers";
64                        break;
65                case RT_DENY_FULL:
66                        string = "Max connections reached on server";
67                        break;
68                case RT_DENY_AUTH:
69                        string = "Authentication failed";
70                        break;
71                default:
72                        string = "Unknown reason";
73        }
74
75        return string;
76}
77
78
79struct rt_format_data_t {
80        char *hostname;
81        int port;
82        int input_fd;
83        int reliable;
84        char *pkt_buffer;
85        char *buf_current;
86        int buf_left;
87
88       
89        struct libtrace_t *dummy_erf;
90        struct libtrace_t *dummy_pcap;
91        struct libtrace_t *dummy_wag;
92};
93
94static struct libtrace_format_t rt;
95
96static int rt_connect(struct libtrace_t *libtrace) {
97        struct hostent *he;
98        struct sockaddr_in remote;
99        rt_header_t connect_msg;
100        rt_deny_conn_t deny_hdr;       
101        rt_hello_t hello_opts;
102        uint8_t reason;
103       
104        if ((he=gethostbyname(RT_INFO->hostname)) == NULL) {
105                perror("gethostbyname");
106                return -1;
107        }
108        if ((RT_INFO->input_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
109                perror("socket");
110                return -1;
111        }
112
113        remote.sin_family = AF_INET;
114        remote.sin_port = htons(RT_INFO->port);
115        remote.sin_addr = *((struct in_addr *)he->h_addr);
116        memset(&(remote.sin_zero), 0, 8);
117
118        if (connect(RT_INFO->input_fd, (struct sockaddr *)&remote,
119                                sizeof(struct sockaddr)) == -1) {
120                trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
121                                "Could not connect to host %s on port %d",
122                                RT_INFO->hostname, RT_INFO->port);
123                return -1;
124        }
125
126       
127#if 0
128        oldflags = fcntl(RT_INFO->input_fd, F_GETFL, 0);
129        if (oldflags == -1) {
130                trace_set_err(libtrace, errno,
131                                "Could not get fd flags from fd %d\n",
132                                RT_INFO->input_fd);
133                return -1;
134        }
135        oldflags |= O_NONBLOCK;
136        if (fcntl(RT_INFO->input_fd, F_SETFL, oldflags) == -1) {
137                trace_set_err(libtrace, errno,
138                                "Could not set fd flags for fd %d\n",
139                                RT_INFO->input_fd);
140                return -1;
141        }
142#endif
143       
144       
145        /* We are connected, now receive message from server */
146       
147        if (recv(RT_INFO->input_fd, (void*)&connect_msg, sizeof(rt_header_t), 0) != sizeof(rt_header_t) ) {
148                trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
149                                "Could not receive connection message from %s",
150                                RT_INFO->hostname);
151                return -1;
152        }
153       
154        switch (connect_msg.type) {
155                case RT_DENY_CONN:
156                       
157                        if (recv(RT_INFO->input_fd, (void*)&deny_hdr, 
158                                                sizeof(rt_deny_conn_t),
159                                                0) != sizeof(rt_deny_conn_t)) {
160                                reason = 0;
161                        }       
162                        reason = deny_hdr.reason;
163                        trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
164                                "Connection attempt is denied: %s",
165                                rt_deny_reason(reason));       
166                        return -1;
167                case RT_HELLO:
168                        /* do something with options */
169                        if (recv(RT_INFO->input_fd, (void*)&hello_opts, 
170                                                sizeof(rt_hello_t), 0)
171                                        != sizeof(rt_hello_t)) {
172                                trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
173                                        "Failed to receive RT_HELLO options");
174                                return -1;
175                        }
176                        reliability = hello_opts.reliable;
177                       
178                        return 0;
179                default:
180                        trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
181                                        "Unknown message type received: %d",
182                                        connect_msg.type);
183                        return -1;
184        }
185        trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
186                        "Somehow you managed to reach this unreachable code");
187        return -1;
188}
189
190
191static int rt_init_input(struct libtrace_t *libtrace) {
192        char *scan;
193        char *uridata = libtrace->uridata;
194        libtrace->format_data = malloc(sizeof(struct rt_format_data_t));
195
196        RT_INFO->dummy_erf = NULL;
197        RT_INFO->dummy_pcap = NULL;
198        RT_INFO->dummy_wag = NULL;
199        RT_INFO->pkt_buffer = NULL;
200        RT_INFO->buf_current = NULL;
201        RT_INFO->buf_left = 0;
202       
203        if (strlen(uridata) == 0) {
204                RT_INFO->hostname =
205                        strdup("localhost");
206                RT_INFO->port =
207                        COLLECTOR_PORT;
208        } else {
209                if ((scan = strchr(uridata,':')) == NULL) {
210                        RT_INFO->hostname =
211                                strdup(uridata);
212                        RT_INFO->port =
213                                COLLECTOR_PORT;
214                } else {
215                        RT_INFO->hostname =
216                                (char *)strndup(uridata,
217                                                (scan - uridata));
218                        RT_INFO->port =
219                                atoi(++scan);
220                }
221        }
222
223        return rt_connect(libtrace);
224}
225       
226static int rt_start_input(struct libtrace_t *libtrace) {
227        rt_header_t start_msg;
228
229        start_msg.type = RT_START;
230        start_msg.length = 0; 
231
232       
233        /* Need to send start message to server */
234        if (send(RT_INFO->input_fd, (void*)&start_msg, sizeof(rt_header_t) +
235                                start_msg.length, 0) != sizeof(rt_header_t)) {
236                printf("Failed to send start message to server\n");
237                return -1;
238        }
239
240        return 0;
241}
242
243static int rt_fin_input(struct libtrace_t *libtrace) {
244        rt_header_t close_msg;
245
246        close_msg.type = RT_CLOSE;
247        close_msg.length = 0; 
248       
249        /* Send a close message to the server */
250        if (send(RT_INFO->input_fd, (void*)&close_msg, sizeof(rt_header_t) + 
251                                close_msg.length, 0) != sizeof(rt_header_t)
252                                + close_msg.length) {
253                printf("Failed to send close message to server\n");
254       
255        }
256        if (RT_INFO->dummy_erf) 
257                trace_destroy_dead(RT_INFO->dummy_erf);
258               
259        if (RT_INFO->dummy_pcap)
260                trace_destroy_dead(RT_INFO->dummy_pcap);
261
262        if (RT_INFO->dummy_wag)
263                trace_destroy_dead(RT_INFO->dummy_wag);
264        close(RT_INFO->input_fd);
265        free(libtrace->format_data);
266        return 0;
267}
268
269#define RT_BUF_SIZE 4000
270
271static int rt_read(struct libtrace_t *libtrace, void **buffer, size_t len, int block) {
272        int numbytes;
273
274        assert(len <= RT_BUF_SIZE);
275       
276        if (!RT_INFO->pkt_buffer) {
277                RT_INFO->pkt_buffer = malloc(RT_BUF_SIZE);
278                RT_INFO->buf_current = RT_INFO->pkt_buffer;
279                RT_INFO->buf_left = 0;
280        }
281
282#ifndef MSG_DONTWAIT
283#define MSG_DONTWAIT 0
284#endif
285
286        if (block)
287                block=0;
288        else
289                block=MSG_DONTWAIT;
290
291       
292        if (len > RT_INFO->buf_left) {
293                memcpy(RT_INFO->pkt_buffer, RT_INFO->buf_current, 
294                                RT_INFO->buf_left);
295                RT_INFO->buf_current = RT_INFO->pkt_buffer;
296
297#ifndef MSG_NOSIGNAL
298#  define MSG_NOSIGNAL 0
299#endif
300                while (len > RT_INFO->buf_left) {
301                        if ((numbytes = recv(RT_INFO->input_fd,
302                                                RT_INFO->pkt_buffer + 
303                                                RT_INFO->buf_left,
304                                                RT_BUF_SIZE-RT_INFO->buf_left,
305                                                MSG_NOSIGNAL|block)) <= 0) {
306                                if (numbytes == 0) {
307                                        trace_set_err(libtrace, TRACE_ERR_BAD_PACKET, 
308                                                        "No data received");
309                                        return -1;
310                                }
311                               
312                                if (errno == EINTR) {
313                                        /* ignore EINTR in case
314                                         * a caller is using signals
315                                         */
316                                        continue;
317                                }
318                                if (errno == EAGAIN) {
319                                        trace_set_err(libtrace,
320                                                        EAGAIN,
321                                                        "EAGAIN");
322                                        return -1;
323                                }
324                               
325                                perror("recv");
326                                trace_set_err(libtrace, errno,
327                                                "Failed to read data into rt recv buffer");
328                                return -1;
329                        }
330                        /*
331                        buf_ptr = RT_INFO->pkt_buffer;
332                        for (i = 0; i < RT_BUF_SIZE ; i++) {
333                                       
334                                printf("%02x", (unsigned char)*buf_ptr);
335                                buf_ptr ++;
336                        }
337                        printf("\n");
338                        */
339                        RT_INFO->buf_left+=numbytes;
340                }
341
342        }
343        *buffer = RT_INFO->buf_current;
344        RT_INFO->buf_current += len;
345        RT_INFO->buf_left -= len;
346        assert(RT_INFO->buf_left >= 0);
347        return len;
348}
349
350
351static int rt_set_format(libtrace_t *libtrace, libtrace_packet_t *packet) 
352{
353       
354        if (packet->type >= RT_DATA_PCAP) {
355                if (!RT_INFO->dummy_pcap) {
356                        RT_INFO->dummy_pcap = trace_create_dead("pcap:-");
357                }
358                packet->trace = RT_INFO->dummy_pcap;
359                return 0;       
360        }
361
362        switch (packet->type) {
363                case RT_DATA_ERF:
364                        if (!RT_INFO->dummy_erf) {
365                                RT_INFO->dummy_erf = trace_create_dead("erf:-");
366                        }
367                        packet->trace = RT_INFO->dummy_erf;
368                        break;
369                case RT_DATA_WAG:
370                        if (!RT_INFO->dummy_wag) {
371                                RT_INFO->dummy_wag = trace_create_dead("wtf:-");
372                        }
373                        packet->trace = RT_INFO->dummy_wag;
374                        break;
375                case RT_DATA_LEGACY_ETH:
376                case RT_DATA_LEGACY_ATM:
377                case RT_DATA_LEGACY_POS:
378                        printf("Sending legacy over RT is currently not supported\n");
379                        trace_set_err(libtrace, TRACE_ERR_BAD_PACKET, "Legacy packet cannot be sent over rt");
380                        return -1;
381                default:
382                        printf("Unrecognised format: %d\n", packet->type);
383                        trace_set_err(libtrace, TRACE_ERR_BAD_PACKET, "Unrecognised packet format");
384                        return -1;
385        }
386        return 0; /* success */
387}               
388
389static void rt_set_payload(struct libtrace_packet_t *packet) {
390        dag_record_t *erfptr;
391       
392        switch (packet->type) {
393                case RT_DATA_ERF:
394                        erfptr = (dag_record_t *)packet->header;
395                       
396                        if (erfptr->flags.rxerror == 1) {
397                                packet->payload = NULL;
398                                break;
399                        }
400                        /* else drop into the default case */
401                default:
402                        packet->payload = (char *)packet->buffer +
403                                trace_get_framing_length(packet);
404                        break;
405        }
406}
407
408static int rt_send_ack(struct libtrace_t *libtrace, 
409                uint32_t seqno)  {
410       
411        static char *ack_buffer = 0;
412        char *buf_ptr;
413        int numbytes = 0;
414        int to_write = 0;
415        rt_header_t *hdr;
416        rt_ack_t *ack_hdr;
417       
418        if (!ack_buffer) {
419                ack_buffer = malloc(sizeof(rt_header_t) + sizeof(rt_ack_t));
420        }
421       
422        hdr = (rt_header_t *) ack_buffer;
423        ack_hdr = (rt_ack_t *) (ack_buffer + sizeof(rt_header_t));
424       
425        hdr->type = RT_ACK;
426        hdr->length = sizeof(rt_ack_t);
427
428        ack_hdr->sequence = seqno;
429       
430        to_write = hdr->length + sizeof(rt_header_t);
431        buf_ptr = ack_buffer;
432
433       
434        while (to_write > 0) {
435                numbytes = send(RT_INFO->input_fd, buf_ptr, to_write, 0); 
436                if (numbytes == -1) {
437                        if (errno == EINTR || errno == EAGAIN) {
438                                continue;
439                        }
440                        else {
441                                printf("Error sending ack\n");
442                                trace_set_err(libtrace, TRACE_ERR_BAD_PACKET, 
443                                                "Error sending ack");
444                                return -1;
445                        }
446                }
447                to_write = to_write - numbytes;
448                buf_ptr = buf_ptr + to_write;
449               
450        }
451
452        return 1;
453}
454
455       
456static int rt_read_packet_versatile(libtrace_t *libtrace,
457                libtrace_packet_t *packet,int blocking) {
458        rt_header_t rt_hdr;
459        rt_header_t *pkt_hdr = &rt_hdr;
460        int pkt_size = 0;
461       
462       
463        if (packet->buf_control == TRACE_CTRL_EXTERNAL || !packet->buffer) {
464                packet->buf_control = TRACE_CTRL_PACKET;
465                packet->buffer = malloc(LIBTRACE_PACKET_BUFSIZE);
466        } 
467
468
469        /* FIXME: Better error handling required */
470        if (rt_read(libtrace, (void **)&pkt_hdr, sizeof(rt_header_t),blocking) !=
471                        sizeof(rt_header_t)) {
472                return -1;
473        }
474
475        packet->type = pkt_hdr->type;
476        pkt_size = pkt_hdr->length;
477        packet->size = pkt_hdr->length;
478
479        if (packet->type >= RT_DATA_SIMPLE) {
480                if (rt_read(libtrace, &packet->buffer, pkt_size,1) != pkt_size) {
481                        printf("Error receiving packet\n");
482                        return -1;
483                }
484                packet->header = packet->buffer;
485               
486                if (rt_set_format(libtrace, packet) < 0) {
487                        return -1;
488                }
489                rt_set_payload(packet);
490
491                if (reliability > 0) {
492                       
493                        if (rt_send_ack(libtrace, pkt_hdr->sequence) 
494                                        == -1)
495                        {
496                                return -1;
497                        }
498                }
499        } else {
500                switch(packet->type) {
501                        case RT_STATUS:
502                        case RT_DUCK:
503                                if (rt_read(libtrace, &packet->buffer, 
504                                                        pkt_size,1) !=
505                                                pkt_size) {
506                                        printf("Error receiving status packet\n");
507                                        return -1;
508                                }
509                                packet->header = 0;
510                                packet->payload = packet->buffer;
511                                break;
512                        case RT_END_DATA:
513                                return 0;
514                        case RT_PAUSE_ACK:
515                                /* FIXME: Do something useful */
516                                break;
517                        case RT_OPTION:
518                                /* FIXME: Do something useful here as well */
519                                break;
520                        case RT_KEYCHANGE:
521                                break;
522                        default:
523                                printf("Bad rt type for client receipt: %d\n",
524                                        pkt_hdr->type);
525                }
526        }
527        /* Return the number of bytes read from the stream */
528        return packet->size; 
529}
530
531static int rt_read_packet(libtrace_t *libtrace,
532                libtrace_packet_t *packet) {
533        return rt_read_packet_versatile(libtrace,packet,1);
534}
535
536
537static int rt_get_capture_length(const struct libtrace_packet_t *packet) {
538        switch (packet->type) {
539                case RT_DUCK:
540                        return 0; /* FIXME */
541                case RT_STATUS:
542                        return sizeof(rt_status_t);
543                case RT_HELLO:
544                        return sizeof(rt_hello_t);
545                case RT_START:
546                        return 0;
547                case RT_ACK:
548                        return sizeof(rt_ack_t);
549                case RT_END_DATA:
550                        return 0;
551                case RT_CLOSE:
552                        return 0;
553                case RT_DENY_CONN:
554                        return sizeof(rt_deny_conn_t);
555                case RT_PAUSE:
556                        return 0; 
557                case RT_PAUSE_ACK:
558                        return 0;
559                case RT_OPTION:
560                        return 0; /* FIXME */
561                case RT_KEYCHANGE:
562                        return 0;
563        }
564        printf("Unknown type: %d\n", packet->type);
565        return 0;
566}
567
568static int rt_get_wire_length(const libtrace_packet_t *packet) {
569        return 0;
570}
571                       
572static int rt_get_framing_length(const libtrace_packet_t *packet) {
573        return 0;
574}
575
576static int rt_get_fd(const libtrace_t *trace) {
577        return ((struct rt_format_data_t *)trace->format_data)->input_fd;
578}
579
580struct libtrace_eventobj_t trace_event_rt(struct libtrace_t *trace, struct libtrace_packet_t *packet) {
581        struct libtrace_eventobj_t event = {0,0,0.0,0};
582        libtrace_err_t read_err;
583
584        assert(trace);
585        assert(packet);
586       
587        if (trace->format->get_fd) {
588                event.fd = trace->format->get_fd(trace);
589        } else {
590                event.fd = 0;
591        }
592
593        event.size = rt_read_packet_versatile(trace, packet, 0);
594        if (event.size == -1) {
595                read_err = trace_get_err(trace);
596                if (read_err.err_num == EAGAIN) {
597                        event.type = TRACE_EVENT_IOWAIT;
598                }
599                else {
600                        printf("packet error\n");
601                        event.type = TRACE_EVENT_PACKET;
602                }
603        } else if (event.size == 0) {
604                event.type = TRACE_EVENT_TERMINATE;
605               
606        }       
607        else {
608                event.type = TRACE_EVENT_PACKET;
609        }
610       
611        return event;
612}
613
614static void rt_help() {
615        printf("rt format module\n");
616        printf("Supported input URIs:\n");
617        printf("\trt:hostname:port\n");
618        printf("\trt:hostname (connects on default port)\n");
619        printf("\n");
620        printf("\te.g.: rt:localhost\n");
621        printf("\te.g.: rt:localhost:32500\n");
622        printf("\n");
623
624}
625
626
627static struct libtrace_format_t rt = {
628        "rt",
629        "$Id$",
630        TRACE_FORMAT_RT,
631        rt_init_input,                  /* init_input */
632        NULL,                           /* config_input */
633        rt_start_input,                 /* start_input */
634        NULL,                           /* init_output */
635        NULL,                           /* config_output */
636        NULL,                           /* start_output */
637        NULL,                           /* pause_output */
638        rt_fin_input,                   /* fin_input */
639        NULL,                           /* fin_output */
640        rt_read_packet,                 /* read_packet */
641        NULL,                           /* fin_packet */
642        NULL,                           /* write_packet */
643        NULL,                           /* get_link_type */
644        NULL,                           /* get_direction */
645        NULL,                           /* set_direction */
646        NULL,                           /* get_erf_timestamp */
647        NULL,                           /* get_timeval */
648        NULL,                           /* get_seconds */
649        NULL,                           /* seek_erf */
650        NULL,                           /* seek_timeval */
651        NULL,                           /* seek_seconds */
652        rt_get_capture_length,          /* get_capture_length */
653        rt_get_wire_length,                     /* get_wire_length */
654        rt_get_framing_length,          /* get_framing_length */
655        NULL,                           /* set_capture_length */
656        rt_get_fd,                      /* get_fd */
657        trace_event_rt,             /* trace_event */
658        rt_help,                        /* help */
659        NULL                            /* next pointer */
660};
661
662void CONSTRUCTOR rt_constructor() {
663        register_format(&rt);
664}
Note: See TracBrowser for help on using the repository browser.