Wait GC completion in cfs_gc_control

knizhnik · knizhnik · commit a19ee0ae0dce · 2016-11-23T10:19:59.000+03:00
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
@@ -54,6 +54,7 @@
 #include "storage/barrier.h"
 #include "storage/bufmgr.h"
 #include "storage/fd.h"
+#include "storage/cfs.h"
 #include "storage/ipc.h"
 #include "storage/large_object.h"
 #include "storage/latch.h"
@@ -121,6 +122,7 @@ int			CheckPointSegments;
 /* Estimated distance between checkpoints, in bytes */
 static double CheckPointDistanceEstimate = 0;
 static double PrevCheckPointDistance = 0;
+static bool SavedGCState = false;
 
 /*
  * GUC support
@@ -9872,6 +9874,8 @@ do_pg_start_backup(const char *backupidstr, bool fast, TimeLineID *starttli_p,
 	XLogCtl->Insert.forcePageWrites = true;
 	WALInsertLockRelease();
 
+	SavedGCState = cfs_control_gc(false); /* disable GC during backup */
+
 	/* Ensure we release forcePageWrites if fail below */
 	PG_ENSURE_ERROR_CLEANUP(pg_start_backup_callback, (Datum) BoolGetDatum(exclusive));
 	{
@@ -10234,6 +10238,8 @@ pg_start_backup_callback(int code, Datum arg)
 		XLogCtl->Insert.forcePageWrites = false;
 	}
 	WALInsertLockRelease();
+	
+	cfs_control_gc(SavedGCState); /* Restore CFS GC activity */
 }
 
 /*
@@ -10594,6 +10600,8 @@ do_pg_stop_backup(char *labelfile, bool waitforarchive, TimeLineID *stoptli_p)
 		ereport(NOTICE,
 				(errmsg("WAL archiving is not enabled; you must ensure that all required WAL segments are copied through other means to complete the backup")));
 
+	cfs_control_gc(SavedGCState); /* Restore CFS GC activity */
+
 	/*
 	 * We're done.  As a convenience, return the ending WAL location.
 	 */
diff --git a/src/backend/storage/file/cfs.c b/src/backend/storage/file/cfs.c
@@ -281,9 +281,11 @@ void cfs_initialize()
 	cfs_state = (CfsState*)ShmemAlloc(sizeof(CfsState));
 	memset(&cfs_state->gc_stat, 0, sizeof cfs_state->gc_stat);
 	pg_atomic_init_flag(&cfs_state->gc_started);
+	pg_atomic_init_u32(&cfs_state->n_active_gc, 0);
 	cfs_state->n_workers = 0;
 	cfs_state->gc_enabled = true;
 	cfs_state->max_iterations = 0;
+	
 	if (cfs_encryption) { 
 		cfs_rc4_init();
 	}
@@ -470,26 +472,34 @@ static bool cfs_gc_file(char* map_path)
 	uint32 virtSize;
 	int suf = strlen(map_path)-4;
 	int fd = -1, fd2 = -1, md2 = -1;
-	bool succeed = true;
+	bool succeed = false;
+
+	pg_atomic_fetch_add_u32(&cfs_state->n_active_gc, 1);
 
 	while (!cfs_state->gc_enabled) { 
-		int rc = WaitLatch(MyLatch,
-						   WL_TIMEOUT | WL_POSTMASTER_DEATH,
-						   CFS_DISABLE_TIMEOUT /* ms */ );
+		int rc;
+
+		pg_atomic_fetch_sub_u32(&cfs_state->n_active_gc, 1);
+
+		rc = WaitLatch(MyLatch,
+					   WL_TIMEOUT | WL_POSTMASTER_DEATH,
+					   CFS_DISABLE_TIMEOUT /* ms */,
+					   WAIT_EVENT_CFS_GC_ENABLE);
 		if (cfs_stop || (rc & WL_POSTMASTER_DEATH)) {
 			exit(1);
 		}		
 	}
 	if (md < 0) { 
 		elog(LOG, "Failed to open map file %s: %m", map_path);
-		return false;
+		goto FinishGC;
 	}
 	map = cfs_mmap(md);
 	if (map == MAP_FAILED) {
 		elog(LOG, "Failed to map file %s: %m", map_path);
 		close(md);
-		return false;
+		goto FinishGC;
 	}
