Release 2 5 pg stop backup decomposition2 (#387)

* Rename pg_checksum_enable() to pg_is_checksum_enabled
* Remove unused instanceState from pg_start_backup()
* Refactor wait_wal_lsn(): remove unused pgBackup * parameter and replace InstanceState * with simple directory string
* Refactor pg_stop_backup(): remove useless conn variable
* Make some functions and variables (from backup.c) accessible from other compilation units
* Remove some references to global stream_wal variable
* Remove unused variable externaldir
* Yet another split of pg_stop_backup(): separate verification of stop_lsn into wait_wal_and_calculate_stop_lsn()
* Create pfilearray_clear_locks() helper function

Author: kulaginm

src/backup.c

@@ -455,14 +455,15 @@ get_index_list(const char *dbname, bool first_db_with_amcheck,
 		ind->heapallindexed_is_supported = heapallindexed_is_supported;
 		ind->amcheck_nspname = pgut_malloc(strlen(amcheck_nspname) + 1);
 		strcpy(ind->amcheck_nspname, amcheck_nspname);
-		pg_atomic_clear_flag(&ind->lock);
 
 		if (index_list == NULL)
 			index_list = parray_new();
 
 		parray_append(index_list, ind);
 	}
 
+	pfilearray_clear_locks(index_list);
+
 	PQclear(res);
 
 	return index_list;

src/checkdb.c

@@ -222,6 +222,8 @@ pgFileInit(const char *rel_path)
 	/* Number of blocks backed up during backup */
 	file->n_headers = 0;
 
+	// May be add?
+	// pg_atomic_clear_flag(file->lock);
 	return file;
 }
 
@@ -1859,3 +1861,17 @@ cleanup_tablespace(const char *path)
 	parray_walk(files, pgFileFree);
 	parray_free(files);
 }
+
+/*
+ * Clear the synchronisation locks in a parray of (pgFile *)'s
+ */
+void
+pfilearray_clear_locks(parray *file_list)
+{
+	int i;
+	for (i = 0; i < parray_num(file_list); i++)
+	{
+		pgFile *file = (pgFile *) parray_get(file_list, i);
+		pg_atomic_clear_flag(&file->lock);
+	}
+}

src/dir.c

@@ -68,8 +68,6 @@ static char	   *backup_path = NULL;
 static CatalogState *catalogState = NULL;
 /* ================ catalogState (END) =========== */
 
-/* colon separated external directories list ("/path1:/path2") */
-char	   *externaldir = NULL;
 /* common options */
 int			num_threads = 1;
 bool		stream_wal = false;

src/pg_probackup.c

@@ -600,7 +600,6 @@ typedef struct
 	int			ret;
 } backup_files_arg;
 
-
 typedef struct timelineInfo timelineInfo;
 
 /* struct to collect info about timelines in WAL archive */
@@ -679,10 +678,11 @@ typedef struct BackupPageHeader2
 	uint16      checksum;
 } BackupPageHeader2;
 
-typedef struct StopBackupCallbackState {
+typedef struct StopBackupCallbackParams
+{
 	PGconn	*conn;
 	int      server_version;
-} StopBackupCallbackState;
+} StopBackupCallbackParams;
 
 /* Special value for compressed_size field */
 #define PageIsOk		 0
@@ -1061,6 +1061,7 @@ extern int pgFileCompareRelPathWithExternalDesc(const void *f1, const void *f2);
 extern int pgFileCompareLinked(const void *f1, const void *f2);
 extern int pgFileCompareSize(const void *f1, const void *f2);
 extern int pgCompareOid(const void *f1, const void *f2);
