Merge remote-tracking branch 'origin/fastpath'

* origin/fastpath:
  and remove superflous print.
  add testcase for subrecords and events add missing binary testcase (Baseline is in master, testcase is missing for some reason) make error output for nonmatching event types much more verbose
  Add more error handling for close() calls.
  add testcase for subrecords to input framework tests

CHANGES

@@ -1,4 +1,13 @@
 
+2.0-884 | 2012-07-26 14:33:21 -0700
+
+  * Add comprehensive error handling for close() calls. (Jon Siwek)
+    
+  * Add more test cases for input framework. (Bernhard Amann)
+
+  * Input framework: make error output for non-matching event types
+    much more verbose. (Bernhard Amann)
+
 2.0-877 | 2012-07-25 17:20:34 -0700
 
   * Fix double close() in FilerSerializer class. (Jon Siwek)

VERSION

@@ -1 +1 @@
-2.0-877
+2.0-884

src/ChunkedIO.cc

@@ -76,7 +76,7 @@ void ChunkedIO::DumpDebugData(const char* basefnname, bool want_reads)
 		ChunkedIOFd io(fd, "dump-file");
 		io.Write(*i);
 		io.Flush();
-		close(fd);
+		safe_close(fd);
 		}
 
 	l->clear();
@@ -127,7 +127,7 @@ ChunkedIOFd::~ChunkedIOFd()
 
 	delete [] read_buffer;
 	delete [] write_buffer;
-	close(fd);
+	safe_close(fd);
 
 	if ( partial )
 		{
@@ -686,7 +686,7 @@ ChunkedIOSSL::~ChunkedIOSSL()
 		ssl = 0;
 		}
 
-	close(socket);
+	safe_close(socket);
 	}
 
 

src/FlowSrc.cc

@@ -58,7 +58,7 @@ void FlowSrc::Process()
 
 void FlowSrc::Close()
 	{
-	close(selectable_fd);
+	safe_close(selectable_fd);
 	}
 
 

src/RemoteSerializer.cc

@@ -647,7 +647,7 @@ void RemoteSerializer::Fork()
 			exit(1); // FIXME: Better way to handle this?
 			}
 
-		close(pipe[1]);
+		safe_close(pipe[1]);
 
 		return;
 		}
@@ -664,12 +664,12 @@ void RemoteSerializer::Fork()
 			}
 
 		child.SetParentIO(io);
-		close(pipe[0]);
+		safe_close(pipe[0]);
 
 		// Close file descriptors.
-		close(0);
+		safe_close(0);
-		close(1);
+		safe_close(1);
-		close(2);
+		safe_close(2);
 
 		// Be nice.
 		setpriority(PRIO_PROCESS, 0, 5);
@@ -4001,7 +4001,7 @@ bool SocketComm::Connect(Peer* peer)
 		if ( connect(sockfd, res->ai_addr, res->ai_addrlen) < 0 )
 			{
 			Error(fmt("connect failed: %s", strerror(errno)), peer);
-			close(sockfd);
+			safe_close(sockfd);
 			sockfd = -1;
 			continue;
 			}
@@ -4174,16 +4174,18 @@ bool SocketComm::Listen()
 			{
 			Error(fmt("can't bind to %s:%s, %s", l_addr_str.c_str(),
 			          port_str, strerror(errno)));
-			close(fd);
 
 			if ( errno == EADDRINUSE )
 				{
 				// Abandon completely this attempt to set up listening sockets,
 				// try again later.
+				safe_close(fd);
 				CloseListenFDs();
 				listen_next_try = time(0) + bind_retry_interval;
 				return false;
 				}
+
+			safe_close(fd);
 			continue;
 			}
 
@@ -4191,7 +4193,7 @@ bool SocketComm::Listen()
 			{
 			Error(fmt("can't listen on %s:%s, %s", l_addr_str.c_str(),
 			          port_str, strerror(errno)));
-			close(fd);
+			safe_close(fd);
 			continue;
 			}
 
