refactor logger: add need_reliable_transfer flag, remove write_wait

src/modules/logger/log_writer.cpp

@@ -118,13 +118,20 @@ void LogWriter::thread_stop()
 	}
 }
 
-bool LogWriter::write(void *ptr, size_t size, uint64_t dropout_start)
+int LogWriter::write_message(void *ptr, size_t size, uint64_t dropout_start)
 {
+	int ret_file = 0, ret_mavlink = 0;
+
 	if (_log_writer_file) {
-		return _log_writer_file->write(ptr, size, dropout_start);
+		ret_file = _log_writer_file->write_message(ptr, size, dropout_start);
 	}
 
-	return true;
+	// file backend errors takes precedence
+	if (ret_file != 0) {
+		return ret_file;
+	}
+
+	return ret_mavlink;
 }
 
 }

src/modules/logger/log_writer.h

@@ -79,11 +79,12 @@ public:
 	bool is_started(Backend backend) const;
 
 	/**
-	 * Write data to be logged. The caller must call lock() before calling this.
+	 * Write a single ulog message (including header). The caller must call lock() before calling this.
 	 * @param dropout_start timestamp when lastest dropout occured. 0 if no dropout at the moment.
-	 * @return true on success, false if not enough space in the buffer left
+	 * @return 0 on success (or if no logging started),
+	 *         -1 if not enough space in the buffer left (file backend), -2 mavlink backend failed
 	 */
-	bool write(void *ptr, size_t size, uint64_t dropout_start = 0);
+	int write_message(void *ptr, size_t size, uint64_t dropout_start = 0);
 
 
 	/* file logging methods */
@@ -124,6 +125,21 @@ public:
 		return 0;
 	}
 
+	/**
+	 * Indicate to the underlying backend whether the following writes() need a reliable
+	 * transfer. Needed for header integrity.
+	 */
+	void set_need_reliable_transfer(bool need_reliable)
+	{
+		if (_log_writer_file) { _log_writer_file->set_need_reliable_transfer(need_reliable); }
+	}
+
+	bool need_reliable_transfer() const
+	{
+		if (_log_writer_file) { return _log_writer_file->need_reliable_transfer(); }
+		return false;
+	}
+
 private:
 
 	LogWriterFile *_log_writer_file = nullptr;

src/modules/logger/log_writer_file.cpp

@@ -251,8 +251,29 @@ void LogWriterFile::run()
 	}
 }
 
-bool LogWriterFile::write(void *ptr, size_t size, uint64_t dropout_start)
+int LogWriterFile::write_message(void *ptr, size_t size, uint64_t dropout_start)
 {
+	if (_need_reliable_transfer) {
+		int ret;
+
+		while ((ret = write(ptr, size, dropout_start)) == -1) {
+			unlock();
+			notify();
+			usleep(3000);
+			lock();
+		}
+
+		return ret;
+	}
+
+	return write(ptr, size, dropout_start);
+}
+
+int LogWriterFile::write(void *ptr, size_t size, uint64_t dropout_start)
+{
+	if (!is_started()) {
+		return 0;
+	}
 
 	// Bytes available to write
 	size_t available = _buffer_size - _count;
@@ -264,7 +285,7 @@ bool LogWriterFile::write(void *ptr, size_t size, uint64_t dropout_start)
 
 	if (size + dropout_size > available) {
 		// buffer overflow
-		return false;
+		return -1;
 	}
 
 	if (dropout_start) {
@@ -275,7 +296,7 @@ bool LogWriterFile::write(void *ptr, size_t size, uint64_t dropout_start)
 	}
 
 	write_no_check(ptr, size);
-	return true;
+	return 0;
 }
 
 void LogWriterFile::write_no_check(void *ptr, size_t size)

src/modules/logger/log_writer_file.h

