PGPRO-1311: Do not call exit() for thread, use pthread_exit() instead

Author: za-arthur

src/backup.c

@@ -47,10 +47,22 @@ const char *progname = "pg_probackup";
 static parray *backup_files_list = NULL;
 
 static pthread_mutex_t start_stream_mut = PTHREAD_MUTEX_INITIALIZER;
+
 /*
  * We need to wait end of WAL streaming before execute pg_stop_backup().
  */
+typedef struct
+{
+	const char *basedir;
+	/*
+	 * Return value from the thread.
+	 * 0 means there is no error, 1 - there is an error.
+	 */
+	int			ret;
+} StreamThreadArg;
+
 static pthread_t stream_thread;
+static StreamThreadArg stream_thread_arg = {"", 1};
 
 static int is_ptrack_enable = false;
 bool is_ptrack_support = false;
@@ -424,6 +436,9 @@ remote_backup_files(void *arg)
 				 file->path, (unsigned long) file->write_size);
 		PQfinish(file_backup_conn);
 	}
+
+	/* Data files transferring is successful */
+	arguments->ret = 0;
 }
 
 /*
@@ -441,6 +456,7 @@ do_backup_instance(void)
 
 	pthread_t	backup_threads[num_threads];
 	backup_files_args *backup_threads_args[num_threads];
+	bool		backup_isok = true;
 
 	pgBackup   *prev_backup = NULL;
 	char		prev_backup_filelist_path[MAXPGPATH];
@@ -541,8 +557,13 @@ do_backup_instance(void)
 		join_path_components(dst_backup_path, database_path, PG_XLOG_DIR);
 		dir_create_dir(dst_backup_path, DIR_PERMISSION);
 
+		stream_thread_arg.basedir = dst_backup_path;
+		/* By default there are some error */
+		stream_thread_arg.ret = 1;
+
 		pthread_mutex_lock(&start_stream_mut);
-		pthread_create(&stream_thread, NULL, (void *(*)(void *)) StreamLog, dst_backup_path);
+		pthread_create(&stream_thread, NULL, (void *(*)(void *)) StreamLog,
+					   &stream_thread_arg);
 		pthread_mutex_lock(&start_stream_mut);
 		if (conn == NULL)
 			elog(ERROR, "Cannot continue backup because stream connect has failed.");
@@ -653,6 +674,8 @@ do_backup_instance(void)
 		arg->prev_backup_start_lsn = prev_backup_start_lsn;
 		arg->thread_backup_conn = NULL;
 		arg->thread_cancel_conn = NULL;
+		/* By default there are some error */
+		arg->ret = 1;
 		backup_threads_args[i] = arg;
 	}
 
@@ -676,9 +699,15 @@ do_backup_instance(void)
 	for (i = 0; i < num_threads; i++)
 	{
 		pthread_join(backup_threads[i], NULL);
+		if (backup_threads_args[i]->ret == 1)
+			backup_isok = false;
+
 		pg_free(backup_threads_args[i]);
 	}
-	elog(LOG, "Data files are transfered");
+	if (backup_isok)
+		elog(LOG, "Data files are transfered");
+	else
+		elog(ERROR, "Data files transferring failed");
 
 	/* clean previous backup file list */
 	if (prev_backup_filelist)
@@ -1784,8 +1813,12 @@ pg_stop_backup(pgBackup *backup)
 		PQclear(res);
 
 		if (stream_wal)
+		{
 			/* Wait for the completion of stream */
 			pthread_join(stream_thread, NULL);
+			if (stream_thread_arg.ret == 1)
+				elog(ERROR, "WAL streaming failed");
+		}
 	}
 
 	/* Fill in fields if that is the correct end of backup. */
@@ -2025,6 +2058,8 @@ backup_files(void *arg)
 	if (arguments->thread_backup_conn)
 		pgut_disconnect(arguments->thread_backup_conn);
 