@@ -4227,7 +4229,7 @@ bool SocketComm::AcceptConnection(int fd)
 		{
 		Error(fmt("accept fail, unknown address family %d",
 		          client.ss.ss_family));
-		close(clientfd);
+		safe_close(clientfd);
 		return false;
 		}
 
@@ -4298,7 +4300,7 @@ const char* SocketComm::MakeLogString(const char* msg, Peer* peer)
 void SocketComm::CloseListenFDs()
 	{
 	for ( size_t i = 0; i < listen_fds.size(); ++i )
-		close(listen_fds[i]);
+		safe_close(listen_fds[i]);
 
 	listen_fds.clear();
 	}

src/Serializer.cc

@@ -746,7 +746,7 @@ FileSerializer::~FileSerializer()
 	if ( io )
 		delete io;  // destructor will call close() on fd
 	else if ( fd >= 0 )
-		close(fd);
+		safe_close(fd);
 	}
 
 bool FileSerializer::Open(const char* file, bool pure)
@@ -810,7 +810,7 @@ void FileSerializer::CloseFile()
 		io->Flush();
 
 	if ( fd >= 0 && ! io ) // destructor of io calls close() on fd
-		close(fd);
+		safe_close(fd);
 	fd = -1;
 
 	delete [] file;

src/input/Manager.cc

@@ -441,9 +441,15 @@ bool Manager::CreateEventStream(RecordVal* fval)
 			return false;
 			}
 
-		if ( !same_type((*args)[2], fields ) )
+		if ( ! same_type((*args)[2], fields ) )
 			{
-			reporter->Error("Incompatible type for event");
+			ODesc desc1;
+			ODesc desc2;
+			(*args)[2]->Describe(&desc1);
+			fields->Describe(&desc2);
+			reporter->Error("Incompatible type '%s':%s for event, which needs type '%s':%s\n",
+					type_name((*args)[2]->Tag()), desc1.Description(),
+					type_name(fields->Tag()), desc2.Description());
 			return false;
 			}
 

src/logging/writers/Ascii.cc

@@ -86,7 +86,7 @@ void Ascii::CloseFile(double t)
 		WriteHeaderField("end", ts);
 		}
 
-	close(fd);
+	safe_close(fd);
 	fd = 0;
 	}
 

src/util.cc

@@ -722,7 +722,7 @@ void init_random_seed(uint32 seed, const char* read_file, const char* write_file
 			{
 			int amt = read(fd, buf + pos,
 					sizeof(uint32) * (bufsiz - pos));
-			close(fd);
+			safe_close(fd);
 
 			if ( amt > 0 )
 				pos += amt / sizeof(uint32);
@@ -1204,7 +1204,7 @@ void _set_processing_status(const char* status)
 		len -= n;
 		}
 
-	close(fd);
+	safe_close(fd);
 
 	errno = old_errno;
 	}
@@ -1353,6 +1353,31 @@ bool safe_write(int fd, const char* data, int len)
 	return true;
 	}
 
