Merge branch 'backup_content_control_updates'

Author: alubennikova

src/backup.c

@@ -182,9 +182,6 @@ do_backup_instance(PGconn *backup_conn)
 		check_external_for_tablespaces(external_dirs, backup_conn);
 	}
 
-	/* Initialize size summary */
-	current.data_bytes = 0;
-
 	/* Obtain current timeline */
 	current.tli = get_current_timeline(false);
 
@@ -445,6 +442,11 @@ do_backup_instance(PGconn *backup_conn)
 	if (prev_backup_filelist)
 		parray_qsort(prev_backup_filelist, pgFileComparePathWithExternal);
 
+	/* write initial backup_content.control file and update backup.control  */
+	write_backup_filelist(&current, backup_files_list,
+						  instance_config.pgdata, external_dirs);
+	write_backup(&current);
+
 	/* init thread args with own file lists */
 	threads = (pthread_t *) palloc(sizeof(pthread_t) * num_threads);
 	threads_args = (backup_files_arg *) palloc(sizeof(backup_files_arg)*num_threads);
@@ -462,6 +464,7 @@ do_backup_instance(PGconn *backup_conn)
 		arg->prev_start_lsn = prev_backup_start_lsn;
 		arg->conn_arg.conn = NULL;
 		arg->conn_arg.cancel_conn = NULL;
+		arg->thread_num = i+1;
 		/* By default there are some error */
 		arg->ret = 1;
 	}
@@ -574,25 +577,14 @@ do_backup_instance(PGconn *backup_conn)
 
 	/* Print the list of files to backup catalog */
 	write_backup_filelist(&current, backup_files_list, instance_config.pgdata,
-						  NULL, external_dirs);
+						  external_dirs);
+	/* update backup control file to update size info */
+	write_backup(&current);
 
 	/* clean external directories list */
 	if (external_dirs)
 		free_dir_list(external_dirs);
 