+	/* Data files transferring is successful */
+	arguments->ret = 0;
 }
 
 /*
@@ -2616,7 +2651,7 @@ StreamLog(void *arg)
 {
 	XLogRecPtr	startpos;
 	TimeLineID	starttli;
-	char	   *basedir = (char *)arg;
+	StreamThreadArg *stream_arg = (StreamThreadArg *) arg;
 
 	/*
 	 * Connect in replication mode to the server
@@ -2682,7 +2717,7 @@ StreamLog(void *arg)
 		ctl.sysidentifier = NULL;
 
 #if PG_VERSION_NUM >= 100000
-		ctl.walmethod = CreateWalDirectoryMethod(basedir, 0, true);
+		ctl.walmethod = CreateWalDirectoryMethod(stream_arg->basedir, 0, true);
 		ctl.replication_slot = replication_slot;
 		ctl.stop_socket = PGINVALID_SOCKET;
 #else
@@ -2713,6 +2748,7 @@ StreamLog(void *arg)
 
 	elog(LOG, _("finished streaming WAL at %X/%X (timeline %u)"),
 		 (uint32) (stop_stream_lsn >> 32), (uint32) stop_stream_lsn, starttli);
+	stream_arg->ret = 0;
 
 	PQfinish(conn);
 	conn = NULL;

src/pg_probackup.h

@@ -241,19 +241,25 @@ typedef struct pgRecoveryTarget
 /* Union to ease operations on relation pages */
 typedef union DataPage
 {
-       PageHeaderData  page_data;
-       char                    data[BLCKSZ];
+	PageHeaderData page_data;
+	char		data[BLCKSZ];
 } DataPage;
 
 typedef struct
 {
 	const char *from_root;
 	const char *to_root;
-	parray *backup_files_list;
-	parray *prev_backup_filelist;
-	XLogRecPtr prev_backup_start_lsn;
-	PGconn *thread_backup_conn;
-	PGcancel *thread_cancel_conn;
+	parray	   *backup_files_list;
+	parray	   *prev_backup_filelist;
+	XLogRecPtr	prev_backup_start_lsn;
+	PGconn	   *thread_backup_conn;
+	PGcancel   *thread_cancel_conn;
+
+	/*
+	 * Return value from the thread.
+	 * 0 means there is no error, 1 - there is an error.
+	 */
+	int			ret;
 } backup_files_args;
 
 /*

src/restore.c

@@ -22,6 +22,12 @@ typedef struct
 {
 	parray *files;
 	pgBackup *backup;
+
+	/*
+	 * Return value from the thread.
+	 * 0 means there is no error, 1 - there is an error.
+	 */
+	int			ret;
 } restore_files_args;
 
 /* Tablespace mapping structures */
@@ -374,6 +380,7 @@ restore_backup(pgBackup *backup)
 	int			i;
 	pthread_t	restore_threads[num_threads];
 	restore_files_args *restore_threads_args[num_threads];
+	bool		restore_isok = true;
 
 	if (backup->status != BACKUP_STATUS_OK)
 		elog(ERROR, "Backup %s cannot be restored because it is not valid",
@@ -419,19 +426,27 @@ restore_backup(pgBackup *backup)
 		restore_files_args *arg = pg_malloc(sizeof(restore_files_args));
 		arg->files = files;
 		arg->backup = backup;
+		/* By default there are some error */
+		arg->ret = 1;
 
 		elog(LOG, "Start thread for num:%li", parray_num(files));
 
 		restore_threads_args[i] = arg;
-		pthread_create(&restore_threads[i], NULL, (void *(*)(void *)) restore_files, arg);
+		pthread_create(&restore_threads[i], NULL,
+					   (void *(*)(void *)) restore_files, arg);
 	}
 
 	/* Wait theads */
 	for (i = 0; i < num_threads; i++)
 	{
 		pthread_join(restore_threads[i], NULL);
+		if (restore_threads_args[i]->ret == 1)
+			restore_isok = false;
+
 		pg_free(restore_threads_args[i]);
 	}
+	if (!restore_isok)
+		elog(ERROR, "Data files restoring failed");
 
 	/* cleanup */
 	parray_walk(files, pgFileFree);
@@ -777,6 +792,9 @@ restore_files(void *arg)
 			elog(LOG, "Restored file %s : %lu bytes",
 				file->path, (unsigned long) file->write_size);
 	}
+
+	/* Data files restoring is successful */
+	arguments->ret = 0;
 }
 
 static void

src/utils/logger.c

@@ -142,7 +142,7 @@ elog_internal(int elevel, bool file_only, const char *fmt, va_list args)
 	 * There is no need to lock if this is elog() from upper elog() and
 	 * logging is not initialized.
 	 */
-	if (write_to_file || write_to_error_log)
+	if (write_to_file || write_to_error_log || write_to_stderr)
 		pthread_mutex_lock(&log_file_mutex);
 
 	/* We need copy args only if we need write to error log file */
@@ -228,12 +228,25 @@ elog_internal(int elevel, bool file_only, const char *fmt, va_list args)
 			va_end(std_args);
 	}
 
-	if (write_to_file || write_to_error_log)
+	if (write_to_file || write_to_error_log || write_to_stderr)
 		pthread_mutex_unlock(&log_file_mutex);
 