@@ -45,7 +45,7 @@ namespace logger
 {
 
 /**
- * @class LogWriter
+ * @class LogWriterFile
  * Writes logging data to a file
  */
 class LogWriterFile
@@ -68,17 +68,10 @@ public:
 
 	void stop_log();
 
-	/**
-	 * whether logging is currently active or not.
-	 */
 	bool is_started() const { return _should_run; }
 
-	/**
+	/** @see LogWriter::write_message() */
-	 * Write data to be logged. The caller must call lock() before calling this.
+	int write_message(void *ptr, size_t size, uint64_t dropout_start = 0);
-	 * @param dropout_start timestamp when lastest dropout occured. 0 if no dropout at the moment.
-	 * @return true on success, false if not enough space in the buffer left
-	 */
-	bool write(void *ptr, size_t size, uint64_t dropout_start = 0);
 
 	void lock()
 	{
@@ -110,6 +103,16 @@ public:
 		return _count;
 	}
 
+	void set_need_reliable_transfer(bool need_reliable)
+	{
+		_need_reliable_transfer = need_reliable;
+	}
+
+	bool need_reliable_transfer() const
+	{
+		return _need_reliable_transfer;
+	}
+
 private:
 	static void *run_helper(void *);
 
@@ -122,6 +125,11 @@ private:
 		_count -= n;
 	}
 
+	/**
+	 * write w/o waiting/blocking
+	 */
+	int write(void *ptr, size_t size, uint64_t dropout_start);
+
 	/**
 	 * Write to the buffer but assuming there is enough space
 	 */
@@ -139,6 +147,7 @@ private:
 	bool		_should_run = false;
 	bool		_running = false;
 	bool 		_exit_thread = false;
+	bool		_need_reliable_transfer = false;
 	pthread_mutex_t		_mtx;
 	pthread_cond_t		_cv;
 	perf_counter_t _perf_write;

src/modules/logger/logger.cpp

@@ -761,7 +761,7 @@ void Logger::run()
 
 						//PX4_INFO("topic: %s, size = %zu, out_size = %zu", sub.metadata->o_name, sub.metadata->o_size, msg_size);
 
-						if (write(_msg_buffer, msg_size)) {
+						if (write_message(_msg_buffer, msg_size)) {
 
 #ifdef DBGPRINT
 							total_bytes += msg_size;
@@ -792,7 +792,7 @@ void Logger::run()
 					memcpy(_msg_buffer + 4, &log_message_sub.get().timestamp, sizeof(ulog_message_logging_s::timestamp));
 					strncpy((char *)(_msg_buffer + 12), message, sizeof(ulog_message_logging_s::message));
 
-					write(_msg_buffer, write_msg_size + ULOG_MSG_HEADER_LEN);
+					write_message(_msg_buffer, write_msg_size + ULOG_MSG_HEADER_LEN);
 				}
 			}
 
@@ -860,9 +860,9 @@ void Logger::run()
 	}
 }
 
