Merge branch 'master' into pgpro_1053_697

Author: gsmolk

src/backup.c

@@ -31,6 +31,7 @@
 
 static int	standby_message_timeout = 10 * 1000;	/* 10 sec = default */
 static XLogRecPtr stop_backup_lsn = InvalidXLogRecPtr;
+static XLogRecPtr stop_stream_lsn = InvalidXLogRecPtr;
 
 /*
  * How long we should wait for streaming end in seconds.
@@ -45,11 +46,23 @@ const char *progname = "pg_probackup";
 /* list of files contained in backup */
 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;
+	PGconn	   *conn;
+
+	/*
+	 * 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 = {"", NULL, 1};
 
 static int is_ptrack_enable = false;
 bool is_ptrack_support = false;
@@ -423,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;
 }
 
 /*
@@ -440,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];
@@ -540,13 +557,40 @@ do_backup_instance(void)
 		join_path_components(dst_backup_path, database_path, PG_XLOG_DIR);
 		dir_create_dir(dst_backup_path, DIR_PERMISSION);
 
-		pthread_mutex_lock(&start_stream_mut);
-		pthread_create(&stream_thread, NULL, (void *(*)(void *)) StreamLog, dst_backup_path);
-		pthread_mutex_lock(&start_stream_mut);
-		if (conn == NULL)
+		stream_thread_arg.basedir = dst_backup_path;
+
+		/*
+		 * Connect in replication mode to the server.
+		 */
+		stream_thread_arg.conn = pgut_connect_replication(pgut_dbname);
+
+		if (!CheckServerVersionForStreaming(stream_thread_arg.conn))
+		{
+			PQfinish(stream_thread_arg.conn);
+			/*
+			 * Error message already written in CheckServerVersionForStreaming().
+			 * There's no hope of recovering from a version mismatch, so don't
+			 * retry.
+			 */
+			elog(ERROR, "Cannot continue backup because stream connect has failed.");
+		}
+
+		/*
+		 * Identify server, obtaining start LSN position and current timeline ID
+		 * at the same time, necessary if not valid data can be found in the
+		 * existing output directory.
+		 */
+		if (!RunIdentifySystem(stream_thread_arg.conn, NULL, NULL, NULL, NULL))
+		{
+			PQfinish(stream_thread_arg.conn);
 			elog(ERROR, "Cannot continue backup because stream connect has failed.");
+		}
+
+		/* By default there are some error */
+		stream_thread_arg.ret = 1;
 
-		pthread_mutex_unlock(&start_stream_mut);
+		pthread_create(&stream_thread, NULL, (void *(*)(void *)) StreamLog,
+					   &stream_thread_arg);
 	}
 
 	/* initialize backup list */
@@ -652,6 +696,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;
 	}
 
@@ -675,9 +721,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)
@@ -776,10 +828,10 @@ do_backup(time_t start_time)
 	is_checksum_enabled = pg_checksum_enable();
 
 	if (is_checksum_enabled)
