[Issue #172] Optimize remote backup in PAGE and PTRACK modes

Author: gsmolk

src/data.c

@@ -1693,7 +1693,6 @@ write_database_map(pgBackup *backup, parray *database_map, parray *backup_files_
 	char		database_dir[MAXPGPATH];
 	char		database_map_path[MAXPGPATH];
 
-//	pgBackupGetPath(backup, path, lengthof(path), DATABASE_DIR);
 	join_path_components(database_dir, backup->root_dir, DATABASE_DIR);
 	join_path_components(database_map_path, database_dir, DATABASE_MAP);
 
@@ -1717,7 +1716,7 @@ write_database_map(pgBackup *backup, parray *database_map, parray *backup_files_
 	file->path = pgut_strdup(DATABASE_MAP);
 	file->crc = pgFileGetCRC(database_map_path, true, false);
 
-	file->write_size = file->read_size;
+	file->write_size = file->size;
 	file->uncompressed_size = file->read_size;
 	parray_append(backup_files_list, file);
 }

src/dir.c

@@ -86,7 +86,10 @@ extern const char  *PROGRAM_EMAIL;
 #endif
 
 /* stdio buffer size */
-#define STDIO_BUFSIZE 65536 
+#define STDIO_BUFSIZE 65536
+
+/* retry attempts */
+#define PAGE_READ_ATTEMPTS 100
 
 /* Check if an XLogRecPtr value is pointed to 0 offset */
 #define XRecOffIsNull(xlrp) \
@@ -170,7 +173,8 @@ typedef struct pgFile
 	bool	exists_in_prev;		/* Mark files, both data and regular, that exists in previous backup */
 	CompressAlg		compress_alg;		/* compression algorithm applied to the file */
 	volatile 		pg_atomic_flag lock;/* lock for synchronization of parallel threads  */
-	datapagemap_t	pagemap;			/* bitmap of pages updated since previous backup */
+	datapagemap_t	pagemap;			/* bitmap of pages updated since previous backup
+										   may take up to 16kB per file */
 	bool			pagemap_isabsent;	/* Used to mark files with unknown state of pagemap,
 										 * i.e. datafiles without _ptrack */
 } pgFile;
@@ -526,9 +530,9 @@ typedef struct BackupPageHeader
 
 /* Special value for compressed_size field */
 #define PageIsOk		 0
+#define SkipCurrentPage -1
 #define PageIsTruncated -2
-#define SkipCurrentPage -3
-#define PageIsCorrupted -4 /* used by checkdb */
+#define PageIsCorrupted -3 /* used by checkdb */
 
 
 /*
@@ -722,6 +726,18 @@ extern void help_command(char *command);
 /* in validate.c */
 extern void pgBackupValidate(pgBackup* backup, pgRestoreParams *params);
 extern int do_validate_all(void);
+extern int validate_one_page(Page page, BlockNumber absolute_blkno,
+							 XLogRecPtr stop_lsn, XLogRecPtr *page_lsn,
+							 uint32 checksum_version);
+
+/* return codes for validate_one_page */
+/* TODO: use enum */
+#define PAGE_IS_VALID (-1)
+#define PAGE_IS_NOT_FOUND (-2)
+#define PAGE_IS_ZEROED (-3)
+#define PAGE_HEADER_IS_INVALID (-4)
+#define PAGE_CHECKSUM_MISMATCH (-5)
+#define PAGE_LSN_FROM_FUTURE (-6)
 
 /* in catalog.c */
 extern pgBackup *read_backup(const char *instance_name, time_t timestamp);
@@ -945,4 +961,19 @@ extern char *pg_ptrack_get_and_clear(Oid tablespace_oid,
 extern XLogRecPtr get_last_ptrack_lsn(PGconn *backup_conn, PGNodeInfo *nodeInfo);
 extern parray * pg_ptrack_get_pagemapset(PGconn *backup_conn, const char *ptrack_schema, XLogRecPtr lsn);
 
+/* FIO */
+extern int fio_send_pages(FILE* in, FILE* out, pgFile *file, XLogRecPtr horizonLsn,
+						   int calg, int clevel, uint32 checksum_version,
+						   datapagemap_t *pagemap, BlockNumber* err_blknum, char **errormsg);
+
+/* return codes for fio_send_pages */
+#define WRITE_FAILED (-1)
+#define REMOTE_ERROR (-2)
+#define PAGE_CORRUPTION (-3)
+#define SEND_OK (-4)
+
+extern void get_header_errormsg(Page page, char **errormsg);
+extern void get_checksum_errormsg(Page page, char **errormsg,
+								  BlockNumber absolute_blkno);
+
 #endif /* PG_PROBACKUP_H */