WIP: convert non-blocking file/stderr logging to osmo-io

WARNING: This likely runs into circular loops, if osmo_io wants to log
itself, ...

Change-Id: I64eb2600e8c1ff87aefcc158ede6b67f6a854540

3 files changed, 68 insertions, 103 deletions

diff --git a/include/osmocom/core/logging.h b/include/osmocom/core/logging.h
index e8433a14..0bc05264 100644
--- a/include/osmocom/core/logging.h
+++ b/include/osmocom/core/logging.h
@@ -11,6 +11,8 @@
 #include <osmocom/core/defs.h>
 #include <osmocom/core/linuxlist.h>
 
+struct osmo_io_fd;
+
 extern struct log_info *osmo_log_info;
 
 #ifndef DEBUG
@@ -332,7 +334,7 @@ struct log_target {
 			FILE *out;
 			const char *fname;
 			/* indirect output via write_queue and osmo_select_main() */
-			struct osmo_wqueue *wqueue;
+			struct osmo_io_fd *iofd;
 		} tgt_file;
 
 		struct {
diff --git a/src/core/logging.c b/src/core/logging.c
index 6941f9bf..a10131e1 100644
--- a/src/core/logging.c
+++ b/src/core/logging.c
@@ -66,7 +66,7 @@
 #include <osmocom/core/timer.h>
 #include <osmocom/core/thread.h>
 #include <osmocom/core/select.h>
-#include <osmocom/core/write_queue.h>
+#include <osmocom/core/osmo_io.h>
 #include <osmocom/core/gsmtap_util.h>
 
 #include <osmocom/vty/logging.h>	/* for LOGGING_STR. */
@@ -941,23 +941,6 @@ void log_set_category_filter(struct log_target *target, int category,
 }
 
 #if (!EMBEDDED)
-/* write-queue tells us we should write another msgb (log line) to the output fd */
-static int _file_wq_write_cb(struct osmo_fd *ofd, struct msgb *msg)
-{
-	int rc;
-
-	rc = write(ofd->fd, msgb_data(msg), msgb_length(msg));
-	if (rc < 0)
-		return rc;
-	if (rc != msgb_length(msg)) {
-		/* pull the number of bytes we have already written */
-		msgb_pull(msg, rc);
-		/* ask write_queue to re-insert the msgb at the head of the queue */
-		return -EAGAIN;
-	}
-	return 0;
-}
-
 /* output via buffered, blocking stdio streams */
 static void _file_output_stream(struct log_target *target, unsigned int level,
 			 const char *log)
@@ -967,15 +950,15 @@ static void _file_output_stream(struct log_target *target, unsigned int level,
 	fflush(target->tgt_file.out);
 }
 
-/* output via non-blocking write_queue, doing internal buffering */
+/* output via non-blocking osmo_io_fd, doing internal buffering */
 static void _file_raw_output(struct log_target *target, int subsys, unsigned int level, const char *file,
 			     int line, int cont, const char *format, va_list ap)
 {
 	struct msgb *msg;
 	int rc;
 
-	OSMO_ASSERT(target->tgt_file.wqueue);
-	msg = msgb_alloc_c(target->tgt_file.wqueue, MAX_LOG_SIZE, "log_file_msg");
+	OSMO_ASSERT(target->tgt_file.iofd);
+	msg = msgb_alloc_c(target->tgt_file.iofd, MAX_LOG_SIZE, "log_file_msg");
 	if (!msg)
 		return;
 
@@ -985,21 +968,9 @@ static void _file_raw_output(struct log_target *target, int subsys, unsigned int
 	rc = _output_buf((char *)msgb_data(msg), msgb_tailroom(msg), target, subsys, level, file, line, cont, format, ap);
 	msgb_put(msg, rc);
 
-	/* attempt a synchronous, non-blocking write, if the write queue is empty */
-	if (target->tgt_file.wqueue->current_length == 0) {
-		rc = _file_wq_write_cb(&target->tgt_file.wqueue->bfd, msg);
-		if (rc == 0) {
-			/* the write was complete, we can exit early */
-			msgb_free(msg);
-			return;
-		}
-	}
-	/* if we reach here, either we already had elements in the write_queue, or the synchronous write
-	 * failed: enqueue the message to the write_queue (backlog) */
-	if (osmo_wqueue_enqueue_quiet(target->tgt_file.wqueue, msg) < 0) {
+	rc = osmo_iofd_write_msgb(target->tgt_file.iofd, msg);
+	if (rc < 0)
 		msgb_free(msg);
-		/* TODO: increment some counter so we can see that messages were dropped */
-	}
 }
 #endif
 
@@ -1102,7 +1073,7 @@ struct log_target *log_target_create_file_stream(const char *fname)
  */
 int log_target_file_switch_to_stream(struct log_target *target)
 {
-	struct osmo_wqueue *wq;
+	struct osmo_io_fd *iofd;
 
 	if (!target)
 		return -ENODEV;
@@ -1112,8 +1083,8 @@ int log_target_file_switch_to_stream(struct log_target *target)
 		return 1;
 	}
 
-	wq = target->tgt_file.wqueue;
-	OSMO_ASSERT(wq);
+	iofd = target->tgt_file.iofd;
+	OSMO_ASSERT(iofd);
 
 	/* re-open output as stream */
 	if (target->type == LOG_TGT_TYPE_STDERR)
@@ -1125,36 +1096,30 @@ int log_target_file_switch_to_stream(struct log_target *target)
 	}
 
 	/* synchronously write anything left in the queue */
-	while (!llist_empty(&wq->msg_queue)) {
-		struct msgb *msg = msgb_dequeue(&wq->msg_queue);
-		fwrite(msgb_data(msg), msgb_length(msg), 1, target->tgt_file.out);
-		msgb_free(msg);
-	}
 
 	/* now that everything succeeded, we can finally close the old output fd */
-	if (target->type == LOG_TGT_TYPE_FILE) {
-		osmo_fd_unregister(&wq->bfd);
-		close(wq->bfd.fd);
-		wq->bfd.fd = -1;
-	}
-
-	/* release the queue itself */
-	talloc_free(wq);
-	target->tgt_file.wqueue = NULL;
+	osmo_iofd_free(iofd); /* will close + unregister + free */
+	target->tgt_file.iofd = NULL;
 	target->output = _file_output_stream;
 	target->raw_output = NULL;
 
 	return 0;
 }
 