-		elog(LOG, "This PostgreSQL instance initialized with data block checksums. "
+		elog(LOG, "This PostgreSQL instance was initialized with data block checksums. "
 					"Data block corruption will be detected");
 	else
-		elog(WARNING, "This PostgreSQL instance initialized without data block checksums. "
+		elog(WARNING, "This PostgreSQL instance was initialized without data block checksums. "
 						"pg_probackup have no way to detect data block corruption without them. "
 						"Reinitialize PGDATA with option '--data-checksums'.");
 
@@ -1544,7 +1596,8 @@ pg_stop_backup(pgBackup *backup)
 	FILE		*fp;
 	pgFile		*file;
 	size_t		len;
-	char		*val = NULL;
+	char	   *val = NULL;
+	char	   *stop_backup_query = NULL;
 
 	/*
 	 * We will use this values if there are no transactions between start_lsn
@@ -1601,26 +1654,25 @@ pg_stop_backup(pgBackup *backup)
 			 * pg_stop_backup(false) copy of the backup label and tablespace map
 			 * so they can be written to disk by the caller.
 			 */
-			sent = pgut_send(conn,
-								"SELECT"
+			stop_backup_query = "SELECT"
 								" pg_catalog.txid_snapshot_xmax(pg_catalog.txid_current_snapshot()),"
 								" current_timestamp(0)::timestamptz,"
 								" lsn,"
 								" labelfile,"
 								" spcmapfile"
-								" FROM pg_catalog.pg_stop_backup(false)",
-								0, NULL, WARNING);
+								" FROM pg_catalog.pg_stop_backup(false)";
+
 		}
 		else
 		{
 
-			sent = pgut_send(conn,
-								"SELECT"
+			stop_backup_query =	"SELECT"
 								" pg_catalog.txid_snapshot_xmax(pg_catalog.txid_current_snapshot()),"
 								" current_timestamp(0)::timestamptz,"
-								" pg_catalog.pg_stop_backup() as lsn",
-								0, NULL, WARNING);
+								" pg_catalog.pg_stop_backup() as lsn";
 		}
+
+		sent = pgut_send(conn, stop_backup_query, 0, NULL, WARNING);
 		pg_stop_backup_is_sent = true;
 		if (!sent)
 			elog(ERROR, "Failed to send pg_stop_backup query");
@@ -1665,10 +1717,23 @@ pg_stop_backup(pgBackup *backup)
 				break;
 			}
 		}
+
+		/* Check successfull execution of pg_stop_backup() */
 		if (!res)
 			elog(ERROR, "pg_stop backup() failed");
 		else
+		{
+			switch (PQresultStatus(res))
+			{
+				case PGRES_TUPLES_OK:
+				case PGRES_COMMAND_OK:
+					break;
+				default:
+					elog(ERROR, "query failed: %s query was: %s",
+						 PQerrorMessage(conn), stop_backup_query);
+			}
 			elog(INFO, "pg_stop backup() successfully executed");
+		}
 
 		backup_in_progress = false;
 
@@ -1771,8 +1836,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. */
@@ -1858,7 +1927,7 @@ backup_cleanup(bool fatal, void *userdata)
 	 */
 	if (current.status == BACKUP_STATUS_RUNNING && current.end_time == 0)
 	{
-		elog(INFO, "Backup %s is running, setting its status to ERROR",
+		elog(WARNING, "Backup %s is running, setting its status to ERROR",
 			 base36enc(current.start_time));
 		current.end_time = time(NULL);
 		current.status = BACKUP_STATUS_ERROR;
@@ -1870,7 +1939,7 @@ backup_cleanup(bool fatal, void *userdata)
 	 */
 	if (backup_in_progress)
 	{
-		elog(LOG, "backup in progress, stop backup");
+		elog(WARNING, "backup in progress, stop backup");
 		pg_stop_backup(NULL);	/* don't care stop_lsn on error case */
 	}
 }
@@ -2012,6 +2081,8 @@ backup_files(void *arg)
 	if (arguments->thread_backup_conn)
 		pgut_disconnect(arguments->thread_backup_conn);
 
+	/* Data files transferring is successful */
+	arguments->ret = 0;
 }
 
 /*
@@ -2548,7 +2619,7 @@ stop_streaming(XLogRecPtr xlogpos, uint32 timeline, bool segment_finished)
 
 	/* we assume that we get called once at the end of each segment */
 	if (segment_finished)