+extern void pfilearray_clear_locks(parray *file_list);
 
 /* in data.c */
 extern bool check_data_file(ConnectionArgs *arguments, pgFile *file,
@@ -1259,4 +1260,42 @@ extern void start_WAL_streaming(PGconn *backup_conn, char *stream_dst_path,
 							   ConnectionOptions *conn_opt,
 							   XLogRecPtr startpos, TimeLineID starttli);
 extern int wait_WAL_streaming_end(parray *backup_files_list);
+
+/* external variables and functions, implemented in backup.c */
+typedef struct PGStopBackupResult
+{
+	/*
+	 * We will use values of snapshot_xid and invocation_time if there are
+	 * no transactions between start_lsn and stop_lsn.
+	 */
+	TransactionId	snapshot_xid;
+	time_t		invocation_time;
+	/*
+	 * Fields that store pg_catalog.pg_stop_backup() result
+	 */
+	XLogRecPtr	lsn;
+	size_t		backup_label_content_len;
+	char		*backup_label_content;
+	size_t		tablespace_map_content_len;
+	char		*tablespace_map_content;
+} PGStopBackupResult;
+
+extern bool backup_in_progress;
+extern parray *backup_files_list;
+
+extern void pg_start_backup(const char *label, bool smooth, pgBackup *backup,
+							PGNodeInfo *nodeInfo, PGconn *conn);
+extern void pg_silent_client_messages(PGconn *conn);
+extern void pg_create_restore_point(PGconn *conn, time_t backup_start_time);
+extern void pg_stop_backup_send(PGconn *conn, int server_version, bool is_started_on_replica, bool is_exclusive, char **query_text);
+extern void pg_stop_backup_consume(PGconn *conn, int server_version,
+		bool is_exclusive, uint32 timeout, const char *query_text,
+		PGStopBackupResult *result);
+extern void pg_stop_backup_write_file_helper(const char *path, const char *filename, const char *error_msg_filename,
+		const void *data, size_t len, parray *file_list);
+extern XLogRecPtr wait_wal_lsn(const char *wal_segment_dir, XLogRecPtr lsn, bool is_start_lsn, TimeLineID tli,
+								bool in_prev_segment, bool segment_only,
+								int timeout_elevel, bool in_stream_dir);
+extern void wait_wal_and_calculate_stop_lsn(const char *xlog_path, XLogRecPtr stop_lsn, pgBackup *backup);
+
 #endif /* PG_PROBACKUP_H */

src/pg_probackup.h

@@ -824,7 +824,7 @@ restore_chain(pgBackup *dest_backup, parray *parent_chain,
 	}
 
 	/*
-	 * Setup directory structure for external directories and file locks
+	 * Setup directory structure for external directories
 	 */
 	for (i = 0; i < parray_num(dest_files); i++)
 	{
@@ -848,11 +848,11 @@ restore_chain(pgBackup *dest_backup, parray *parent_chain,
 			elog(VERBOSE, "Create external directory \"%s\"", dirpath);
 			fio_mkdir(dirpath, file->mode, FIO_DB_HOST);
 		}
-
-		/* setup threads */
-		pg_atomic_clear_flag(&file->lock);
 	}
 
+	/* setup threads */
+	pfilearray_clear_locks(dest_files);
+
 	/* Get list of files in destination directory and remove redundant files */
 	if (params->incremental_mode != INCR_NONE || cleanup_pgdata)
 	{

src/restore.c

@@ -175,7 +175,7 @@ parray_rm(parray *array, const void *key, int(*compare)(const void *, const void
 size_t
 parray_num(const parray *array)
 {
-	return array->used;
+	return array!= NULL ? array->used : (size_t) 0;
 }
 
 void

src/utils/parray.c

@@ -130,11 +130,7 @@ pgBackupValidate(pgBackup *backup, pgRestoreParams *params)
 //														params->partial_restore_type);
 
 	/* setup threads */
-	for (i = 0; i < parray_num(files); i++)
-	{
-		pgFile	   *file = (pgFile *) parray_get(files, i);
-		pg_atomic_clear_flag(&file->lock);
-	}
+	pfilearray_clear_locks(files);
 
 	/* init thread args with own file lists */
 	threads = (pthread_t *) palloc(sizeof(pthread_t) * num_threads);