-	/* Compute summary of size of regular files in the backup */
-	for (i = 0; i < parray_num(backup_files_list); i++)
-	{
-		pgFile	   *file = (pgFile *) parray_get(backup_files_list, i);
-
-		if (S_ISDIR(file->mode))
-			current.data_bytes += 4096;
-
-		/* Count the amount of the data actually copied */
-		if (S_ISREG(file->mode))
-			current.data_bytes += file->write_size;
-	}
-
 	/* Cleanup */
 	if (backup_list)
 	{
@@ -1804,7 +1796,7 @@ pg_stop_backup(pgBackup *backup, PGconn *pg_startbackup_conn)
 			 */
 			if (backup_files_list)
 			{
-				file = pgFileNew(backup_label, backup_label, true, 0,
+				file = pgFileNew(backup_label, PG_BACKUP_LABEL_FILE, true, 0,
 								 FIO_BACKUP_HOST);
 				file->crc = pgFileGetCRC(file->path, true, false,
 										 &file->read_size, FIO_BACKUP_HOST);
@@ -1850,7 +1842,7 @@ pg_stop_backup(pgBackup *backup, PGconn *pg_startbackup_conn)
 
 			if (backup_files_list)
 			{
-				file = pgFileNew(tablespace_map, tablespace_map, true, 0,
+				file = pgFileNew(tablespace_map, PG_TABLESPACE_MAP_FILE, true, 0,
 								 FIO_BACKUP_HOST);
 				if (S_ISREG(file->mode))
 				{
@@ -1987,6 +1979,9 @@ backup_files(void *arg)
 	int			i;
 	backup_files_arg *arguments = (backup_files_arg *) arg;
 	int			n_backup_files_list = parray_num(arguments->files_list);
+	static time_t prev_time;
+
+	prev_time = current.start_time;
 
 	/* backup a file */
 	for (i = 0; i < n_backup_files_list; i++)
@@ -1995,6 +1990,20 @@ backup_files(void *arg)
 		struct stat	buf;
 		pgFile	   *file = (pgFile *) parray_get(arguments->files_list, i);
 
+		if (arguments->thread_num == 1)
+		{
+			/* update backup_content.control every 10 seconds */
+			if ((difftime(time(NULL), prev_time)) > 10)
+			{
+				prev_time = time(NULL);
+
+				write_backup_filelist(&current, arguments->files_list, arguments->from_root,
+									  arguments->external_dirs);
+				/* update backup control file to update size info */
+				write_backup(&current);
+			}
+		}
+
 		if (!pg_atomic_test_set_flag(&file->lock))
 			continue;
 		elog(VERBOSE, "Copying file:  \"%s\" ", file->path);

src/catalog.c

@@ -630,24 +630,104 @@ write_backup(pgBackup *backup)
  */
 void
 write_backup_filelist(pgBackup *backup, parray *files, const char *root,
-					  const char *external_prefix, parray *external_list)
+					  parray *external_list)
 {
-	FILE	   *fp;
+	FILE	   *out;
 	char		path[MAXPGPATH];
 	char		path_temp[MAXPGPATH];
 	int			errno_temp;
+	size_t		i = 0;
+	#define BUFFERSZ BLCKSZ*500
+	char		buf[BUFFERSZ];
+	size_t		write_len = 0;
+	int64 		backup_size_on_disk = 0;
 
 	pgBackupGetPath(backup, path, lengthof(path), DATABASE_FILE_LIST);
 	snprintf(path_temp, sizeof(path_temp), "%s.tmp", path);
 
-	fp = fio_fopen(path_temp, PG_BINARY_W, FIO_BACKUP_HOST);
-	if (fp == NULL)
+	out = fio_fopen(path_temp, PG_BINARY_W, FIO_BACKUP_HOST);
+	if (out == NULL)
 		elog(ERROR, "Cannot open file list \"%s\": %s", path_temp,
 			 strerror(errno));
 
-	print_file_list(fp, files, root, external_prefix, external_list);
+	/* print each file in the list */
+	while(i < parray_num(files))
+	{
+		pgFile	   *file = (pgFile *) parray_get(files, i);
+		char	   *path = file->path; /* for streamed WAL files */
+		char	line[BLCKSZ];
+		int 	len = 0;
 
-	if (fio_fflush(fp) || fio_fclose(fp))
+		i++;
+
+		if (S_ISDIR(file->mode))
+			backup_size_on_disk += 4096;
+
+		/* Count the amount of the data actually copied */
+		if (S_ISREG(file->mode) && file->write_size > 0)
+			backup_size_on_disk += file->write_size;
+
+		/* for files from PGDATA and external files use rel_path
+		 * streamed WAL files has rel_path relative not to "database/"
+		 * but to "database/pg_wal", so for them use path.
+		 */
+		if ((root && strstr(path, root) == path) ||
+			(file->external_dir_num && external_list))
+				path = file->rel_path;
+
+		len = sprintf(line, "{\"path\":\"%s\", \"size\":\"" INT64_FORMAT "\", "
+					 "\"mode\":\"%u\", \"is_datafile\":\"%u\", "
+					 "\"is_cfs\":\"%u\", \"crc\":\"%u\", "
+					 "\"compress_alg\":\"%s\", \"external_dir_num\":\"%d\"",
+					path, file->write_size, file->mode,
+					file->is_datafile ? 1 : 0,
+					file->is_cfs ? 1 : 0,
+					file->crc,
+					deparse_compress_alg(file->compress_alg),
+					file->external_dir_num);
+
+		if (file->is_datafile)
+			len += sprintf(line+len, ",\"segno\":\"%d\"", file->segno);
+
+		if (file->linked)
+			len += sprintf(line+len, ",\"linked\":\"%s\"", file->linked);
+
+		if (file->n_blocks != BLOCKNUM_INVALID)
+			len += sprintf(line+len, ",\"n_blocks\":\"%i\"", file->n_blocks);
+
+		len += sprintf(line+len, "}\n");
+
+		if (write_len + len <= BUFFERSZ)
+		{
+			memcpy(buf+write_len, line, len);
+			write_len += len;
+		}
+		else
+		{
+			/* write buffer to file */
+			if (fio_fwrite(out, buf, write_len) != write_len)
+			{
+				errno_temp = errno;
+				fio_unlink(path_temp, FIO_BACKUP_HOST);
+				elog(ERROR, "Cannot write file list \"%s\": %s",
+					path_temp, strerror(errno));
+			}
+			/* reset write_len */
+			write_len = 0;
+		}
+	}
+
+	/* write what is left in the buffer to file */
+	if (write_len > 0)
+		if (fio_fwrite(out, buf, write_len) != write_len)
+		{
+			errno_temp = errno;
+			fio_unlink(path_temp, FIO_BACKUP_HOST);
+			elog(ERROR, "Cannot write file list \"%s\": %s",
+				path_temp, strerror(errno));
+		}
+
+	if (fio_fflush(out) || fio_fclose(out))
 	{
 		errno_temp = errno;
 		fio_unlink(path_temp, FIO_BACKUP_HOST);
@@ -662,6 +742,9 @@ write_backup_filelist(pgBackup *backup, parray *files, const char *root,
 		elog(ERROR, "Cannot rename configuration file \"%s\" to \"%s\": %s",
 			 path_temp, path, strerror(errno_temp));
 	}
+
+	/* use extra variable to avoid reset of previous data_bytes value in case of error */
+	backup->data_bytes = backup_size_on_disk;
 }
 
 /*

src/dir.c

@@ -1271,52 +1271,6 @@ get_external_remap(char *current_dir)
 	return current_dir;
 }
 
-/*
- * Print backup content list.
- */
-void
-print_file_list(FILE *out, const parray *files, const char *root,
-				const char *external_prefix, parray *external_list)
-{
-	size_t		i;
-
-	/* print each file in the list */
-	for (i = 0; i < parray_num(files); i++)
-	{
-		pgFile	   *file = (pgFile *) parray_get(files, i);
-		char	   *path = file->path;
-
-		/* omit root directory portion */
-		if (root && strstr(path, root) == path)
-			path = GetRelativePath(path, root);
-		else if (file->external_dir_num && !external_prefix)
-		{
-			Assert(external_list);
-			path = GetRelativePath(path, parray_get(external_list,
-													file->external_dir_num - 1));
-		}
-
-		fio_fprintf(out, "{\"path\":\"%s\", \"size\":\"" INT64_FORMAT "\", "
-					 "\"mode\":\"%u\", \"is_datafile\":\"%u\", "
-					 "\"is_cfs\":\"%u\", \"crc\":\"%u\", "
-					 "\"compress_alg\":\"%s\", \"external_dir_num\":\"%d\"",
-				path, file->write_size, file->mode,
-				file->is_datafile ? 1 : 0, file->is_cfs ? 1 : 0, file->crc,
-				deparse_compress_alg(file->compress_alg), file->external_dir_num);
-
-		if (file->is_datafile)
-			fio_fprintf(out, ",\"segno\":\"%d\"", file->segno);
-
-		if (file->linked)
-			fio_fprintf(out, ",\"linked\":\"%s\"", file->linked);
-
-		if (file->n_blocks != BLOCKNUM_INVALID)
-			fio_fprintf(out, ",\"n_blocks\":\"%i\"", file->n_blocks);
-
-		fio_fprintf(out, "}\n");
-	}
-}
-
 /* Parsing states for get_control_value() */
 #define CONTROL_WAIT_NAME			1
 #define CONTROL_INNAME				2

src/merge.c

@@ -368,8 +368,7 @@ merge_backups(pgBackup *to_backup, pgBackup *from_backup)
 	else
 		to_backup->wal_bytes = BYTES_INVALID;
 
-	write_backup_filelist(to_backup, files, from_database_path,
-						  from_external_prefix, NULL);
+	write_backup_filelist(to_backup, files, from_database_path, NULL);
 	write_backup(to_backup);
 
 delete_source_backup:

src/pg_probackup.h

@@ -544,6 +544,7 @@ extern int do_validate_all(void);
 extern pgBackup *read_backup(time_t timestamp);
 extern void write_backup(pgBackup *backup);
 extern void write_backup_status(pgBackup *backup, BackupStatus status);
+extern void write_backup_data_bytes(pgBackup *backup);
 extern bool lock_backup(pgBackup *backup);
 
 extern const char *pgBackupGetBackupMode(pgBackup *backup);
@@ -555,8 +556,7 @@ extern pgBackup *catalog_get_last_data_backup(parray *backup_list,
 											  TimeLineID tli);
 extern void pgBackupWriteControl(FILE *out, pgBackup *backup);
 extern void write_backup_filelist(pgBackup *backup, parray *files,
-								  const char *root, const char *external_prefix,
-								  parray *external_list);
+								  const char *root, parray *external_list);
 
 extern void pgBackupGetPath(const pgBackup *backup, char *path, size_t len,
 							const char *subdir);