Spoc 384 (second part)

include/spock_group.h

@@ -56,12 +56,12 @@ typedef struct DumpCtx
 
 
 /* Hash Key */
-typedef struct SpockGroupKey
+typedef struct SpockProgressKey
 {
 	Oid			dbid;
 	Oid			node_id;
 	Oid			remote_node_id;
-} SpockGroupKey;
+} SpockProgressKey;
 
 /*
  * Columns for the UI routine get_apply_group_progress.
@@ -84,77 +84,82 @@ typedef enum
 } GroupProgressTupDescColumns;
 
 /*
- * Logical Replication Progress has made by a group of apply workers.
+ * Logical Replication Progress made by a group of apply workers.
  *
+ * IMPORTANT: The 'key' field must remain the first member of this structure
+ * because the hash table uses SpockProgressKey as its hash key type. The
+ * dynahash code accesses the key portion directly, so changing this layout
+ * would break the hash table operations.
+ *
+ * Field descriptions:
+ * key - Identifies the replication group (dbid, node_id, remote_node_id).
+ *       MUST be first field for hash table compatibility.
  * remote_commit_ts - the most advanced timestamp of COMMIT commands, already
- * applied by the replication group. In fact, an apply worker may finish
- * the COMMIT apply if only all other commits with smaller timestamps have
- * already been committed by other workers. So, this value tells us about
- * the real progress.
+ *       applied by the replication group. In fact, an apply worker may finish
+ *       the COMMIT apply if only all other commits with smaller timestamps have
+ *       already been committed by other workers. So, this value tells us about
+ *       the real progress.
  * prev_remote_ts - XXX: It seems do nothing at the moment and should
- * be considered to be removed.
+ *       be considered to be removed.
  * remote_commit_lsn - LSN of the COMMIT corresponding to the remote_commit_ts.
  * remote_insert_lsn - an LSN of the most advanced WAL record written to
- * the WAL on the remote side. Replication protocol attempts to update it as
- * frequently as possible, but it still be a little stale.
+ *       the WAL on the remote side. Replication protocol attempts to update it as
+ *       frequently as possible, but it still be a little stale.
  * received_lsn - an LSN of the most advanced WAL record that was received by
- * the group.
+ *       the group.
  * last_updated_ts - timestamp when remote COMMIT command (identified by the
- * remote_commit_ts and remote_commit_lsn) was applied locally.
- * Spock employs this value to calculate replication_lag.
+ *       remote_commit_ts and remote_commit_lsn) was applied locally.
+ *       Spock employs this value to calculate replication_lag.
  * updated_by_decode - obsolete value. It was needed to decide on the LR lag
- * that seems not needed if we have NULL value for a timestamp column.
+ *       that seems not needed if we have NULL value for a timestamp column.
  */
 typedef struct SpockApplyProgress
 {
-	SpockGroupKey key;			/* common elements */
-	TimestampTz remote_commit_ts;	/* committed remote txn ts */
+	SpockProgressKey	key;			/* MUST be first field */
+
+	TimestampTz			remote_commit_ts;	/* committed remote txn ts */
 
 	/*
 	 * Bit of duplication of remote_commit_ts. Serves the same purpose, except
 	 * keep the last updated value
 	 */
-	TimestampTz prev_remote_ts;
-	XLogRecPtr	remote_commit_lsn;	/* LSN of remote commit on origin */
-	XLogRecPtr	remote_insert_lsn;	/* origin insert/end LSN reported */
+	TimestampTz			prev_remote_ts;
+	XLogRecPtr			remote_commit_lsn;	/* LSN of remote commit on origin */
+	XLogRecPtr			remote_insert_lsn;	/* origin insert/end LSN reported */
 
 	/*
 	 * The largest received LSN by the group. It is more or equal to the
 	 * remote_commit_lsn.
 	 */
-	XLogRecPtr	received_lsn;
+	XLogRecPtr			received_lsn;
 
-	TimestampTz last_updated_ts;	/* when we set this */
-	bool		updated_by_decode;	/* set by decode or apply. OBSOLETE. Used
+	TimestampTz			last_updated_ts;	/* when we set this */
+	bool				updated_by_decode;	/* set by decode or apply. OBSOLETE. Used
 									 * in versions <=5.x.x only */
 } SpockApplyProgress;
 
 /* Hash entry: one per group (stable pointer; not moved by dynahash) */
 typedef struct SpockGroupEntry
 {
-	SpockGroupKey key;			/* hash key */
-	SpockApplyProgress progress;
-	pg_atomic_uint32 nattached;
-	ConditionVariable prev_processed_cv;
+	SpockApplyProgress	progress;
+	pg_atomic_uint32	nattached;
+	ConditionVariable	prev_processed_cv;
 } SpockGroupEntry;
 
 /* shmem setup */
 extern void spock_group_shmem_request(void);
