Reworking thread termination logic.

Turns out the finish methods weren't called correctly, caused by a
mess up with method names which all sounded too similar and the wrong
one ended up being called. I've reworked this by changing the
thread/writer/reader interfaces, which actually also simplifies them
by getting rid of the requirement for writer backends to call their
parent methods (i.e., less opportunity for errors).

This commit also includes the following (because I noticed the problem
above when working on some of these):

     - The ASCII log writer now includes "#start <timestamp>" and
      "#end <timestamp> lines in the each file. The latter supersedes
      Bernhard's "EOF" patch.

      This required a number of tests updates. The standard canonifier
      removes the timestamps, but some tests compare files directly,
      which doesn't work if they aren't printing out the same
      timestamps (like the comm tests).

     - The above required yet another change to the writer API to
       network_time to methods.

     - Renamed ASCII logger "header" options to "meta".

     - Fixes #763 "Escape # when first character in log file line".

All btests pass for me on Linux FC15. Will try MacOS next.

src/logging/Manager.cc

@@ -771,6 +771,7 @@ bool Manager::Write(EnumVal* id, RecordVal* columns)
 
 			WriterBackend::WriterInfo info;
 			info.path = path;
+			info.network_time = network_time;
 
 			HashKey* k;
 			IterCookie* c = filter->config->AsTable()->InitForIteration();
@@ -1156,7 +1157,7 @@ bool Manager::Flush(EnumVal* id)
 
 	for ( Stream::WriterMap::iterator i = stream->writers.begin();
 	      i != stream->writers.end(); i++ )
-		i->second->writer->Flush();
+		i->second->writer->Flush(network_time);
 
 	RemoveDisabledWriters(stream);
 

src/logging/WriterBackend.cc

