[Issue #240] Handle SIGPIPE. Also when remote agent encounters an error condition during exectuion of asynchronous tasks, the error message is saved in global variable and reported when agent status is checked

Author: gsmolk

src/archive.c

@@ -421,6 +421,8 @@ push_file_internal_uncompressed(const char *wal_file_name, const char *pg_xlog_d
 	int			partial_try_count = 0;
 	int			partial_file_size = 0;
 	bool		partial_is_stale = true;
+	/* remote agent error message */
+	char       *errmsg = NULL;
 
 	/* from path */
 	join_path_components(from_fullpath, pg_xlog_dir, wal_file_name);
@@ -579,7 +581,7 @@ push_file_internal_uncompressed(const char *wal_file_name, const char *pg_xlog_d
 						from_fullpath, strerror(errno));
 		}
 
-		if (read_len > 0 && fio_write(out, buf, read_len) != read_len)
+		if (read_len > 0 && fio_write_async(out, buf, read_len) != read_len)
 		{
 			fio_unlink(to_fullpath_part, FIO_BACKUP_HOST);
 			elog(ERROR, "Cannot write to destination temp file \"%s\": %s",
@@ -593,6 +595,14 @@ push_file_internal_uncompressed(const char *wal_file_name, const char *pg_xlog_d
 	/* close source file */
 	fclose(in);
 
+	/* Writing is asynchronous in case of push in remote mode, so check agent status */
+	if (fio_check_error_fd(out, &errmsg))
+	{
+		fio_unlink(to_fullpath_part, FIO_BACKUP_HOST);
+		elog(ERROR, "Cannot write to the remote file \"%s\": %s",
+					to_fullpath_part, errmsg);
+	}
+
 	/* close temp file */
 	if (fio_close(out) != 0)
 	{
@@ -652,6 +662,8 @@ push_file_internal_gz(const char *wal_file_name, const char *pg_xlog_dir,
 	int			partial_try_count = 0;
 	int			partial_file_size = 0;
 	bool		partial_is_stale = true;
+	/* remote agent errormsg */
+	char       *errmsg = NULL;
 
 	/* from path */
 	join_path_components(from_fullpath, pg_xlog_dir, wal_file_name);
@@ -804,6 +816,7 @@ push_file_internal_gz(const char *wal_file_name, const char *pg_xlog_dir,
 	}
 
 	/* copy content */
+	/* TODO: move to separate function */
 	for (;;)
 	{
 		size_t  read_len = 0;
@@ -831,7 +844,15 @@ push_file_internal_gz(const char *wal_file_name, const char *pg_xlog_dir,
 	/* close source file */
 	fclose(in);
 
-	/* close temp file */
+	/* Writing is asynchronous in case of push in remote mode, so check agent status */
+	if (fio_check_error_fd_gz(out, &errmsg))
+	{
+		fio_unlink(to_fullpath_gz_part, FIO_BACKUP_HOST);
+		elog(ERROR, "Cannot write to the remote compressed file \"%s\": %s",
+					to_fullpath_gz_part, errmsg);
+	}
+
+	/* close temp file, TODO: make it synchronous */
 	if (fio_gzclose(out) != 0)
 	{
 		fio_unlink(to_fullpath_gz_part, FIO_BACKUP_HOST);

src/data.c

@@ -1084,21 +1084,22 @@ restore_data_file_internal(FILE *in, FILE *out, pgFile *file, uint32 backup_vers
 			cur_pos_out = write_pos;
 		}
 
-		/* If page is compressed and restore is in remote mode, send compressed
-		 * page to the remote side.
+		/*
+		 * If page is compressed and restore is in remote mode,
+		 * send compressed page to the remote side.
 		 */
 		if (is_compressed)
 		{
 			ssize_t rc;
-			rc = fio_fwrite_compressed(out, page.data, compressed_size, file->compress_alg);
+			rc = fio_fwrite_async_compressed(out, page.data, compressed_size, file->compress_alg);
 
 			if (!fio_is_remote_file(out) && rc != BLCKSZ)
 				elog(ERROR, "Cannot write block %u of \"%s\": %s, size: %u",
 					 blknum, to_fullpath, strerror(errno), compressed_size);
 		}
 		else
 		{
-			if (fio_fwrite(out, page.data, BLCKSZ) != BLCKSZ)
+			if (fio_fwrite_async(out, page.data, BLCKSZ) != BLCKSZ)
 				elog(ERROR, "Cannot write block %u of \"%s\": %s",
 					 blknum, to_fullpath, strerror(errno));
 		}

src/pg_probackup.h

@@ -1185,7 +1185,7 @@ extern datapagemap_t *fio_get_lsn_map(const char *fullpath, uint32 checksum_vers
 							fio_location location);
 extern pid_t fio_check_postmaster(const char *pgdata, fio_location location);
 
-extern int32 fio_decompress(void* dst, void const* src, size_t size, int compress_alg);
+extern int32 fio_decompress(void* dst, void const* src, size_t size, int compress_alg, char **errormsg);
 
 /* return codes for fio_send_pages() and fio_send_file() */
 #define SEND_OK       (0)

src/restore.c

@@ -1109,6 +1109,7 @@ restore_files(void *arg)
 		bool     already_exists = false;
 		PageState      *checksum_map = NULL; /* it should take ~1.5MB at most */
 		datapagemap_t  *lsn_map = NULL;      /* it should take 16kB at most */
+		char           *errmsg = NULL;       /* remote agent error message */
 		pgFile	*dest_file = (pgFile *) parray_get(arguments->dest_files, i);
 
 		/* Directories were created before */
@@ -1262,6 +1263,10 @@ restore_files(void *arg)
 		}
 
 done:
+		/* Writing is asynchronous in case of restore in remote mode, so check the agent status */
+		if (fio_check_error_file(out, &errmsg))
+			elog(ERROR, "Cannot write to the remote file \"%s\": %s", to_fullpath, errmsg);
+
 		/* close file */
 		if (fio_fclose(out) != 0)
 			elog(ERROR, "Cannot close file \"%s\": %s", to_fullpath,