-extern void spock_group_shmem_startup(int napply_groups, bool found);
-
-SpockGroupEntry *spock_group_attach(Oid dbid, Oid node_id, Oid remote_node_id);
-void		spock_group_detach(void);
-bool		spock_group_progress_update(const SpockApplyProgress *sap);
-void		spock_group_progress_update_ptr(SpockGroupEntry *e, const SpockApplyProgress *sap);
-SpockApplyProgress *apply_worker_get_progress(void);
-SpockGroupEntry *spock_group_lookup(Oid dbid, Oid node_id, Oid remote_node_id);
+extern void spock_group_shmem_startup(int napply_groups);
 
-/* Iterate all groups */
-typedef void (*SpockGroupIterCB) (const SpockGroupEntry *e, void *arg);
-void		spock_group_foreach(SpockGroupIterCB cb, void *arg);
+extern SpockGroupEntry *spock_group_attach(Oid dbid, Oid node_id,
+										   Oid remote_node_id);
+extern void spock_group_detach(void);
+extern bool spock_group_progress_update(const SpockApplyProgress *sap);
+extern void spock_group_progress_update_ptr(SpockGroupEntry *entry,
+											const SpockApplyProgress *sap);
+extern SpockApplyProgress *apply_worker_get_progress(void);
 
 extern void spock_group_resource_dump(void);
-extern void spock_group_resource_load(void);
 extern void spock_checkpoint_hook(XLogRecPtr checkPointRedo, int flags);
+extern void spock_group_progress_update_list(List *lst);
 
 #endif							/* SPOCK_GROUP_H */

include/spock_rmgr.h

@@ -47,6 +47,5 @@ extern void spock_rmgr_cleanup(void);
 
 /* WAL helpers */
 extern XLogRecPtr spock_apply_progress_add_to_wal(const SpockApplyProgress *sap);
-extern void spock_group_emit_progress_wal_cb(const SpockGroupEntry *e, void *arg);
 
 #endif							/* SPOCK_RMGR_H */

include/spock_worker.h

@@ -159,7 +159,6 @@ extern void handle_sigterm(SIGNAL_ARGS);
 extern void spock_subscription_changed(Oid subid, bool kill);
 
 extern void spock_worker_shmem_init(void);
-extern void spock_shmem_attach(void);
 
 extern int	spock_worker_register(SpockWorker *worker);
 extern void spock_worker_attach(int slot, SpockWorkerType type);

src/spock_apply.c

@@ -2703,36 +2703,21 @@ send_feedback(PGconn *conn, XLogRecPtr recvpos, int64 now, bool force)
 	return true;
 }
 
-/*
- * Static value to track local progress.
- *
- * Follow the CommitTsShmemInit() code as an example how to initialize such data
- * types in shared memory. Keep the order according to the SpockApplyProgress
- * declaration.
- *
- * NOTE: we initialize empty timestamp with 0 because of multiple places where
- * semantics assumes empty timestamp is 0.
- */
-static SpockApplyProgress apply_progress =
-{
-	.remote_commit_ts = 0,
-	.prev_remote_ts = 0,
-	.remote_commit_lsn = InvalidXLogRecPtr,
-	.remote_insert_lsn = InvalidXLogRecPtr,
-	.received_lsn = InvalidXLogRecPtr,
-	.last_updated_ts = 0,
-	.updated_by_decode = false
-};
-
 /*
  * Update frequently changing statistics of the apply group
  */
 static void
 UpdateWorkerStats(XLogRecPtr last_received, XLogRecPtr last_inserted)
 {
-	apply_progress.received_lsn = last_received;
-	apply_progress.remote_insert_lsn = last_inserted;
-	spock_group_progress_update_ptr(MyApplyWorker->apply_group, &apply_progress);
+	SpockApplyProgress sap = {0};
+
+	sap.key.dbid = MyDatabaseId;
+	sap.key.node_id = MySubscription->target->id;
+	sap.key.remote_node_id = MySubscription->origin->id;
+
+	sap.received_lsn = last_received;
+	sap.remote_insert_lsn = last_inserted;
+	spock_group_progress_update_ptr(MyApplyWorker->apply_group, &sap);
 }
 
 /*
@@ -2768,12 +2753,6 @@ apply_work(PGconn *streamConn)
 
 	MemoryContextSwitchTo(MessageContext);
 
-	/* Initialize our static progress variable */
-	memset(&apply_progress.key, 0, sizeof(SpockGroupKey));
-	apply_progress.key.dbid = MyDatabaseId;
-	apply_progress.key.node_id = MySubscription->target->id;
-	apply_progress.key.remote_node_id = MySubscription->origin->id;
-
 	/* mark as idle, before starting to loop */
 	pgstat_report_activity(STATE_IDLE, NULL);
 

