source: lib/format_rt.c @ 32f365f

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

Corrected length of close message

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