-		elog(LOG, _("finished segment at %X/%X (timeline %u)\n"),
+		elog(VERBOSE, _("finished segment at %X/%X (timeline %u)"),
 			 (uint32) (xlogpos >> 32), (uint32) xlogpos, timeline);
 
 	/*
@@ -2566,7 +2637,10 @@ stop_streaming(XLogRecPtr xlogpos, uint32 timeline, bool segment_finished)
 	if (!XLogRecPtrIsInvalid(stop_backup_lsn))
 	{
 		if (xlogpos > stop_backup_lsn)
+		{
+			stop_stream_lsn = xlogpos;
 			return true;
+		}
 
 		/* pg_stop_backup() was executed, wait for the completion of stream */
 		if (stream_stop_timeout == 0)
@@ -2600,45 +2674,13 @@ StreamLog(void *arg)
 {
 	XLogRecPtr	startpos;
 	TimeLineID	starttli;
-	char	   *basedir = (char *)arg;
-
-	/*
-	 * Connect in replication mode to the server
-	 */
-	if (conn == NULL)
-		conn = pgut_connect_replication(pgut_dbname);
-	if (!conn)
-	{
-		pthread_mutex_unlock(&start_stream_mut);
-		/* Error message already written in GetConnection() */
-		return;
-	}
-
-	if (!CheckServerVersionForStreaming(conn))
-	{
-		/*
-		 * Error message already written in CheckServerVersionForStreaming().
-		 * There's no hope of recovering from a version mismatch, so don't
-		 * retry.
-		 */
-		disconnect_and_exit(1);
-	}
-
-	/*
-	 * Identify server, obtaining start LSN position and current timeline ID
-	 * at the same time, necessary if not valid data can be found in the
-	 * existing output directory.
-	 */
-	if (!RunIdentifySystem(conn, NULL, &starttli, &startpos, NULL))
-		disconnect_and_exit(1);
-
-	/* Ok we have normal stream connect and main process can work again */
-	pthread_mutex_unlock(&start_stream_mut);
+	StreamThreadArg *stream_arg = (StreamThreadArg *) arg;
 
 	/*
 	 * We must use startpos as start_lsn from start_backup
 	 */
 	startpos = current.start_lsn;
+	starttli = current.tli;
 
 	/*
 	 * Always start streaming at the beginning of a segment
@@ -2652,7 +2694,7 @@ StreamLog(void *arg)
 	/*
 	 * Start the replication
 	 */
-	elog(LOG, _("starting log streaming at %X/%X (timeline %u)\n"),
+	elog(LOG, _("started streaming WAL at %X/%X (timeline %u)"),
 		 (uint32) (startpos >> 32), (uint32) startpos, starttli);
 
 #if PG_VERSION_NUM >= 90600
@@ -2666,11 +2708,11 @@ 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
-		ctl.basedir = basedir;
+		ctl.basedir = (char *) stream_arg->basedir;
 #endif
 
 		ctl.stream_stop = stop_streaming;
@@ -2679,7 +2721,7 @@ StreamLog(void *arg)
 		ctl.synchronous = false;
 		ctl.mark_done = false;
 
-		if(ReceiveXlogStream(conn, &ctl) == false)
+		if(ReceiveXlogStream(stream_arg->conn, &ctl) == false)
 			elog(ERROR, "Problem in receivexlog");
 
 #if PG_VERSION_NUM >= 100000
@@ -2689,14 +2731,18 @@ StreamLog(void *arg)
 #endif
 	}
 #else
-	if(ReceiveXlogStream(conn, startpos, starttli, NULL, basedir,
-					  stop_streaming, standby_message_timeout, NULL,
-					  false, false) == false)
+	if(ReceiveXlogStream(stream_arg->conn, startpos, starttli, NULL, basedir,
+						 stop_streaming, standby_message_timeout, NULL,
+						 false, false) == false)
 		elog(ERROR, "Problem in receivexlog");
 #endif
 
-	PQfinish(conn);
-	conn = NULL;
+	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(stream_arg->conn);
+	stream_arg->conn = NULL;
 }
 
 /*

src/pg_probackup.c

@@ -321,9 +321,6 @@ main(int argc, char *argv[])
 	if (rc != -1 && !S_ISDIR(stat_buf.st_mode))
 		elog(ERROR, "-B, --backup-path must be a path to directory");
 
-	/* Initialize logger */
-	init_logger(backup_path);
-
 	/* command was initialized for a few commands */
 	if (command)
 	{
@@ -376,6 +373,9 @@ main(int argc, char *argv[])
 		pgut_readopt(path, options, ERROR);
 	}
 
+	/* Initialize logger */
+	init_logger(backup_path);
+
 	/*
 	 * We have read pgdata path from command line or from configuration file.
 	 * Ensure that pgdata is an absolute path.