+void safe_close(int fd)
+	{
+	/*
+	 * Failure cases of close(2) are ...
+	 * EBADF: Indicative of programming logic error that needs to be fixed, we
+	 *        should always be attempting to close a valid file descriptor.
+	 * EINTR: Ignore signal interruptions, most implementations will actually
+	 *        reclaim the open descriptor and POSIX standard doesn't leave many
+	 *        options by declaring the state of the descriptor as "unspecified".
+	 *        Attempting to inspect actual state or re-attempt close() is not
+	 *        thread safe.
+	 * EIO:   Again the state of descriptor is "unspecified", but don't recover
+	 *        from an I/O error, safe_write() won't either.
+	 *
+	 * Note that we don't use the reporter here to allow use from different threads.
+	 */
+	if ( close(fd) < 0 && errno != EINTR )
+		{
+		char buf[128];
+		strerror_r(errno, buf, sizeof(buf));
+		fprintf(stderr, "safe_close error %d: %s\n", errno, buf);
+		abort();
+		}
+	}
+
 void out_of_memory(const char* where)
 	{
 	reporter->FatalError("out of memory in %s.\n", where);

src/util.h

@@ -297,6 +297,9 @@ inline size_t pad_size(size_t size)
 // thread-safe as long as no two threads write to the same descriptor.
 extern bool safe_write(int fd, const char* data, int len);
 
+// Wraps close(2) to emit error messages and abort on unrecoverable errors.
+extern void safe_close(int fd);
+
 extern void out_of_memory(const char* where);
 
 inline void* safe_realloc(void* ptr, size_t size)

testing/btest/Baseline/scripts.base.frameworks.input.subrecord-event/out

@@ -0,0 +1,12 @@
+[sub=[b=T, e=SSH::LOG, c=21, p=123/unknown, sn=10.0.0.0/24, two=[a=1.2.3.4, d=3.14]], t=1315801931.273616, iv=100.0, s=hurz, sc={
+2,
+4,
+1,
+3
+}, ss={
+CC,
+AA,
+BB
+}, se={
+
+}, vc=[10, 20, 30], ve=[]]

testing/btest/Baseline/scripts.base.frameworks.input.subrecord/out

@@ -0,0 +1,14 @@
+{
+[-42] = [sub=[b=T, e=SSH::LOG, c=21, p=123/unknown, sn=10.0.0.0/24, two=[a=1.2.3.4, d=3.14]], t=1315801931.273616, iv=100.0, s=hurz, sc={
+2,
+4,
+1,
+3
+}, ss={
+CC,
+AA,
+BB
+}, se={
+
+}, vc=[10, 20, 30], ve=[]]
+}

testing/btest/scripts/base/frameworks/input/binary.bro

@@ -0,0 +1,56 @@
+# (uses listen.bro just to ensure input sources are more reliably fully-read).
+# @TEST-SERIALIZE: comm
+#
+# @TEST-EXEC: btest-bg-run bro bro -b %INPUT
+# @TEST-EXEC: btest-bg-wait -k 5
+# @TEST-EXEC: btest-diff out
+
+redef InputAscii::separator = "|";
+redef InputAscii::set_separator = ",";
+redef InputAscii::empty_field = "(empty)";
+redef InputAscii::unset_field = "-";
+
+@TEST-START-FILE input.log
+#separator |
+#set_separator|,
+#empty_field|(empty)
+#unset_field|-
+#path|ssh
+#start|2012-07-20-01-49-19
+#fields|data|data2
+#types|string|string
+abc\x0a\xffdef|DATA2
+abc\x7c\xffdef|DATA2
+abc\xff\x7cdef|DATA2
+#end|2012-07-20-01-49-19
+@TEST-END-FILE
+
+@load frameworks/communication/listen
+
+global outfile: file;
+global try: count;
+
+type Val: record {
+		data: string;
+		data2: string;
+};
+
+event line(description: Input::EventDescription, tpe: Input::Event, a: string, b: string)
+	{
+	print outfile, a;
+	print outfile, b;
+	try = try + 1;
+	if ( try == 3 )
+		{
+		close(outfile);
+		terminate();
+		}
+	}
+
+event bro_init()
+	{
+	try = 0;
+	outfile = open("../out");
+	Input::add_event([$source="../input.log", $name="input", $fields=Val, $ev=line]);
+	Input::remove("input");
+	}

testing/btest/scripts/base/frameworks/input/subrecord-event.bro

@@ -0,0 +1,75 @@
+# (uses listen.bro just to ensure input sources are more reliably fully-read).
+# @TEST-SERIALIZE: comm
+#
+# @TEST-EXEC: btest-bg-run bro bro -b %INPUT
+# @TEST-EXEC: btest-bg-wait -k 5
+# @TEST-EXEC: btest-diff out
+
+@TEST-START-FILE input.log
+#separator \x09
+#path	ssh
+#fields	sub.b	i	sub.e	sub.c	sub.p	sub.sn	sub.two.a	sub.two.d	t	iv	s	sc	ss	se	vc	ve	f
+#types	bool	int	enum	count	port	subnet	addr	double	time	interval	string	table	table	table	vector	vector	func
+T	-42	SSH::LOG	21	123	10.0.0.0/24	1.2.3.4	3.14	1315801931.273616	100.000000	hurz	2,4,1,3	CC,AA,BB	EMPTY	10,20,30	EMPTY	SSH::foo\x0a{ \x0aif (0 < SSH::i) \x0a\x09return (Foo);\x0aelse\x0a\x09return (Bar);\x0a\x0a}
+@TEST-END-FILE
+
+@load base/protocols/ssh
+@load frameworks/communication/listen
+
+global outfile: file;
+global try: count;
+
+redef InputAscii::empty_field = "EMPTY";
+
+module A;
+
+type Idx: record {
+	i: int;
+};
+
+type SubVal2: record {
+	a: addr;
+	d: double;
+};
+
+type SubVal: record {
+	b: bool;
+	e: Log::ID;
+	c: count;
+	p: port;
+	sn: subnet;
+	two: SubVal2;
+};
+
+type Val: record {
+	sub: SubVal;
+	t: time;
+	iv: interval;
+	s: string;
+	sc: set[count];
+	ss: set[string];
+	se: set[string];
+	vc: vector of int;
+	ve: vector of int;
+};
+
+
+
+event line(description: Input::EventDescription, tpe: Input::Event, value: Val)
+	{
+	print outfile, value;
+	try = try + 1;
+	if ( try == 1 )
+		{
+		close(outfile);
+		terminate();
+		}
+	}
+
+event bro_init()
+	{
+	try = 0;	
+	outfile = open("../out");
+	Input::add_event([$source="../input.log", $name="ssh", $fields=Val, $ev=line, $want_record=T]);
+	Input::remove("ssh");
+	}

testing/btest/scripts/base/frameworks/input/subrecord.bro

@@ -0,0 +1,70 @@
+# (uses listen.bro just to ensure input sources are more reliably fully-read).
+# @TEST-SERIALIZE: comm
+#
+# @TEST-EXEC: btest-bg-run bro bro -b %INPUT
+# @TEST-EXEC: btest-bg-wait -k 5
+# @TEST-EXEC: btest-diff out
+
+@TEST-START-FILE input.log
+#separator \x09
+#path	ssh
+#fields	sub.b	i	sub.e	sub.c	sub.p	sub.sn	sub.two.a	sub.two.d	t	iv	s	sc	ss	se	vc	ve	f
+#types	bool	int	enum	count	port	subnet	addr	double	time	interval	string	table	table	table	vector	vector	func
+T	-42	SSH::LOG	21	123	10.0.0.0/24	1.2.3.4	3.14	1315801931.273616	100.000000	hurz	2,4,1,3	CC,AA,BB	EMPTY	10,20,30	EMPTY	SSH::foo\x0a{ \x0aif (0 < SSH::i) \x0a\x09return (Foo);\x0aelse\x0a\x09return (Bar);\x0a\x0a}
+@TEST-END-FILE
+
+@load base/protocols/ssh
+@load frameworks/communication/listen
+
+global outfile: file;
+
+redef InputAscii::empty_field = "EMPTY";
+
+module A;
+
+type Idx: record {
+	i: int;
+};
+
+type SubVal2: record {
+	a: addr;
+	d: double;
+};
+
+type SubVal: record {
+	b: bool;
+	e: Log::ID;
+	c: count;
+	p: port;
+	sn: subnet;
+	two: SubVal2;
+};
+
+type Val: record {
+	sub: SubVal;
+	t: time;
+	iv: interval;
+	s: string;
+	sc: set[count];
+	ss: set[string];
+	se: set[string];
+	vc: vector of int;
+	ve: vector of int;
+};
+
+global servers: table[int] of Val = table();
+
+event bro_init()
+	{
+	outfile = open("../out");
+	# first read in the old stuff into the table...
+	Input::add_table([$source="../input.log", $name="ssh", $idx=Idx, $val=Val, $destination=servers]);
+	Input::remove("ssh");
+	}
+
+event Input::update_finished(name: string, source:string)
+	{
+	print outfile, servers;
+	close(outfile);
+	terminate();
+	}