-	/* Exit with code if it is an error */
-	if (elevel > WARNING)
-		exit(elevel);
+	/*
+	 * Exit with code if it is an error.
+	 * Check for in_cleanup flag to avoid deadlock in case of ERROR in cleanup
+	 * routines.
+	 */
+	if (elevel > WARNING && !in_cleanup)
+	{
+		/* Interrupt other possible routines */
+		interrupted = true;
+
+		/* If this is not the main thread then don't call exit() */
+		if (!pthread_equal(main_tid, pthread_self()))
+			pthread_exit(NULL);
+		else
+			exit(elevel);
+	}
 }
 
 /*

src/utils/pgut.c

@@ -42,6 +42,8 @@ static char	   *password = NULL;
 bool			prompt_password = true;
 bool			force_password = false;
 
+pthread_t main_tid = 0;
+
 /* Database connections */
 static PGcancel *volatile cancel_conn = NULL;
 
@@ -1065,6 +1067,7 @@ pgut_getopt(int argc, char **argv, pgut_option options[])
 
 	init_cancel_handler();
 	atexit(on_cleanup);
+	main_tid = pthread_self();
 
 	return optind;
 }

src/utils/pgut.h

@@ -15,6 +15,7 @@
 #include "pqexpbuffer.h"
 
 #include <assert.h>
+#include <pthread.h>
 #include <sys/time.h>
 
 #include "logger.h"
@@ -93,6 +94,9 @@ extern const char  *PROGRAM_VERSION;
 extern const char  *PROGRAM_URL;
 extern const char  *PROGRAM_EMAIL;
 
+/* ID of the main thread */
+extern pthread_t main_tid;
+
 extern void	pgut_help(bool details);
 
 /*

src/validate.c

@@ -23,6 +23,12 @@ typedef struct
 {
 	parray *files;
 	bool corrupted;
+
+	/*
+	 * Return value from the thread.
+	 * 0 means there is no error, 1 - there is an error.
+	 */
+	int			ret;
 } validate_files_args;
 
 /*
@@ -35,6 +41,7 @@ pgBackupValidate(pgBackup *backup)
 	char		path[MAXPGPATH];
 	parray	   *files;
 	bool		corrupted = false;
+	bool		validation_isok = true;
 	pthread_t	validate_threads[num_threads];
 	validate_files_args *validate_threads_args[num_threads];
 	int			i;
@@ -79,8 +86,12 @@ pgBackupValidate(pgBackup *backup)
 		validate_files_args *arg = pg_malloc(sizeof(validate_files_args));
 		arg->files = files;
 		arg->corrupted = false;
+		/* By default there are some error */
+		arg->ret = 1;
+
 		validate_threads_args[i] = arg;
-		pthread_create(&validate_threads[i], NULL, (void *(*)(void *)) pgBackupValidateFiles, arg);
+		pthread_create(&validate_threads[i], NULL,
+					   (void *(*)(void *)) pgBackupValidateFiles, arg);
 	}
 
 	/* Wait theads */
@@ -89,8 +100,12 @@ pgBackupValidate(pgBackup *backup)
 		pthread_join(validate_threads[i], NULL);
 		if (validate_threads_args[i]->corrupted)
 			corrupted = true;
+		if (validate_threads_args[i]->ret == 1)
+			validation_isok = false;
 		pg_free(validate_threads_args[i]);
 	}
+	if (!validation_isok)
+		elog(ERROR, "Data files validation failed");
 
 	/* cleanup */
 	parray_walk(files, pgFileFree);
@@ -159,7 +174,7 @@ pgBackupValidateFiles(void *arg)
 				elog(WARNING, "Cannot stat backup file \"%s\": %s",
 					file->path, strerror(errno));
 			arguments->corrupted = true;
-			return;
+			break;
 		}
 
 		if (file->write_size != st.st_size)
@@ -168,7 +183,7 @@ pgBackupValidateFiles(void *arg)
 				 file->path, (unsigned long) file->write_size,
 				 (unsigned long) st.st_size);
 			arguments->corrupted = true;
-			return;
+			break;
 		}
 
 		crc = pgFileGetCRC(file);
@@ -177,9 +192,12 @@ pgBackupValidateFiles(void *arg)
 			elog(WARNING, "Invalid CRC of backup file \"%s\" : %X. Expected %X",
 					file->path, file->crc, crc);
 			arguments->corrupted = true;
-			return;
+			break;
 		}
 	}
+
+	/* Data files validation is successful */
+	arguments->ret = 0;
 }
 
 /*