-/*! switch from blocking + buffered file output to non-blocking write-queue based output.
+static const struct osmo_io_ops log_file_ioops = {
+	.read_cb = NULL, 	/* we never read log files */
+	.write_cb = NULL,	/* we don't care about write completion events */
+};
+
+/*! switch from blocking + buffered file output to non-blocking osmo_io_fd/write-queue based output.
  *  \param[in] target log target which we should switch
  *  \return 0 on success; 1 if already switched before; negative on error
  *  Must be called with mutex osmo_log_tgt_mutex held, see log_tgt_mutex_lock.
  */
 int log_target_file_switch_to_wqueue(struct log_target *target)
 {
-	struct osmo_wqueue *wq;
+	struct osmo_io_fd *iofd;
+	const char *name;
 	int rc;
 
 	if (!target)
@@ -1165,6 +1130,12 @@ int log_target_file_switch_to_wqueue(struct log_target *target)
 		return 1;
 	}
 
+	if (target->type == LOG_TGT_TYPE_FILE)
+		name = target->tgt_file.fname;
+	else
+		name = "stderr";
+
+#if 0
 	/* we create a ~640kB sized talloc pool within the write-queue to ensure individual
 	 * log lines (stored as msgbs) will not put result in malloc() calls, and also to
 	 * reduce the OOM probability within logging, as the pool is already allocated */
@@ -1173,27 +1144,29 @@ int log_target_file_switch_to_wqueue(struct log_target *target)
 	if (!wq)
 		return -ENOMEM;
 	osmo_wqueue_init(wq, LOG_WQUEUE_LEN);