-bool Logger::write(void *ptr, size_t size)
+bool Logger::write_message(void *ptr, size_t size)
 {
-	if (_writer.write(ptr, size, _dropout_start)) {
+	if (_writer.write_message(ptr, size, _dropout_start) != -1) {
 
 		if (_dropout_start) {
 			float dropout_duration = (float)(hrt_elapsed_time(&_dropout_start) / 1000) / 1.e3f;
@@ -1089,11 +1089,13 @@ void Logger::start_log_file()
 	_next_topic_id = 0;
 
 	_writer.start_log_file(file_name);
+	_writer.set_need_reliable_transfer(true);
 	write_header();
 	write_version();
 	write_formats();
 	write_parameters();
 	write_all_add_logged_msg();
+	_writer.set_need_reliable_transfer(false);
 	_writer.notify();
 	_start_time = hrt_absolute_time();
 }
@@ -1103,18 +1105,6 @@ void Logger::stop_log_file()
 	_writer.stop_log_file();
 }
 
-bool Logger::write_wait(void *ptr, size_t size)
-{
-	while (!_writer.write(ptr, size)) {
-		_writer.unlock();
-		_writer.notify();
-		usleep(_log_interval);
-		_writer.lock();
-	}
-
-	return true;
-}
-
 void Logger::write_formats()
 {
 	_writer.lock();
@@ -1127,7 +1117,7 @@ void Logger::write_formats()
 		size_t msg_size = sizeof(msg) - sizeof(msg.format) + format_len;
 		msg.msg_size = msg_size - ULOG_MSG_HEADER_LEN;
 
-		write_wait(&msg, msg_size);
+		write_message(&msg, msg_size);
 	}
 
 	_writer.unlock();
@@ -1163,7 +1153,10 @@ void Logger::write_add_logged_msg(LoggerSubscription &subscription, int instance
 	size_t msg_size = sizeof(msg) - sizeof(msg.message_name) + message_name_len;
 	msg.msg_size = msg_size - ULOG_MSG_HEADER_LEN;
 
-	write_wait(&msg, msg_size);
+	bool prev_reliable = _writer.need_reliable_transfer();
+	_writer.set_need_reliable_transfer(true);
+	write_message(&msg, msg_size);
+	_writer.set_need_reliable_transfer(prev_reliable);
 
 	++_next_topic_id;
 }
@@ -1188,7 +1181,7 @@ void Logger::write_info(const char *name, const char *value)
 
 		msg->msg_size = msg_size - ULOG_MSG_HEADER_LEN;
 
-		write_wait(buffer, msg_size);
+		write_message(buffer, msg_size);
 	}
 
 	_writer.unlock();
@@ -1210,7 +1203,7 @@ void Logger::write_info(const char *name, int32_t value)
 
 	msg->msg_size = msg_size - ULOG_MSG_HEADER_LEN;
 
-	write_wait(buffer, msg_size);
+	write_message(buffer, msg_size);
 
 	_writer.unlock();
 }
@@ -1228,7 +1221,7 @@ void Logger::write_header()
 	header.magic[7] = 0x00; //file version 0
 	header.timestamp = hrt_absolute_time();
 	_writer.lock();
-	write_wait(&header, sizeof(header));
+	write_message(&header, sizeof(header));
 	_writer.unlock();
 }
 
@@ -1299,7 +1292,7 @@ void Logger::write_parameters()
 
 			msg->msg_size = msg_size - ULOG_MSG_HEADER_LEN;
 
-			write_wait(buffer, msg_size);
+			write_message(buffer, msg_size);
 		}
 	} while ((param != PARAM_INVALID) && (param_idx < (int) param_count()));
 
@@ -1358,7 +1351,7 @@ void Logger::write_changed_parameters()
 			/* msg_size is now 1 (msg_type) + 2 (msg_size) + 1 (key_len) + key_len + value_size */
 			msg->msg_size = msg_size - ULOG_MSG_HEADER_LEN;
 
-			write_wait(buffer, msg_size);
+			write_message(buffer, msg_size);
 		}
 	} while ((param != PARAM_INVALID) && (param_idx < (int) param_count()));
 

src/modules/logger/logger.h

@@ -175,17 +175,11 @@ private:
 	bool copy_if_updated_multi(LoggerSubscription &sub, int multi_instance, void *buffer);
 
 	/**
-	 * Write data to the logger. Waits if buffer is full until all data is written.
+	 * Write exactly one ulog message to the logger and handle dropouts.
-	 * Must be called with _writer.lock() held.
-	 */
-	bool write_wait(void *ptr, size_t size);
-
-	/**
-	 * Write data to the logger and handle dropouts.
 	 * Must be called with _writer.lock() held.
 	 * @return true if data written, false otherwise (on overflow)
 	 */
-	bool write(void *ptr, size_t size);
+	bool write_message(void *ptr, size_t size);
 
 	/**
 	 * Get the time for log file name