source: lib/format_rt.c @ 49babe0

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

Added initial format_rt and rt_protocol source files
Updated wag to be consistent with latest lt2 fixes and Dean's framing
Added a type field back into libtrace_packet
Fixed return value for trace_start

  • Property mode set to 100644
File size: 12.2 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
242static int rt_read_packet(struct libtrace_t *libtrace, struct libtrace_packet_t *packet) {
243       
244        int numbytes = 0;
245        char buf[RP_BUFSIZE];
246        int read_required = 0;
247        rt_header_t pkt_hdr;
248        uint16_t format;
249        char msg_buf[RP_BUFSIZE];
250
251        dag_record_t *erfptr;
252       
253        void *buffer = 0;
254
255        packet->trace = libtrace;
256
257        if (packet->buf_control == EXTERNAL) {
258                packet->buf_control = PACKET;
259                packet->buffer = malloc(LIBTRACE_PACKET_BUFSIZE);
260        }
261
262        buffer = packet->buffer;
263        packet->header = packet->buffer;
264
265
266        do {
267                if (tracefifo_out_available(libtrace->fifo) == 0 || read_required) {
268                        if ((numbytes = rt_read(
269                                        libtrace,buf,RP_BUFSIZE))<=0) {
270                                return numbytes;
271                        }
272                        tracefifo_write(libtrace->fifo,buf,numbytes);
273                        read_required = 0;
274                }
275                // Read rt header
276                if (tracefifo_out_read(libtrace->fifo,
277                                &pkt_hdr, sizeof(rt_header_t)) == 0) {
278                        read_required = 1;
279                        continue;
280                }
281                tracefifo_out_update(libtrace->fifo, sizeof(rt_header_t));
282               
283                packet->size = pkt_hdr.length;
284                packet->type = pkt_hdr.type;
285
286                switch (packet->type) {
287                        case RT_DATA:
288                                if (tracefifo_out_read(libtrace->fifo, 
289                                                        &format, 
290                                                        sizeof(uint16_t)) == 0) 
291                                {
292                                        tracefifo_out_reset(libtrace->fifo);
293                                        read_required = 1;
294                                        break;
295                                }
296                                if (tracefifo_out_read(libtrace->fifo, buffer, 
297                                                        packet->size - 
298                                                        sizeof(uint16_t)) == 0)
299                                {
300                                        tracefifo_out_reset(libtrace->fifo);
301                                        read_required = 1;
302                                        break;
303                                }
304                                // set packet->trace
305                                switch (format) {
306                                        case RT_FORMAT_ERF:
307                                                if (!RT_INFO->dummy_erf) {
308                                                        RT_INFO->dummy_erf = trace_create_dead("erf:-");
309                                                }
310                                                packet->trace = RT_INFO->dummy_erf;
311                                                break;
312                                        case RT_FORMAT_PCAP:
313                                                if (!RT_INFO->dummy_pcap) {
314                                                        RT_INFO->dummy_pcap = trace_create_dead("pcap:-");
315                                                }
316                                                packet->trace = RT_INFO->dummy_pcap;
317                                                break;
318                                        case RT_FORMAT_WAG:
319                                                if (!RT_INFO->dummy_wag) {
320                                                        RT_INFO->dummy_wag = trace_create_dead("wtf:-");
321                                                }
322                                                packet->trace = RT_INFO->dummy_wag;
323                                                break;
324                                        default:
325                                                printf("Unrecognised format: %d\n", format);
326                                                return -1;
327                                }
328                               
329                                // set packet->payload
330                                switch (format) {
331                                        case RT_FORMAT_ERF:
332                                                erfptr = (dag_record_t *)packet->header;
333                                               
334                                                if (erfptr->flags.rxerror == 1) {
335                                                        packet->payload = NULL;
336                                                } else {
337                                                        packet->payload = (char *)packet->buffer
338                                                                + trace_get_framing_length(packet);
339                                                }
340                                                break;
341                                        default:
342                                                packet->payload = (char *)packet->buffer +
343                                                        trace_get_framing_length(packet);
344                                                break;
345                                }
346                                // send ack
347                                break;
348                        case RT_STATUS:
349                                if (tracefifo_out_read(libtrace->fifo, buffer,
350                                                sizeof(rt_status_t)) == 0)
351                                {
352                                        tracefifo_out_reset(libtrace->fifo);
353                                        read_required = 1;
354                                        break;
355                                }
356                                break;
357                        case RT_DUCK:
358                                if (tracefifo_out_read(libtrace->fifo, buffer,
359                                                sizeof(rt_duck_t)) == 0)
360                                {
361                                        tracefifo_out_reset(libtrace->fifo);
362                                        read_required = 1;
363                                        break;
364                                }
365                                break;
366
367                        case RT_END_DATA:
368                                // need to do something sensible here
369                               
370                                break;
371
372                        case RT_PAUSE_ACK:
373                                // Check if we asked for a pause
374                               
375                               
376                                break;
377
378                        case RT_OPTION:
379                                // Server is requesting some option?
380
381                                break;
382
383                        default:
384                                printf("Bad rt type: %d\n", packet->type);
385                                return -1;
386                               
387                }
388                if (read_required)
389                        continue;
390                               
391               
392                // got in our whole packet, so...
393                tracefifo_out_update(libtrace->fifo,packet->size);
394
395                tracefifo_ack_update(libtrace->fifo,packet->size +
396                                sizeof(rt_header_t));
397                return 1;
398        } while(1);
399       
400}
401
402static int rt_get_fd(const struct libtrace_packet_t *packet) {
403        return packet->trace->format_data->input_fd;
404}
405
406
407
408static void rt_help() {
409        printf("rt format module\n");
410        printf("Supported input URIs:\n");
411        printf("\trt:hostname:port\n");
412        printf("\trt:hostname (connects on default port)\n");
413        printf("\n");
414        printf("\te.g.: rt:localhost\n");
415        printf("\te.g.: rt:localhost:32500\n");
416        printf("\n");
417
418}
419
420
421static struct libtrace_format_t rt = {
422        "rt",
423        "$Id$",
424        "rt",
425        rt_init_input,            /* init_input */
426        NULL,                           /* config_input */
427        rt_start_input,           /* start_input */
428        NULL,                           /* init_output */
429        NULL,                           /* config_output */
430        NULL,                           /* start_output */
431        rt_fin_input,             /* fin_input */
432        NULL,                           /* fin_output */
433        rt_read_packet,           /* read_packet */
434        NULL,                           /* write_packet */
435        NULL,                           /* get_link_type */
436        NULL,                           /* get_direction */
437        NULL,                           /* set_direction */
438        NULL,                           /* get_erf_timestamp */
439        NULL,                           /* get_timeval */
440        NULL,                           /* get_seconds */
441        NULL,                           /* get_capture_length */
442        NULL,                           /* get_wire_length */
443        NULL,                           /* get_framing_length */
444        NULL,                           /* set_capture_length */
445        rt_get_fd,                /* get_fd */
446        trace_event_device,             /* trace_event */
447        rt_help                   /* help */
448};
449
450void __attribute__((constructor)) rt_constructor() {
451        register_format(&rt);
452}
Note: See TracBrowser for help on using the repository browser.