+#endif
+	iofd = osmo_iofd_setup(target, -1, name, OSMO_IO_FD_MODE_READ_WRITE, &log_file_ioops, target);
+	if (!iofd)
+		return -ENOMEM;
+	osmo_iofd_set_txqueue_max_length(iofd, LOG_WQUEUE_LEN);
 
 	fflush(target->tgt_file.out);
-	if (target->type == LOG_TGT_TYPE_FILE) {
+	if (target->type == LOG_TGT_TYPE_FILE)
 		rc = open(target->tgt_file.fname, O_WRONLY|O_APPEND|O_CREAT|O_NONBLOCK, 0660);
-		if (rc < 0) {
-			talloc_free(wq);
-			return -errno;
-		}
-	} else {
-		rc = STDERR_FILENO;
+	else
+		rc = dup(STDERR_FILENO);
+	if (rc < 0) {
+		osmo_iofd_free(iofd);
+		return -errno;
 	}
-	wq->bfd.fd = rc;
-	wq->bfd.when = OSMO_FD_WRITE;
-	wq->write_cb = _file_wq_write_cb;
 
-	rc = osmo_fd_register(&wq->bfd);
+	rc = osmo_iofd_register(iofd, rc);
 	if (rc < 0) {
-		talloc_free(wq);
+		osmo_iofd_free(iofd);
 		return -EIO;
 	}
-	target->tgt_file.wqueue = wq;
+
+	target->tgt_file.iofd = iofd;
 	target->raw_output = _file_raw_output;
 	target->output = NULL;
 
@@ -1212,7 +1185,7 @@ int log_target_file_switch_to_wqueue(struct log_target *target)
 struct log_target *log_target_create_file(const char *fname)
 {
 	struct log_target *target;
-	struct osmo_wqueue *wq;
+	struct osmo_io_fd *iofd;
 	int rc;
 
 	target = log_target_create();
@@ -1220,6 +1193,7 @@ struct log_target *log_target_create_file(const char *fname)
 		return NULL;
 
 	target->type = LOG_TGT_TYPE_FILE;
+#if 0
 	/* we create a ~640kB sized talloc pool within the write-queue to ensure individual
 	 * log lines (stored as msgbs) will not put result in malloc() calls, and also to
 	 * reduce the OOM probability within logging, as the pool is already allocated */
@@ -1229,24 +1203,27 @@ struct log_target *log_target_create_file(const char *fname)
 		log_target_destroy(target);
 		return NULL;
 	}
-	osmo_wqueue_init(wq, LOG_WQUEUE_LEN);
-	wq->bfd.fd = open(fname, O_WRONLY|O_APPEND|O_CREAT|O_NONBLOCK, 0660);
-	if (wq->bfd.fd < 0) {
-		talloc_free(wq);
+#endif
+	iofd = osmo_iofd_setup(target, -1, fname, OSMO_IO_FD_MODE_READ_WRITE, &log_file_ioops, target);
+	if (!iofd) {
+		log_target_destroy(target);
+		return NULL;
+	}
+	rc = open(fname, O_WRONLY|O_APPEND|O_CREAT|O_NONBLOCK, 0660);
+	if (rc < 0) {
+		osmo_iofd_free(iofd);
 		log_target_destroy(target);
 		return NULL;
 	}
-	wq->bfd.when = OSMO_FD_WRITE;
-	wq->write_cb = _file_wq_write_cb;
 
-	rc = osmo_fd_register(&wq->bfd);
+	rc = osmo_iofd_register(iofd, rc);
 	if (rc < 0) {
-		talloc_free(wq);
+		osmo_iofd_free(iofd);
 		log_target_destroy(target);
 		return NULL;
 	}
 
-	target->tgt_file.wqueue = wq;
+	target->tgt_file.iofd = iofd;
 	target->raw_output = _file_raw_output;
 	target->tgt_file.fname = talloc_strdup(target, fname);
 
@@ -1291,7 +1268,7 @@ void log_target_destroy(struct log_target *target)
 	log_del_target(target);
 
 #if (!EMBEDDED)
-	struct osmo_wqueue *wq;
+	struct osmo_io_fd *iofd;
 	switch (target->type) {
 	case LOG_TGT_TYPE_FILE:
 	case LOG_TGT_TYPE_STDERR:
@@ -1300,18 +1277,9 @@ void log_target_destroy(struct log_target *target)
 				fclose(target->tgt_file.out);
 			target->tgt_file.out = NULL;
 		}
-		wq = target->tgt_file.wqueue;
-		if (wq) {
-			if (wq->bfd.fd >= 0) {
-				osmo_fd_unregister(&wq->bfd);
-				if (target->type == LOG_TGT_TYPE_FILE)
-					close(wq->bfd.fd);
-				wq->bfd.fd = -1;
-			}
-			osmo_wqueue_clear(wq);
-			talloc_free(wq);
-			target->tgt_file.wqueue = NULL;
-		}
+		iofd = target->tgt_file.iofd;
+		if (iofd)
+			osmo_iofd_free(iofd);
 		talloc_free((void *)target->tgt_file.fname);
 		target->tgt_file.fname = NULL;
 		break;
@@ -1337,11 +1305,11 @@ void log_target_destroy(struct log_target *target)
  *  \returns 0 in case of success; negative otherwise */
 int log_target_file_reopen(struct log_target *target)
 {
-	struct osmo_wqueue *wq;
+	struct osmo_io_fd *iofd;
 	int rc;
 
 	OSMO_ASSERT(target->type == LOG_TGT_TYPE_FILE || target->type == LOG_TGT_TYPE_STDERR);
-	OSMO_ASSERT(target->tgt_file.out || target->tgt_file.wqueue);
+	OSMO_ASSERT(target->tgt_file.out || target->tgt_file.iofd);
 
 	if (target->tgt_file.out) {
 		fclose(target->tgt_file.out);
@@ -1349,18 +1317,13 @@ int log_target_file_reopen(struct log_target *target)
 		if (!target->tgt_file.out)
 			return -errno;
 	} else {
-		wq = target->tgt_file.wqueue;
-		if (wq->bfd.fd >= 0) {
-			osmo_fd_unregister(&wq->bfd);
-			close(wq->bfd.fd);
-			wq->bfd.fd = -1;
-		}
+		iofd = target->tgt_file.iofd;
+		osmo_iofd_close(iofd);
 
 		rc = open(target->tgt_file.fname, O_WRONLY|O_APPEND|O_CREAT|O_NONBLOCK, 0660);
 		if (rc < 0)
 			return -errno;
-		wq->bfd.fd = rc;
-		rc = osmo_fd_register(&wq->bfd);
+		rc = osmo_iofd_register(iofd, rc);
 		if (rc < 0)
 			return rc;
 	}
diff --git a/src/vty/logging_vty.c b/src/vty/logging_vty.c
index bfe73bb1..d05ca6f0 100644
--- a/src/vty/logging_vty.c
+++ b/src/vty/logging_vty.c
@@ -1015,7 +1015,7 @@ static int config_write_log_single(struct vty *vty, struct log_target *tgt)
 		return 1;
 		break;
 	case LOG_TGT_TYPE_STDERR:
-		if (tgt->tgt_file.wqueue)
+		if (tgt->tgt_file.iofd)
 			vty_out(vty, "log stderr%s", VTY_NEWLINE);
 		else
 			vty_out(vty, "log stderr blocking-io%s", VTY_NEWLINE);
@@ -1029,7 +1029,7 @@ static int config_write_log_single(struct vty *vty, struct log_target *tgt)
 #endif
 		break;
 	case LOG_TGT_TYPE_FILE:
-		if (tgt->tgt_file.wqueue)
+		if (tgt->tgt_file.iofd)
 			vty_out(vty, "log file %s%s", tgt->tgt_file.fname, VTY_NEWLINE);
 		else
 			vty_out(vty, "log file %s blocking-io%s", tgt->tgt_file.fname, VTY_NEWLINE);