@@ -18,7 +18,7 @@ namespace logging  {
 class RotationFinishedMessage : public threading::OutputMessage<WriterFrontend>
 {
 public:
-        RotationFinishedMessage(WriterFrontend* writer, string new_name, string old_name,
+	RotationFinishedMessage(WriterFrontend* writer, string new_name, string old_name,
 				double open, double close, bool terminating)
 		: threading::OutputMessage<WriterFrontend>("RotationFinished", writer),
 		new_name(new_name), old_name(old_name), open(open),
@@ -260,9 +260,9 @@ bool WriterBackend::Rotate(string rotated_path, double open,
 	return true;
 	}
 
-bool WriterBackend::Flush()
+bool WriterBackend::Flush(double network_time)
 	{
-	if ( ! DoFlush() )
+	if ( ! DoFlush(network_time) )
 		{
 		DisableFrontend();
 		return false;
@@ -271,13 +271,15 @@ bool WriterBackend::Flush()
 	return true;
 	}
 
-bool WriterBackend::DoHeartbeat(double network_time, double current_time)
+bool WriterBackend::OnFinish(double network_time)
 	{
-	MsgThread::DoHeartbeat(network_time, current_time);
+	return DoFinish(network_time);
+	}
 
+bool WriterBackend::OnHeartbeat(double network_time, double current_time)
+	{
 	SendOut(new FlushWriteBufferMessage(frontend));
-
-	return true;
+	return DoHeartbeat(network_time, current_time);
 	}
 
 string WriterBackend::Render(const threading::Value::addr_t& addr) const

src/logging/WriterBackend.h

@@ -67,6 +67,11 @@ public:
 		 */
 		double rotation_base;
 
+		/**
+		 * The network time when the writer is created.
+		 */
+		double network_time;
+
 		/**
 		 * A map of key/value pairs corresponding to the relevant
 		 * filter's "config" table.
@@ -129,9 +134,11 @@ public:
 	 * Flushes any currently buffered output, assuming the writer
 	 * supports that. (If not, it will be ignored).
 	 *
+	 * @param network_time The network time when the flush was triggered.
+	 *
 	 * @return False if an error occured.
 	 */
-	bool Flush();
+	bool Flush(double network_time);
 
 	/**
 	 * Triggers rotation, if the writer supports that. (If not, it will
@@ -213,6 +220,10 @@ public:
 	  */
 	string Render(double d) const;
 
+	// Overridden from MsgThread.
+	virtual bool OnHeartbeat(double network_time, double current_time);
+	virtual bool OnFinish(double network_time);
+
 protected:
 	friend class FinishMessage;
 
@@ -272,8 +283,10 @@ protected:
 	 * will then be disabled and eventually deleted. When returning
 	 * false, an implementation should also call Error() to indicate what
 	 * happened.
+	 *
+	 * @param network_time The network time when the flush was triggered.
 	 */
-	virtual bool DoFlush() = 0;
+	virtual bool DoFlush(double network_time) = 0;
 
 	/**
 	 * Writer-specific method implementing log rotation.  Most directly
@@ -314,20 +327,19 @@ protected:
 
 	/**
 	 * Writer-specific method called just before the threading system is
-	 * going to shutdown.
+	 * going to shutdown. It is assumed that once this messages returns,
+	 * the thread can be safely terminated.
 	 *
-	 * This method can be overridden but one must call
-	 * WriterBackend::DoFinish().
+	 * @param network_time The network time when the finish is triggered.
 	 */
-	virtual bool DoFinish() { return MsgThread::DoFinish(); }
-
+	virtual bool DoFinish(double network_time) = 0;
 	/**
 	 * Triggered by regular heartbeat messages from the main thread.
 	 *
-	 * This method can be overridden but one must call
-	 * WriterBackend::DoHeartbeat().
+	 * This method can be overridden. Default implementation does
+	 * nothing.
 	 */
-	virtual bool DoHeartbeat(double network_time, double current_time);
+	virtual bool DoHeartbeat(double network_time, double current_time) = 0;
 
 private:
 	/**

src/logging/WriterFrontend.cc

@@ -81,19 +81,13 @@ private:
 class FlushMessage : public threading::InputMessage<WriterBackend>
 {
 public:
-	FlushMessage(WriterBackend* backend)
-		: threading::InputMessage<WriterBackend>("Flush", backend)	{}
+	FlushMessage(WriterBackend* backend, double network_time)
+		: threading::InputMessage<WriterBackend>("Flush", backend),
+		network_time(network_time) {}
 
-	virtual bool Process() { return Object()->Flush(); }
-};
-
-class FinishMessage : public threading::InputMessage<WriterBackend>
-{
-public:
-	FinishMessage(WriterBackend* backend)
-		: threading::InputMessage<WriterBackend>("Finish", backend)	{}
-
-	virtual bool Process() { return Object()->DoFinish(); }
+	virtual bool Process() { return Object()->Flush(network_time); }
+private:
+	double network_time;
 };
 
 }
@@ -240,7 +234,7 @@ void WriterFrontend::SetBuf(bool enabled)
 		FlushWriteBuffer();
 	}
 
-void WriterFrontend::Flush()
+void WriterFrontend::Flush(double network_time)
 	{
 	if ( disabled )
 		return;
@@ -248,7 +242,7 @@ void WriterFrontend::Flush()
 	FlushWriteBuffer();
 
 	if ( backend )
-		backend->SendIn(new FlushMessage(backend));
+		backend->SendIn(new FlushMessage(backend, network_time));
 	}
 
 void WriterFrontend::Rotate(string rotated_path, double open, double close, bool terminating)
@@ -266,17 +260,6 @@ void WriterFrontend::Rotate(string rotated_path, double open, double close, bool
 		log_mgr->FinishedRotation(0, "", rotated_path, open, close, terminating);
 	}
 
-void WriterFrontend::Finish()
-	{
-	if ( disabled )
-		return;
-
-	FlushWriteBuffer();
-
-	if ( backend )
-		backend->SendIn(new FinishMessage(backend));
-	}
-
 void WriterFrontend::DeleteVals(Value** vals)
 	{
 	// Note this code is duplicated in Manager::DeleteVals().

src/logging/WriterFrontend.h

@@ -114,8 +114,10 @@ public:
 	 * message back that will asynchronously call Disable().
 	 *
 	 * This method must only be called from the main thread.
+	 *
+	 * @param network_time The network time when the flush was triggered.
 	 */
-	void Flush();
+	void Flush(double network_time);
 
 	/**
 	 * Triggers log rotation.
@@ -138,8 +140,10 @@ public:
 	 * sends a message back that will asynchronously call Disable().
 	 *
 	 * This method must only be called from the main thread.
+	 *
+	 * @param network_time The network time when the finish was triggered.
 	 */
-	void Finish();
+	void Finish(double network_time);
 
 	/**
 	 * Explicitly triggers a transfer of all potentially buffered Write()

src/logging/writers/Ascii.cc

@@ -18,7 +18,7 @@ Ascii::Ascii(WriterFrontend* frontend) : WriterBackend(frontend)
 	file = 0;
 
 	output_to_stdout = BifConst::LogAscii::output_to_stdout;
-	include_header = BifConst::LogAscii::include_header;
+	include_meta = BifConst::LogAscii::include_meta;
 
 	separator_len = BifConst::LogAscii::separator->Len();
 	separator = new char[separator_len];
@@ -40,10 +40,10 @@ Ascii::Ascii(WriterFrontend* frontend) : WriterBackend(frontend)
 	memcpy(unset_field, BifConst::LogAscii::unset_field->Bytes(),
 	       unset_field_len);
 
-	header_prefix_len = BifConst::LogAscii::header_prefix->Len();
-	header_prefix = new char[header_prefix_len];
-	memcpy(header_prefix, BifConst::LogAscii::header_prefix->Bytes(),
-	       header_prefix_len);
+	meta_prefix_len = BifConst::LogAscii::meta_prefix->Len();
+	meta_prefix = new char[meta_prefix_len];
+	memcpy(meta_prefix, BifConst::LogAscii::meta_prefix->Bytes(),
+	       meta_prefix_len);
 
 	desc.EnableEscaping();
 	desc.AddEscapeSequence(separator, separator_len);
@@ -51,24 +51,39 @@ Ascii::Ascii(WriterFrontend* frontend) : WriterBackend(frontend)
 
 Ascii::~Ascii()
 	{
+	// Normally, the file will be closed here already via the Finish()
+	// message. But when we terminate abnormally, we may still have it
+	// open.
 	if ( file )
-		fclose(file);
+		CloseFile(0);
 
 	delete [] separator;
 	delete [] set_separator;
 	delete [] empty_field;
 	delete [] unset_field;
-	delete [] header_prefix;
+	delete [] meta_prefix;
 	}
 
 bool Ascii::WriteHeaderField(const string& key, const string& val)
 	{
-	string str = string(header_prefix, header_prefix_len) +
+	string str = string(meta_prefix, meta_prefix_len) +
 		key + string(separator, separator_len) + val + "\n";
 
 	return (fwrite(str.c_str(), str.length(), 1, file) == 1);
 	}
 
+void Ascii::CloseFile(double t)
+	{
+	if ( ! file )
+		return;
+
+	if ( include_meta )
+		WriteHeaderField("end", t ? Timestamp(t) : "<abnormal termination>");
+
+	fclose(file);
+	file = 0;
+	}
+
 bool Ascii::DoInit(const WriterInfo& info, int num_fields, const Field* const * fields)
 	{
 	string path = info.path;
@@ -81,17 +96,17 @@ bool Ascii::DoInit(const WriterInfo& info, int num_fields, const Field* const *
 	if ( ! (file = fopen(fname.c_str(), "w")) )
 		{
 		Error(Fmt("cannot open %s: %s", fname.c_str(),
-			  strerror(errno)));
+			  Strerror(errno)));
 
 		return false;
 		}
 
-	if ( include_header )
+	if ( include_meta )
 		{
 		string names;
 		string types;
 
-		string str = string(header_prefix, header_prefix_len)
+		string str = string(meta_prefix, meta_prefix_len)
 			+ "separator " // Always use space as separator here.
 			+ get_escaped_string(string(separator, separator_len), false)
 			+ "\n";
@@ -105,8 +120,9 @@ bool Ascii::DoInit(const WriterInfo& info, int num_fields, const Field* const *
 		            string(empty_field, empty_field_len), false)) &&
 		        WriteHeaderField("unset_field", get_escaped_string(
 		            string(unset_field, unset_field_len), false)) &&
-		        WriteHeaderField("path", get_escaped_string(path, false))) )
-			goto write_error;
+		        WriteHeaderField("path", get_escaped_string(path, false)) &&
+			WriteHeaderField("start", Timestamp(info.network_time))) )
+		goto write_error;
 
 		for ( int i = 0; i < num_fields; ++i )
 			{
@@ -128,21 +144,23 @@ bool Ascii::DoInit(const WriterInfo& info, int num_fields, const Field* const *
 	return true;
 
 write_error:
-	Error(Fmt("error writing to %s: %s", fname.c_str(), strerror(errno)));
+	Error(Fmt("error writing to %s: %s", fname.c_str(), Strerror(errno)));
 	return false;
 	}
 
-bool Ascii::DoFlush()
+bool Ascii::DoFlush(double network_time)
 	{
 	fflush(file);
 	return true;
 	}
 
-bool Ascii::DoFinish()
+bool Ascii::DoFinish(double network_time)
 	{
-	return WriterBackend::DoFinish();
+	CloseFile(network_time);
+	return true;
 	}
 
+
 bool Ascii::DoWriteOne(ODesc* desc, Value* val, const Field* field)
 	{
 	if ( ! val->present )
@@ -307,16 +325,33 @@ bool Ascii::DoWrite(int num_fields, const Field* const * fields,
 
 	desc.AddRaw("\n", 1);
 
-	if ( fwrite(desc.Bytes(), desc.Len(), 1, file) != 1 )
+	const char* bytes = (const char*)desc.Bytes();
+	int len = desc.Len();
+
+	// Make sure the line doesn't look like meta information.
+	if ( strncmp(bytes, meta_prefix, meta_prefix_len) == 0 )
 		{
-		Error(Fmt("error writing to %s: %s", fname.c_str(), strerror(errno)));
-		return false;
+		// It would so escape the first character.
+		char buf[16];
+		snprintf(buf, sizeof(buf), "\\x%02x", bytes[0]);
+		if ( fwrite(buf, strlen(buf), 1, file) != 1 )
+			goto write_error;
+
+		++bytes;
+		--len;
 		}
 
+	if ( fwrite(bytes, len, 1, file) != 1 )
+		goto write_error;
+
 	if ( IsBuf() )
 		fflush(file);
 
 	return true;
+
+write_error:
+	Error(Fmt("error writing to %s: %s", fname.c_str(), Strerror(errno)));
+	return false;
 	}
 
 bool Ascii::DoRotate(string rotated_path, double open, double close, bool terminating)
@@ -325,8 +360,7 @@ bool Ascii::DoRotate(string rotated_path, double open, double close, bool termin
 	if ( ! file || IsSpecial(Info().path) )
 		return true;
 
-	fclose(file);
-	file = 0;
+	CloseFile(close);
 
 	string nname = rotated_path + "." + LogExt();
 	rename(fname.c_str(), nname.c_str());
@@ -346,9 +380,28 @@ bool Ascii::DoSetBuf(bool enabled)
 	return true;
 	}
 
+bool Ascii::DoHeartbeat(double network_time, double current_time)
+	{
+	// Nothing to do.
+	return true;
+	}
+
 string Ascii::LogExt()
 	{
 	const char* ext = getenv("BRO_LOG_SUFFIX");
 	if ( ! ext ) ext = "log";
 	return ext;
 	}
+
+string Ascii::Timestamp(double t)
+	{
+	struct tm tm;
+	char buf[128];
+	const char* const date_fmt = "%Y-%m-%d-%H-%M-%S";
+	time_t teatime = time_t(t);
+
+	localtime_r(&teatime, &tm);
+	strftime(buf, sizeof(buf), date_fmt, &tm);
+	return buf;
+	}
+

src/logging/writers/Ascii.h

@@ -26,13 +26,16 @@ protected:
 	virtual bool DoSetBuf(bool enabled);
 	virtual bool DoRotate(string rotated_path, double open,
 			      double close, bool terminating);
-	virtual bool DoFlush();
-	virtual bool DoFinish();
+	virtual bool DoFlush(double network_time);
+	virtual bool DoFinish(double network_time);
+	virtual bool DoHeartbeat(double network_time, double current_time);
 
 private:
 	bool IsSpecial(string path) 	{ return path.find("/dev/") == 0; }
 	bool DoWriteOne(ODesc* desc, threading::Value* val, const threading::Field* field);
 	bool WriteHeaderField(const string& key, const string& value);
+	void CloseFile(double t);
+	string Timestamp(double t);
 
 	FILE* file;
 	string fname;
@@ -40,7 +43,7 @@ private:
 
 	// Options set from the script-level.
 	bool output_to_stdout;
-	bool include_header;
+	bool include_meta;
 
 	char* separator;
 	int separator_len;
@@ -54,8 +57,8 @@ private:
 	char* unset_field;
 	int unset_field_len;
 
-	char* header_prefix;
-	int header_prefix_len;
+	char* meta_prefix;
+	int meta_prefix_len;
 };
 
 }

src/logging/writers/DataSeries.cc

@@ -311,7 +311,7 @@ bool DataSeries::DoInit(const WriterInfo& info, int num_fields, const threading:
 			}
 
 		else
-			Error(Fmt("cannot dump schema: %s", strerror(errno)));
+			Error(Fmt("cannot dump schema: %s", Strerror(errno)));
 		}
 
 	compress_type = Extent::compress_all;
@@ -343,7 +343,7 @@ bool DataSeries::DoInit(const WriterInfo& info, int num_fields, const threading:
 	return OpenLog(info.path);
 	}
 
-bool DataSeries::DoFlush()
+bool DataSeries::DoFlush(double network_time)
 {
 	// Flushing is handled by DataSeries automatically, so this function
 	// doesn't do anything.
@@ -366,11 +366,10 @@ void DataSeries::CloseLog()
 	log_file = 0;
 	}
 
-bool DataSeries::DoFinish()
+bool DataSeries::DoFinish(double network_time)
 {
 	CloseLog();
-
-	return WriterBackend::DoFinish();
+	return true;
 }
 
 bool DataSeries::DoWrite(int num_fields, const threading::Field* const * fields,
@@ -420,4 +419,9 @@ bool DataSeries::DoSetBuf(bool enabled)
 	return true;
 }
 
+bool DataSeries::DoHeartbeat(double network_time, double current_time)
+{
+	return true;
+}
+
 #endif /* USE_DATASERIES */

src/logging/writers/DataSeries.h

@@ -34,8 +34,9 @@ protected:
 	virtual bool DoSetBuf(bool enabled);
 	virtual bool DoRotate(string rotated_path, double open,
 			      double close, bool terminating);
-	virtual bool DoFlush();
-	virtual bool DoFinish();
+	virtual bool DoFlush(double network_time);
+	virtual bool DoFinish(double network_time);
+	virtual bool DoHeartbeat(double network_time, double current_time);
 
 private:
 	static const size_t ROW_MIN = 2048;			// Minimum extent size.

src/logging/writers/None.h

@@ -26,8 +26,9 @@ protected:
 	virtual bool DoSetBuf(bool enabled)	{ return true; }
 	virtual bool DoRotate(string rotated_path, double open,
 			      double close, bool terminating);
-	virtual bool DoFlush()	{ return true; }
-	virtual bool DoFinish()	{ WriterBackend::DoFinish(); return true; }
+	virtual bool DoFlush(double network_time)	{ return true; }
+	virtual bool DoFinish(double network_time)	{ return true; }
+	virtual bool DoHeartbeat(double network_time, double current_time)	{ return true; }
 };
 
 }