Fix handlings snapshots in DTMD

knizhnik · knizhnik · commit 9bbb35743b01 · 2015-09-18T12:08:18.000+03:00
diff --git a/contrib/pg_xtm/dtmd/include/snapshot.h b/contrib/pg_xtm/dtmd/include/snapshot.h
@@ -5,9 +5,6 @@
 #include "limits.h"
 
 typedef struct Snapshot {
-	// initially 0, which means 'invalid snapshot'
-	int seqno;
-
 	xid_t xmin;
 	xid_t xmax;
 	int nactive;
diff --git a/contrib/pg_xtm/dtmd/include/transaction.h b/contrib/pg_xtm/dtmd/include/transaction.h
@@ -7,6 +7,8 @@
 #include "snapshot.h"
 #include "limits.h"
 
+#define MAX_SNAPSHOTS_PER_TRANS 8
+
 typedef struct Transaction {
 	// true if the transaction was started on the node
 	bool active;
@@ -15,14 +17,15 @@ typedef struct Transaction {
 	int vote;
 
 	xid_t xid;
-	Snapshot snapshot;
+	Snapshot snapshot[MAX_SNAPSHOTS_PER_TRANS];
 
 	// if this is equal to seqno, we need to generate a new snapshot (for each node)
-	int sent_seqno;
+	int snapshot_no;
 } Transaction;
 
 #define CHAR_TO_INDEX(C) ((C) - 'a')
 typedef struct GlobalTransaction {
+    int n_snapshots;
 	Transaction participants[MAX_NODES];
 	void *listeners[CHAR_TO_INDEX('z')]; // we are going to use 'a' to 'z' for indexing
 } GlobalTransaction;
diff --git a/contrib/pg_xtm/dtmd/src/main.c b/contrib/pg_xtm/dtmd/src/main.c
@@ -152,8 +152,7 @@ static char *onbegin(void *stream, void *clientdata, cmd_t *cmd) {
 		t->node = node;
 		t->vote = DOUBT;
 		t->xid = xid;
-		t->snapshot.seqno = 0;
-		t->sent_seqno = 0;
+		t->snapshot_no = 0;
 
 		if (xid > xmax[node]) {
 			xmax[node] = xid;
@@ -329,14 +328,15 @@ static void gen_snapshot(Snapshot *s, int node) {
 		}
 	}
 	snapshot_sort(s);
-	s->seqno++;
 }
 
 static void gen_snapshots(GlobalTransaction *gt) {
 	int n;
+    assert(gt->n_snapshots < MAX_SNAPSHOTS_PER_TRANS);
 	for (n = 0; n < MAX_NODES; n++) {
-		gen_snapshot(&gt->participants[n].snapshot, n);
+		gen_snapshot(&gt->participants[n].snapshot[gt->n_snapshots], n);
 	}
+    gt->n_snapshots += 1;
 }
 
 static char *onsnapshot(void *stream, void *clientdata, cmd_t *cmd) {
@@ -373,15 +373,14 @@ static char *onsnapshot(void *stream, void *clientdata, cmd_t *cmd) {
 		return strdup("-");
 	}
 
-	GlobalTransaction *gt = transactions + i;
-	Transaction *t = gt->participants + node;
-	if (t->sent_seqno == t->snapshot.seqno) {
+	GlobalTransaction *gt = &transactions[i];
+	Transaction *t = &gt->participants[node];
+	if (t->snapshot_no == gt->n_snapshots) {
 		gen_snapshots(gt);
 	}
-	assert(t->sent_seqno < t->snapshot.seqno);
+	assert(t->snapshot_no < gt->n_snapshots);
 
-	t->sent_seqno++;
-	return snapshot_serialize(&t->snapshot);
+	return snapshot_serialize(&t->snapshot[t->snapshot_no++]);
 }
 
 static bool queue_for_transaction_finish(void *stream, void *clientdata, int node, xid_t xid, char cmd) {
@@ -402,7 +401,7 @@ static bool queue_for_transaction_finish(void *stream, void *clientdata, int nod
 		return false;
 	}
 
-	global_transaction_push_listener(transactions + i, cmd, stream);
+	global_transaction_push_listener(&transactions[i], cmd, stream);
 	return true;
 }
 
diff --git a/contrib/pg_xtm/dtmd/src/snapshot.c b/contrib/pg_xtm/dtmd/src/snapshot.c
@@ -26,8 +26,6 @@ void snapshot_sort(Snapshot *s) {
 }
    
 char *snapshot_serialize(Snapshot *s) {
-	assert(s->seqno > 0);
-
 	int numberlen = 16;
 	int numbers = 3 + s->nactive; // xmin, xmax, n, active...
 	int len = 1 + numberlen * numbers; // +1 for '+'
diff --git a/contrib/pg_xtm/dtmd/src/transaction.c b/contrib/pg_xtm/dtmd/src/transaction.c
@@ -58,6 +58,7 @@ bool global_transaction_mark(clog_t clg, GlobalTransaction *gt, int status) {
 
 void global_transaction_clear(GlobalTransaction *gt) {
 	int i;
+    gt->n_snapshots = 0;
 	for (i = 0; i < MAX_NODES; i++) {
 		gt->participants[i].active = false;
 	}

Original file line number	Diff line number	Diff line change
`@@ -152,8 +152,7 @@ static char onbegin(void stream, void clientdata, cmd_t cmd) {`
`152`	`152`	`t->node = node;`
`153`	`153`	`t->vote = DOUBT;`
`154`	`154`	`t->xid = xid;`
`155`		`- t->snapshot.seqno = 0;`
`156`		`- t->sent_seqno = 0;`
	`155`	`+ t->snapshot_no = 0;`
`157`	`156`
`158`	`157`	`if (xid > xmax[node]) {`
`159`	`158`	`xmax[node] = xid;`
`@@ -329,14 +328,15 @@ static void gen_snapshot(Snapshot *s, int node) {`
`329`	`328`	`}`
`330`	`329`	`}`
`331`	`330`	`snapshot_sort(s);`
`332`		`- s->seqno++;`
`333`	`331`	`}`
`334`	`332`
`335`	`333`	`static void gen_snapshots(GlobalTransaction *gt) {`
`336`	`334`	`int n;`
	`335`	`+ assert(gt->n_snapshots < MAX_SNAPSHOTS_PER_TRANS);`
`337`	`336`	`for (n = 0; n < MAX_NODES; n++) {`
`338`		`- gen_snapshot(&gt->participants[n].snapshot, n);`
	`337`	`+ gen_snapshot(&gt->participants[n].snapshot[gt->n_snapshots], n);`
`339`	`338`	`}`
	`339`	`+ gt->n_snapshots += 1;`
`340`	`340`	`}`
`341`	`341`
`342`	`342`	`static char onsnapshot(void stream, void clientdata, cmd_t cmd) {`
`@@ -373,15 +373,14 @@ static char onsnapshot(void stream, void clientdata, cmd_t cmd) {`
`373`	`373`	`return strdup("-");`
`374`	`374`	`}`
`375`	`375`
`376`		`- GlobalTransaction *gt = transactions + i;`
`377`		`- Transaction *t = gt->participants + node;`
`378`		`- if (t->sent_seqno == t->snapshot.seqno) {`
	`376`	`+ GlobalTransaction *gt = &transactions[i];`
	`377`	`+ Transaction *t = &gt->participants[node];`
	`378`	`+ if (t->snapshot_no == gt->n_snapshots) {`
`379`	`379`	`gen_snapshots(gt);`
`380`	`380`	`}`
`381`		`- assert(t->sent_seqno < t->snapshot.seqno);`
	`381`	`+ assert(t->snapshot_no < gt->n_snapshots);`
`382`	`382`
`383`		`- t->sent_seqno++;`
`384`		`- return snapshot_serialize(&t->snapshot);`
	`383`	`+ return snapshot_serialize(&t->snapshot[t->snapshot_no++]);`
`385`	`384`	`}`
`386`	`385`
`387`	`386`	`static bool queue_for_transaction_finish(void stream, void clientdata, int node, xid_t xid, char cmd) {`
`@@ -402,7 +401,7 @@ static bool queue_for_transaction_finish(void stream, void clientdata, int nod`
`402`	`401`	`return false;`
`403`	`402`	`}`
`404`	`403`
`405`		`- global_transaction_push_listener(transactions + i, cmd, stream);`
	`404`	`+ global_transaction_push_listener(&transactions[i], cmd, stream);`
`406`	`405`	`return true;`
`407`	`406`	`}`
`408`	`407`
Original file line number	Diff line number	Diff line change
`@@ -26,8 +26,6 @@ void snapshot_sort(Snapshot *s) {`
`26`	`26`	`}`
`27`	`27`
`28`	`28`	`char snapshot_serialize(Snapshot s) {`
`29`		`- assert(s->seqno > 0);`
`30`		`-`
`31`	`29`	`int numberlen = 16;`
`32`	`30`	`int numbers = 3 + s->nactive; // xmin, xmax, n, active...`
`33`	`31`	`int len = 1 + numberlen * numbers; // +1 for '+'`
Original file line number	Diff line number	Diff line change
`@@ -58,6 +58,7 @@ bool global_transaction_mark(clog_t clg, GlobalTransaction *gt, int status) {`
`58`	`58`
`59`	`59`	`void global_transaction_clear(GlobalTransaction *gt) {`
`60`	`60`	`int i;`
	`61`	`+ gt->n_snapshots = 0;`
`61`	`62`	`for (i = 0; i < MAX_NODES; i++) {`
`62`	`63`	`gt->participants[i].active = false;`
`63`	`64`	`}`