[tarantool-patches] Fwd: [patches] [log 2/2] say: Fix log_rotate

["Ilya Markov" <imarkov@tarantool.org>]

[-- Attachment #1: Type: text/plain, Size: 9025 bytes --]




-------- Пересылаемое сообщение --------
От кого: Ilya Markov <imarkov@tarantool.org>
Кому: georgy@tarantool.org
Копия: patches@tarantool.org
Дата: Вторник, 27 марта 2018, 17:09 +03:00
Тема: [patches] [log 2/2] say: Fix log_rotate

* Refactor tests.
* Add ev_async and fiber_cond for thread-safe log_rotate usage.

Follow up #3015
---
 src/errinj.h           |   1 +
 src/say.c              |  31 +++++++++++++-
 src/say.h              |  13 ++++++
 test/box/errinj.result |   2 +
 test/unit/say.c        | 107 ++++++++++++++++++++++++++++++++++---------------
 5 files changed, 120 insertions(+), 34 deletions(-)

diff --git a/src/errinj.h b/src/errinj.h
index a5417ba..6a5ba33 100644
--- a/src/errinj.h
+++ b/src/errinj.h
@@ -108,6 +108,7 @@ struct errinj {
 	_(ERRINJ_VY_DELAY_PK_LOOKUP, ERRINJ_BOOL, {.bparam = false}) \
 	_(ERRINJ_VY_RUN_WRITE_STMT_TIMEOUT, ERRINJ_DOUBLE, {.dparam = 0}) \
 	_(ERRINJ_IPROTO_TX_DELAY, ERRINJ_BOOL, {.bparam = false}) \
+	_(ERRINJ_LOG_ROTATE, ERRINJ_BOOL, {.bparam = false}) \
 
 ENUM0(errinj_id, ERRINJ_LIST);
 extern struct errinj errinjs[];
diff --git a/src/say.c b/src/say.c
index b92514d..2d9357a 100644
--- a/src/say.c
+++ b/src/say.c
@@ -30,6 +30,7 @@
  */
 #include "say.h"
 #include "fiber.h"
+#include "errinj.h"
 
 #include <errno.h>
 #include <stdarg.h>
@@ -239,9 +240,11 @@ write_to_syslog(struct log *log, int total);
 static int
 log_rotate(struct log *log)
 {
-	if (log->type != SAY_LOGGER_FILE) {
+	if (pm_atomic_load(&log->type) != SAY_LOGGER_FILE)
 		return 0;
-	}
+
+	ERROR_INJECT(ERRINJ_LOG_ROTATE, { usleep(10); });
+
 	int fd = open(log->path, O_WRONLY | O_APPEND | O_CREAT,
 		      S_IRUSR | S_IWUSR | S_IRGRP);
 	if (fd < 0) {
@@ -282,6 +285,7 @@ log_rotate(struct log *log)
 struct rotate_task {
 	struct coio_task base;
 	struct log *log;
+	struct ev_loop *loop;
 };
 
 static int
@@ -291,6 +295,7 @@ logrotate_cb(struct coio_task *ptr)
 	if (log_rotate(task->log) < 0) {
 		diag_log();
 	}
+	ev_async_send(task->loop, &task->log->log_async);
 	return 0;
 }
 
@@ -303,6 +308,16 @@ logrotate_cleanup_cb(struct coio_task *ptr)
 	return 0;
 }
 
+static void
+log_rotate_async_cb(struct ev_loop *loop, struct ev_async *watcher, int events)
+{
+	(void)loop;
+	(void)events;
+	struct log *log = container_of(watcher, struct log, log_async);
+	log->rotating_threads--;
+	fiber_cond_signal(&log->rotate_cond);
+}
+
 void
 say_logrotate(struct ev_loop *loop, struct ev_signal *w, int revents)
 {
@@ -320,8 +335,11 @@ say_logrotate(struct ev_loop *loop, struct ev_signal *w, int revents)
 			diag_log();
 			continue;
 		}
+		ev_async_start(loop(), &log->log_async);
+		log->rotating_threads++;
 		coio_task_create(&task->base, logrotate_cb, logrotate_cleanup_cb);
 		task->log = log;
+		task->loop = loop();
 		coio_task_post(&task->base, 0);
 	}
 	errno = saved_errno;
@@ -502,6 +520,9 @@ log_create(struct log *log, const char *init_str, bool nonblock)
 	log->format_func = NULL;
 	log->level = S_INFO;
 	log->nonblock = nonblock;
+	log->rotating_threads = 0;
+	fiber_cond_create(&log->rotate_cond);
+	ev_async_init(&log->log_async, log_rotate_async_cb);
 	setvbuf(stderr, NULL, _IONBF, 0);
 
 	if (init_str != NULL) {
@@ -1016,10 +1037,16 @@ void
 log_destroy(struct log *log)
 {
 	assert(log != NULL);
+	while(log->rotating_threads > 0)
+		fiber_cond_wait(&log->rotate_cond);
+	pm_atomic_store(&log->type, SAY_LOGGER_BOOT);
+
 	if (log->fd != -1)
 		close(log->fd);
 	free(log->syslog_ident);
 	rlist_del_entry(log, in_log_list);
+	ev_async_stop(loop(), &log->log_async);
+	fiber_cond_destroy(&log->rotate_cond);
 }
 
 static inline int
diff --git a/src/say.h b/src/say.h
index 46e6976..c05ec74 100644
--- a/src/say.h
+++ b/src/say.h
@@ -36,7 +36,9 @@
 #include <stdarg.h>
 #include <errno.h>
 #include <sys/types.h> /* pid_t */
+#include <tarantool_ev.h>
 #include "small/rlist.h"
+#include "fiber_cond.h"
 
 #if defined(__cplusplus)
 extern "C" {
@@ -119,6 +121,17 @@ struct log {
 	pid_t pid;
 	/* Application identifier used to group syslog messages. */
 	char *syslog_ident;
+	/**
+	 * Used to wake up the main logger thread from a eio thread.
+	 */
+	ev_async log_async;
+	/**
+	 * Conditional variable securing variable below
+	 * from concurrent usage.
+	 */
+	struct fiber_cond rotate_cond;
+	/** Counter identifying number of threads executing log_rotate. */
+	int rotating_threads;
 	struct rlist in_log_list;
 };
 
diff --git a/test/box/errinj.result b/test/box/errinj.result
index cd0c35c..1cb5c23 100644
--- a/test/box/errinj.result
+++ b/test/box/errinj.result
@@ -60,6 +60,8 @@ errinj.info()
     state: false
   ERRINJ_WAL_ROTATE:
     state: false
+  ERRINJ_LOG_ROTATE:
+    state: false
   ERRINJ_VY_POINT_ITER_WAIT:
     state: false
   ERRINJ_RELAY_EXIT_DELAY:
diff --git a/test/unit/say.c b/test/unit/say.c
index 8c996ba..0ffd267 100644
--- a/test/unit/say.c
+++ b/test/unit/say.c
@@ -4,6 +4,8 @@
 #include "unit.h"
 #include "say.h"
 #include <pthread.h>
+#include <errinj.h>
+#include <coio_task.h>
 
 int
 parse_logger_type(const char *input)
@@ -67,28 +69,34 @@ pthread_cond_t cond_sync = PTHREAD_COND_INITIALIZER;
 
 bool is_raised = false;
 int created_logs = 0;
+const char *tmp_dir;
+
+struct create_log {
+	struct log logger;
+	int id;
+};
 
 static void *
 dummy_log(void *arg)
 {
-	const char *tmp_dir = (const char *) arg;
+	struct create_log *create_log = (struct create_log *) arg;
+
 	char tmp_filename[30];
-	sprintf(tmp_filename, "%s/%i.log", tmp_dir, (int) pthread_self());
-	pthread_mutex_lock(&mutex);
-	struct log test_log;
-	log_create(&test_log, tmp_filename, false);
-	// signal that log is created
+	sprintf(tmp_filename, "%s/%i.log", tmp_dir, create_log->id);
+	tt_pthread_mutex_lock(&mutex);
+	log_create(&create_log->logger, tmp_filename, false);
+
+	/* signal that log is created */
 	created_logs++;
-	pthread_cond_signal(&cond_sync);
+	tt_pthread_cond_signal(&cond_sync);
 
-	// wait until rotate signal is raised
+	/* wait until rotate signal is raised */
 	while (!is_raised)
-		pthread_cond_wait(&cond, &mutex);
-
-	log_destroy(&test_log);
+		tt_pthread_cond_wait(&cond, &mutex);
 	created_logs--;
-	pthread_cond_signal(&cond_sync);
-	pthread_mutex_unlock(&mutex);
+	if (created_logs == 0)
+		pthread_cond_signal(&cond_sync);
+	tt_pthread_mutex_unlock(&mutex);
 	return NULL;
 }
 
@@ -96,26 +104,54 @@ static void
 test_log_rotate()
 {
 	char template[] = "/tmp/tmpdir.XXXXXX";
-	const char *tmp_dir = mkdtemp(template);
+	tmp_dir = mkdtemp(template);
+	const int NUMBER_LOGGERS = 10;
+	struct create_log *loggers = (struct create_log *) calloc(NUMBER_LOGGERS,
+						  sizeof(struct create_log));
+	if (loggers == NULL) {
+		return;
+	}
 	int running = 0;
-	for (int i = 0; i < 10; i++) {
+	for (int i = 0; i < NUMBER_LOGGERS; i++) {
 		pthread_t thread;
-		if (pthread_create(&thread, NULL, dummy_log, (void *) tmp_dir) >= 0)
+		loggers[i].id = i;
+		if (tt_pthread_create(&thread, NULL, dummy_log,
+				   (void *) &loggers[i]) >= 0)
 			running++;
 	}
-	pthread_mutex_lock(&mutex);
-	// wait loggers are created
-	while (created_logs < running) {
-		pthread_cond_wait(&cond_sync, &mutex);
+	tt_pthread_mutex_lock(&mutex);
+	/* wait loggers are created */
+	while (created_logs < running)
+		tt_pthread_cond_wait(&cond_sync, &mutex);
+	tt_pthread_mutex_unlock(&mutex);
+	say_logrotate(NULL, NULL, 0);
+
+	for (int i = 0; i < created_logs; i++) {
+		log_destroy(&loggers[i].logger);
 	}
-	raise(SIGHUP);
+	memset(loggers, '#', NUMBER_LOGGERS * sizeof(struct create_log));
+	free(loggers);
+
 	is_raised = true;
-	pthread_cond_broadcast(&cond);
+	tt_pthread_cond_broadcast(&cond);
 
-	// wait until loggers are closed
-	while(created_logs != 0)
-		pthread_cond_wait(&cond_sync, &mutex);
-	pthread_mutex_unlock(&mutex);
+	tt_pthread_mutex_lock(&mutex);
+	/* wait threads are finished */
+	while (created_logs > 0)
+		tt_pthread_cond_wait(&cond_sync, &mutex);
+	tt_pthread_mutex_unlock(&mutex);
+}
+
+static int
+main_f(va_list ap)
+{
+	struct errinj *inj = errinj_by_name("ERRINJ_LOG_ROTATE");
+	inj->bparam = true;
+	/* test on log_rotate signal handling */
+	test_log_rotate();
+	inj->bparam = false;
+	ev_break(loop(), EVBREAK_ALL);
+	return 0;
 }
 
 int main()
@@ -187,11 +223,18 @@ int main()
 	}
 	log_destroy(&test_log);
 
-	// test on log_rotate signal handling
-	struct ev_signal ev_sig;
-	ev_signal_init(&ev_sig, say_logrotate, SIGHUP);
-	ev_signal_start(loop(), &ev_sig);
-	test_log_rotate();
-	ev_signal_stop(loop(), &ev_sig);
+	coio_init();
+	coio_enable();
+
+	struct fiber *test = fiber_new("loggers", main_f);
+	if (test == NULL) {
+		diag_log();
+		return check_plan();
+	}
+	fiber_wakeup(test);
+	ev_run(loop(), 0);
+
+	fiber_free();
+	memory_free();
 	return check_plan();
 }
-- 
2.7.4


----------------------------------------------------------------------

-- 
Thank you,
Ilya Markov

[-- Attachment #2: Type: text/html, Size: 11596 bytes --]