various bugfixes and improvements for archiving and buffering

Author: gsmolk

src/archive.c

@@ -587,14 +587,11 @@ push_file_internal_uncompressed(const char *wal_file_name, const char *pg_xlog_d
 						thread_num, from_fullpath, strerror(errno));
 		}
 
-		if (read_len > 0)
+		if (read_len > 0 && fio_write(out, buf, read_len) != read_len)
 		{
-			if (fio_write(out, buf, read_len) != read_len)
-			{
-				fio_unlink(to_fullpath_part, FIO_BACKUP_HOST);
-				elog(ERROR, "Thread [%d]: Cannot write to destination temp file \"%s\": %s",
-							thread_num, to_fullpath_part, strerror(errno));
-			}
+			fio_unlink(to_fullpath_part, FIO_BACKUP_HOST);
+			elog(ERROR, "Thread [%d]: Cannot write to destination temp file \"%s\": %s",
+						thread_num, to_fullpath_part, strerror(errno));
 		}
 
 		if (feof(in))
@@ -832,14 +829,11 @@ push_file_internal_gz(const char *wal_file_name, const char *pg_xlog_dir,
 								thread_num, from_fullpath, strerror(errno));
 		}
 
-		if (read_len > 0)
+		if (read_len > 0 && fio_gzwrite(out, buf, read_len) != read_len)
 		{
-			if (fio_gzwrite(out, buf, read_len) != read_len)
-			{
-				fio_unlink(to_fullpath_gz_part, FIO_BACKUP_HOST);
-				elog(ERROR, "Thread [%d]: Cannot write to compressed temp WAL file \"%s\": %s",
-							 thread_num, to_fullpath_gz_part, get_gz_error(out, errno));
-			}
+			fio_unlink(to_fullpath_gz_part, FIO_BACKUP_HOST);
+			elog(ERROR, "Thread [%d]: Cannot write to compressed temp WAL file \"%s\": %s",
+						 thread_num, to_fullpath_gz_part, get_gz_error(out, errno));
 		}
 
 		if (feof(in))

src/data.c

@@ -855,7 +855,7 @@ restore_data_file(parray *parent_chain, pgFile *dest_file, FILE *out, const char
 {
 	int    i;
 	size_t total_write_len = 0;
-	char  *in_buf;
+	char  *in_buf = pgut_malloc(STDIO_BUFSIZE);
 
 	for (i = parray_num(parent_chain) - 1; i >= 0; i--)
 	{
@@ -902,7 +902,7 @@ restore_data_file(parray *parent_chain, pgFile *dest_file, FILE *out, const char
 			elog(ERROR, "Cannot open backup file \"%s\": %s", from_fullpath,
 				 strerror(errno));
 
-		in_buf = pgut_malloc(STDIO_BUFSIZE);
+		/* set stdio buffering for input data file */
 		setvbuf(in, in_buf, _IOFBF, STDIO_BUFSIZE);
 
 		/*

src/merge.c

@@ -1132,11 +1132,11 @@ merge_data_file(parray *parent_chain, pgBackup *full_backup,
 				pgBackup *dest_backup, pgFile *dest_file, pgFile *tmp_file,
 				const char *full_database_dir)
 {
-	FILE	*out = NULL;
-	char	to_fullpath[MAXPGPATH];
-	char	to_fullpath_tmp1[MAXPGPATH]; /* used for restore */
-	char	to_fullpath_tmp2[MAXPGPATH]; /* used for backup */
-	char 	buffer[STDIO_BUFSIZE];
+	FILE   *out = NULL;
+	char   *buffer = pgut_malloc(STDIO_BUFSIZE);
+	char    to_fullpath[MAXPGPATH];
+	char    to_fullpath_tmp1[MAXPGPATH]; /* used for restore */
+	char    to_fullpath_tmp2[MAXPGPATH]; /* used for backup */
 
 	/* The next possible optimization is copying "as is" the file
 	 * from intermediate incremental backup, that didn`t changed in
@@ -1158,6 +1158,15 @@ merge_data_file(parray *parent_chain, pgBackup *full_backup,
 	/* restore file into temp file */
 	tmp_file->size = restore_data_file(parent_chain, dest_file, out, to_fullpath_tmp1);
 	fclose(out);
+	pg_free(buffer);
+
+	/* tmp_file->size is greedy, even if there is single 8KB block in file,
+	 * that was overwritten twice during restore_data_file, we would assume that its size is
+	 * 16KB.
+	 * TODO: maybe we should just trust dest_file->n_blocks?
+	 * No, we can`t, because current binary can be used to merge
+	 * 2 backups of old versions, were n_blocks is missing.
+	 */
 
 	backup_data_file(NULL, tmp_file, to_fullpath_tmp1, to_fullpath_tmp2,
 				 InvalidXLogRecPtr, BACKUP_MODE_FULL,

src/pg_probackup.h

@@ -1016,7 +1016,7 @@ extern int fio_send_pages(FILE* in, FILE* out, pgFile *file, XLogRecPtr horizonL
 						   int calg, int clevel, uint32 checksum_version,
 						   datapagemap_t *pagemap, BlockNumber* err_blknum, char **errormsg);
 /* return codes for fio_send_pages */
-#define OUT_BUF_SIZE (1024 * 1024)
+#define OUT_BUF_SIZE (512 * 1024)
 extern int fio_send_file_gz(const char *from_fullpath, const char *to_fullpath, FILE* out, int thread_num);
 extern int fio_send_file(const char *from_fullpath, const char *to_fullpath, FILE* out, int thread_num);
 

src/restore.c

@@ -859,7 +859,7 @@ restore_files(void *arg)
 		/* Restore destination file */
 		if (dest_file->is_datafile && !dest_file->is_cfs)
 		{
-			/* enable stdio buffering for local destination file */
+			/* enable stdio buffering for local destination non-data file */
 			if (!fio_is_remote_file(out))
 				setvbuf(out, out_buf, _IOFBF, STDIO_BUFSIZE);
 			/* Destination file is data file */
@@ -868,7 +868,7 @@ restore_files(void *arg)
 		}
 		else
 		{
-			/* disable stdio buffering for local destination file */
+			/* disable stdio buffering for local destination data file */
 			if (!fio_is_remote_file(out))
 				setvbuf(out, NULL, _IONBF, BUFSIZ);
 			/* Destination file is non-data file */