Context Navigation

source: lib/format_rt.c @ e233141

4.0.1-hotfixescachetimestampsdevelopdpdk-ndagetsilivegetfragoffhelplibtrace4ndag_formatpfringrc-4.0.1rc-4.0.2rc-4.0.3rc-4.0.4ringdecrementfixringperformanceringtimestampfixes

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

Note: See TracBrowser for help on using the repository browser.

Download in other formats: