Skip to content

Commit 6b07b6b

Browse files
knizhnikknizhnik
committed
Fix bug in MMTS
1 parent 99a6f6a commit 6b07b6b

File tree

9 files changed

+92
-173
lines changed

9 files changed

+92
-173
lines changed

‎contrib/mmts/arbiter.c‎

Lines changed: 12 additions & 125 deletions
Original file line numberDiff line numberDiff line change
@@ -101,14 +101,10 @@ static char const* const messageText[] =
101101
{
102102
"INVALID",
103103
"READY",
104-
"BEGIN_PREPARE",
105104
"PREPARE",
106-
"END_PREPARE",
107105
"COMMIT",
108106
"ABORT",
109-
"BEGIN_PREPARED",
110107
"PREPARED",
111-
"END_PREPARED",
112108
"COMMITTED",
113109
"ABORTED"
114110
};
@@ -401,6 +397,7 @@ static void MtmTransSender(Datum arg)
401397

402398
static void MtmWakeUpBackend(MtmTransState* ts)
403399
{
400+
ts->done = true;
404401
SetLatch(&ProcGlobal->allProcs[ts->procno].procLatch);
405402
}
406403

@@ -469,79 +466,6 @@ static void MtmTransReceiver(Datum arg)
469466
Assert((unsigned)(msg->node-1) <= (unsigned)nNodes);
470467
ts->xids[msg->node-1] = msg->sxid;
471468

472-
#ifdef FAST_COMMIT_PROTOCOL
473-
if (MtmIsCoordinator(ts)) {
474-
switch (msg->code) {
475-
case MSG_READY:
476-
Assert(ts->status == TRANSACTION_STATUS_ABORTED || ts->status == TRANSACTION_STATUS_UNKNOWN);
477-
Assert(ts->nVotes < ds->nNodes);
478-
if (msg->csn > ts->csn) {
479-
ts->csn = msg->csn;
480-
MtmSyncClock(ts->csn);
481-
}
482-
if (++ts->nVotes == ds->nNodes) {
483-
/* All nodes are finished their transactions */
484-
if (ts->status == TRANSACTION_STATUS_UNKNOWN) {
485-
ts->nVotes = 1; /* I voted myself */
486-
ts->cmd = MSG_COMMIT;
487-
} else {
488-
ts->status = TRANSACTION_STATUS_ABORTED;
489-
ts->cmd = MSG_ABORT;
490-
MtmWakeUpBackend(ts);
491-
}
492-
MtmAdjustSubtransactions(ts);
493-
MtmSendNotificationMessage(ts);
494-
}
495-
break;
496-
case MSG_COMMITTED:
497-
Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
498-
Assert(ts->nVotes < ds->nNodes);
499-
if (++ts->nVotes == ds->nNodes) {
500-
ts->status = TRANSACTION_STATUS_COMMITTED;
501-
MtmAdjustSubtransactions(ts);
502-
MtmWakeUpBackend(ts);
503-
}
504-
break;
505-
case MSG_ABORTED:
506-
Assert(ts->status == TRANSACTION_STATUS_ABORTED || ts->status == TRANSACTION_STATUS_UNKNOWN);
507-
Assert(ts->nVotes < ds->nNodes);
508-
ts->status = TRANSACTION_STATUS_ABORTED;
509-
if (++ts->nVotes == ds->nNodes) {
510-
ts->cmd = MSG_ABORT;
511-
MtmAdjustSubtransactions(ts);
512-
MtmSendNotificationMessage(ts);
513-
MtmWakeUpBackend(ts);
514-
}
515-
break;
516-
default:
517-
Assert(false);
518-
}
519-
} else { /* replica */
520-
switch (msg->code) {
521-
case MSG_COMMIT:
522-
Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
523-
Assert(ts->csn <= msg->csn);
524-
ts->csn = msg->csn;
525-
MtmSyncClock(ts->csn);
526-
ts->status = TRANSACTION_STATUS_COMMITTED;
527-
ts->cmd = MSG_COMMITTED;
528-
MtmAdjustSubtransactions(ts);
529-
MtmSendNotificationMessage(ts);
530-
MtmWakeUpBackend(ts);
531-
break;
532-
case MSG_ABORT:
533-
if (ts->status != TRANSACTION_STATUS_ABORTED) {
534-
Assert(ts->status == TRANSACTION_STATUS_UNKNOWN || ts->status == TRANSACTION_STATUS_IN_PROGRESS);
535-
ts->status = TRANSACTION_STATUS_ABORTED;
536-
MtmAdjustSubtransactions(ts);
537-
MtmWakeUpBackend(ts);
538-
}
539-
break;
540-
default:
541-
Assert(false);
542-
}
543-
}
544-
#else
545469
if (MtmIsCoordinator(ts)) {
546470
switch (msg->code) {
547471
case MSG_READY:
@@ -551,31 +475,18 @@ static void MtmTransReceiver(Datum arg)
551475
/* All nodes are finished their transactions */
552476
if (ts->status == TRANSACTION_STATUS_IN_PROGRESS) {
553477
ts->nVotes = 1; /* I voted myself */
554-
ts->cmd = MSG_BEGIN_PREPARE;
478+
ts->cmd = MSG_PREPARE;
555479
} else {
556480
ts->status = TRANSACTION_STATUS_ABORTED;
557481
ts->cmd = MSG_ABORT;
482+
MtmAdjustSubtransactions(ts);
558483
MtmWakeUpBackend(ts);
559484
}
560-
MtmAdjustSubtransactions(ts);
561485
MtmSendNotificationMessage(ts);
562486
}
563487
break;
564-
case MSG_BEGIN_PREPARED:
565-
Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
566-
Assert(ts->nVotes < ds->nNodes);
567-
if (++ts->nVotes == ds->nNodes) {
568-
/* All nodes are in in-doubt state */
569-
ts->nVotes = 1; /* I voted myself */
570-
ts->status = TRANSACTION_STATUS_UNKNOWN;
571-
ts->cmd = MSG_PREPARE;
572-
ts->csn = MtmAssignCSN();
573-
MtmAdjustSubtransactions(ts);
574-
MtmSendNotificationMessage(ts);
575-
}
576-
break;
577488
case MSG_PREPARED:
578-
Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
489+
Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
579490
Assert(ts->nVotes < ds->nNodes);
580491
if (msg->csn > ts->csn) {
581492
ts->csn = msg->csn;
@@ -584,27 +495,18 @@ static void MtmTransReceiver(Datum arg)
584495
if (++ts->nVotes == ds->nNodes) {
585496
/* ts->csn is maximum of CSNs at all nodes */
586497
ts->nVotes = 1; /* I voted myself */
587-
ts->cmd = MSG_END_PREPARE;
588-
MtmAdjustSubtransactions(ts);
589-
MtmSendNotificationMessage(ts);
590-
}
591-
break;
592-
case MSG_END_PREPARED:
593-
Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
594-
Assert(ts->nVotes < ds->nNodes);
595-
if (++ts->nVotes == ds->nNodes) {
596-
/* All nodes have now same CSN: do commits */
597-
ts->nVotes = 1; /* I voted myself */
598498
ts->cmd = MSG_COMMIT;
499+
ts->csn = MtmAssignCSN();
500+
ts->status = TRANSACTION_STATUS_UNKNOWN;
501+
MtmAdjustSubtransactions(ts);
599502
MtmSendNotificationMessage(ts);
600503
}
601504
break;
602505
case MSG_COMMITTED:
603506
Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
604507
Assert(ts->nVotes < ds->nNodes);
605508
if (++ts->nVotes == ds->nNodes) {
606-
ts->status = TRANSACTION_STATUS_COMMITTED;
607-
MtmAdjustSubtransactions(ts);
509+
/* All nodes have the same CSN */
608510
MtmWakeUpBackend(ts);
609511
}
610512
break;
@@ -624,33 +526,19 @@ static void MtmTransReceiver(Datum arg)
624526
}
625527
} else { /* replica */
626528
switch (msg->code) {
627-
case MSG_BEGIN_PREPARE:
529+
case MSG_PREPARE:
628530
Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
629531
ts->status = TRANSACTION_STATUS_UNKNOWN;
630-
ts->cmd = MSG_BEGIN_PREPARED;
631-
ts->csn = MtmAssignCSN();
632-
MtmAdjustSubtransactions(ts);
633-
MtmSendNotificationMessage(ts);
634-
break;
635-
case MSG_PREPARE:
636-
Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
637532
ts->csn = MtmAssignCSN();
638533
ts->cmd = MSG_PREPARED;
639534
MtmSendNotificationMessage(ts);
640535
break;
641-
case MSG_END_PREPARE:
642-
Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
643-
Assert(ts->csn <= msg->csn);
644-
ts->csn = msg->csn;
645-
ts->cmd = MSG_END_PREPARED;
646-
MtmAdjustSubtransactions(ts);
647-
MtmSyncClock(ts->csn);
648-
MtmSendNotificationMessage(ts);
649536
break;
650537
case MSG_COMMIT:
651538
Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
652-
Assert(ts->csn == msg->csn);
653-
ts->status = TRANSACTION_STATUS_COMMITTED;
539+
Assert(ts->csn < msg->csn);
540+
ts->csn = msg->csn;
541+
MtmSyncClock(ts->csn);
654542
ts->cmd = MSG_COMMITTED;
655543
MtmAdjustSubtransactions(ts);
656544
MtmSendNotificationMessage(ts);
@@ -668,7 +556,6 @@ static void MtmTransReceiver(Datum arg)
668556
Assert(false);
669557
}
670558
}
671-
#endif
672559
}
673560
MtmUnlock();
674561

‎contrib/mmts/multimaster--1.0.sql‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,19 @@
11
-- complain if script is sourced in psql, rather than via CREATE EXTENSION
22
\echo Use "CREATE EXTENSION multimaster" to load this file. \quit
33

4-
CREATE FUNCTION mm_start_replication() RETURNS void
5-
AS 'MODULE_PATHNAME','mm_start_replication'
4+
CREATE FUNCTION mtm_start_replication() RETURNS void
5+
AS 'MODULE_PATHNAME','mtm_start_replication'
66
LANGUAGE C;
77

8-
CREATE FUNCTION mm_stop_replication() RETURNS void
9-
AS 'MODULE_PATHNAME','mm_stop_replication'
8+
CREATE FUNCTION mtm_stop_replication() RETURNS void
9+
AS 'MODULE_PATHNAME','mtm_stop_replication'
1010
LANGUAGE C;
1111

12-
CREATE FUNCTION mm_drop_node(node integer, drop_slot bool default false) RETURNS void
13-
AS 'MODULE_PATHNAME','mm_drop_node'
12+
CREATE FUNCTION mtm_drop_node(node integer, drop_slot bool default false) RETURNS void
13+
AS 'MODULE_PATHNAME','mtm_drop_node'
14+
LANGUAGE C;
15+
16+
CREATE FUNCTION mtm_get_snapshot() RETURNS bigint
17+
AS 'MODULE_PATHNAME','mtm_get_snapshot'
1418
LANGUAGE C;
1519

0 commit comments

Comments
 (0)