src/spock_functions.c

@@ -3477,35 +3477,36 @@ get_apply_group_progress(PG_FUNCTION_ARGS)
 	hash_seq_init(&it, SpockGroupHash);
 	while ((e = (SpockGroupEntry *) hash_seq_search(&it)) != NULL)
 	{
-		Datum		values[_GP_LAST_];
-		bool		nulls[_GP_LAST_] = {0};
+		SpockApplyProgress *sap = &e->progress;
+		Datum				values[_GP_LAST_];
+		bool				nulls[_GP_LAST_] = {0};
 
 		/*
 		 * Centralise conversion of local representation of the progress data
 		 * to an external representation. This is a good place to check
 		 * correctness of each value (valid oid and timestamps).
 		 */
-		Assert(OidIsValid(e->key.dbid) && OidIsValid(e->key.node_id) &&
-			   OidIsValid(e->key.remote_node_id));
+		Assert(OidIsValid(sap->key.dbid) && OidIsValid(sap->key.node_id) &&
+			   OidIsValid(sap->key.remote_node_id));
 
-		values[GP_DBOID] = ObjectIdGetDatum(e->progress.key.dbid);
-		values[GP_NODE_ID] = ObjectIdGetDatum(e->progress.key.node_id);
-		values[GP_REMOTE_NODE_ID] = ObjectIdGetDatum(e->progress.key.remote_node_id);
+		values[GP_DBOID] = ObjectIdGetDatum(sap->key.dbid);
+		values[GP_NODE_ID] = ObjectIdGetDatum(sap->key.node_id);
+		values[GP_REMOTE_NODE_ID] = ObjectIdGetDatum(sap->key.remote_node_id);
 
-		if (e->progress.remote_commit_ts != 0)
+		if (sap->remote_commit_ts != 0)
 		{
-			Assert(IS_VALID_TIMESTAMP(e->progress.remote_commit_ts));
+			Assert(IS_VALID_TIMESTAMP(sap->remote_commit_ts));
 			values[GP_REMOTE_COMMIT_TS] =
-				TimestampTzGetDatum(e->progress.remote_commit_ts);
+				TimestampTzGetDatum(sap->remote_commit_ts);
 		}
 		else
 			nulls[GP_REMOTE_COMMIT_TS] = true;
 
-		if (e->progress.prev_remote_ts != 0)
+		if (sap->prev_remote_ts != 0)
 		{
-			Assert(IS_VALID_TIMESTAMP(e->progress.prev_remote_ts));
+			Assert(IS_VALID_TIMESTAMP(sap->prev_remote_ts));
 			values[GP_PREV_REMOTE_TS] =
-				TimestampTzGetDatum(e->progress.prev_remote_ts);
+				TimestampTzGetDatum(sap->prev_remote_ts);
 		}
 		else
 			nulls[GP_PREV_REMOTE_TS] = true;
@@ -3515,20 +3516,20 @@ get_apply_group_progress(PG_FUNCTION_ARGS)
 		 * and the current LSN. Moreover, LSN=0 physically makes sense. So,
 		 * don't introduce NULL value for these LSN fields.
 		 */
-		values[GP_REMOTE_COMMIT_LSN] = LSNGetDatum(e->progress.remote_commit_lsn);
-		values[GP_REMOTE_INSERT_LSN] = LSNGetDatum(e->progress.remote_insert_lsn);
-		values[GP_RECEIVED_LSN] = LSNGetDatum(e->progress.received_lsn);
+		values[GP_REMOTE_COMMIT_LSN] = LSNGetDatum(sap->remote_commit_lsn);
+		values[GP_REMOTE_INSERT_LSN] = LSNGetDatum(sap->remote_insert_lsn);
+		values[GP_RECEIVED_LSN] = LSNGetDatum(sap->received_lsn);
 
-		if (e->progress.last_updated_ts != 0)
+		if (sap->last_updated_ts != 0)
 		{
-			Assert(IS_VALID_TIMESTAMP(e->progress.last_updated_ts));
+			Assert(IS_VALID_TIMESTAMP(sap->last_updated_ts));
 			values[GP_LAST_UPDATED_TS] =
-				TimestampTzGetDatum(e->progress.last_updated_ts);
+				TimestampTzGetDatum(sap->last_updated_ts);
 		}
 		else
 			nulls[GP_LAST_UPDATED_TS] = true;
 
-		values[GP_UPDATED_BY_DECODE] = BoolGetDatum(e->progress.updated_by_decode);
+		values[GP_UPDATED_BY_DECODE] = BoolGetDatum(sap->updated_by_decode);
 
 		tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);
 	}