+	succeed = true;
 	usedSize = pg_atomic_read_u32(&map->usedSize);
 	physSize = pg_atomic_read_u32(&map->physSize);
 	virtSize = pg_atomic_read_u32(&map->virtSize);
@@ -525,7 +535,8 @@ static bool cfs_gc_file(char* map_path)
 				break;
 			}
 			if (cfs_stop) { 
-				return false;
+				succeed = false;
+				goto FinishGC;
 			}
 			if (access_count >= CFS_GC_LOCK) { 
 				/* Uhhh... looks like last GC was interrupted.
@@ -734,6 +745,8 @@ static bool cfs_gc_file(char* map_path)
 		elog(LOG, "Failed to close file %s: %m", map_path);
 		succeed = false;
 	}
+  FinishGC:	
+	pg_atomic_fetch_sub_u32(&cfs_state->n_active_gc, 1);
 	return succeed;
 }
 
@@ -833,6 +846,25 @@ void cfs_start_background_gc()
 	elog(LOG, "Start %d background CFS background workers", i);
 }
 
+bool cfs_control_gc(bool enabled) 
+{ 
+	bool was_enabled = cfs_state->gc_enabled;	
+	cfs_state->gc_enabled = enabled;
+	if (was_enabled && !enabled) { 
+		/* Wait until there are no active GC workers */
+		while (pg_atomic_read_u32(&cfs_state->n_active_gc) != 0) { 
+			int rc = WaitLatch(MyLatch,
+							   WL_TIMEOUT | WL_POSTMASTER_DEATH,
+							   CFS_DISABLE_TIMEOUT /* ms */,
+							   WAIT_EVENT_CFS_GC_ENABLE);
+			if (rc & WL_POSTMASTER_DEATH) {
+				exit(1);
+			}		
+		}
+	}
+	return was_enabled;
+}
+
 PG_MODULE_MAGIC;
 
 PG_FUNCTION_INFO_V1(cfs_start_gc);
@@ -883,11 +915,10 @@ Datum cfs_start_gc(PG_FUNCTION_ARGS)
 	PG_RETURN_INT32(i);
 }
 
+
 Datum cfs_enable_gc(PG_FUNCTION_ARGS)
-{												
-	bool prev = cfs_state->gc_enabled;
-	cfs_state->gc_enabled = PG_GETARG_BOOL(0);
-	PG_RETURN_BOOL(prev);
+{
+	PG_RETURN_BOOL(cfs_control_gc(PG_GETARG_BOOL(0)));											
 }
 
 Datum cfs_version(PG_FUNCTION_ARGS)
diff --git a/src/include/storage/cfs.h b/src/include/storage/cfs.h
@@ -55,6 +55,7 @@ typedef struct
 typedef struct
 {
 	pg_atomic_flag gc_started;
+	pg_atomic_uint32 n_active_gc;
 	int            n_workers;
 	int            max_iterations;
 	bool           gc_enabled;
@@ -77,7 +78,7 @@ void     cfs_lock_file(FileMap* map, char const* path);
 void     cfs_unlock_file(FileMap* map);
 uint32   cfs_alloc_page(FileMap* map, uint32 oldSize, uint32 newSize);
 void     cfs_extend(FileMap* map, uint32 pos);
-
+bool     cfs_control_gc(bool enabled);
 int      cfs_msync(FileMap* map);
 FileMap* cfs_mmap(int md);
 int      cfs_munmap(FileMap* map);
