agora inbox for [email protected]
help / color / mirror / Atom feedRe: Global snapshots
12+ messages / 8 participants
[nested] [flat]
* Re: Global snapshots
@ 2018-09-24 08:58 Andrey Borodin <[email protected]>
2018-09-26 15:02 ` Re: Global snapshots Arseny Sher <[email protected]>
0 siblings, 1 reply; 12+ messages in thread
From: Andrey Borodin @ 2018-09-24 08:58 UTC (permalink / raw)
To: Arseny Sher <[email protected]>; +Cc: Stas Kelvich <[email protected]>; pgsql-hackers
Hi!
I want to review this patch set. Though I understand that it probably will be quite long process.
I like the idea that with this patch set universally all postgres instances are bound into single distributed DB, even if they never heard about each other before :) This is just amazing. Or do I get something wrong?
I've got few questions:
1. If we coordinate HA-clusters with replicas, can replicas participate if their part of transaction is read-only?
2. How does InDoubt transaction behave when we add or subtract leap seconds?
Also, I could not understand some notes from Arseny:
> 25 июля 2018 г., в 16:35, Arseny Sher <[email protected]> написал(а):
>
> * One drawback of these patches is that only REPEATABLE READ is
> supported. For READ COMMITTED, we must export every new snapshot
> generated on coordinator to all nodes, which is fairly easy to
> do. SERIALIZABLE will definitely require chattering between nodes,
> but that's much less demanded isolevel (e.g. we still don't support
> it on replicas).
If all shards are executing transaction in SERIALIZABLE, what anomalies does it permit?
If you have transactions on server A and server B, there are transactions 1 and 2, transaction A1 is serialized before A2, but B1 is after B2, right?
Maybe we can somehow abort 1 or 2?
>
> * Another somewhat serious issue is that there is a risk of recency
> guarantee violation. If client starts transaction at node with
> lagging clocks, its snapshot might not include some recently
> committed transactions; if client works with different nodes, she
> might not even see her own changes. CockroachDB describes at [1] how
> they and Google Spanner overcome this problem. In short, both set
> hard limit on maximum allowed clock skew. Spanner uses atomic
> clocks, so this skew is small and they just wait it at the end of
> each transaction before acknowledging the client. In CockroachDB, if
> tuple is not visible but we are unsure whether it is truly invisible
> or it's just the skew (the difference between snapshot and tuple's
> csn is less than the skew), transaction is restarted with advanced
> snapshot. This process is not infinite because the upper border
> (initial snapshot + max skew) stays the same; this is correct as we
> just want to ensure that our xact sees all the committed ones before
> it started. We can implement the same thing.
I think that this situation is also covered in Clock-SI since transactions will not exit InDoubt state before we can see them. But I'm not sure, chances are that I get something wrong, I'll think more about it. I'd be happy to hear comments from Stas about this.
>
>
> * 003_bank_shared.pl test is removed. In current shape (loading one
> node) it is useless, and if we bombard both nodes, deadlock surely
> appears. In general, global snaphots are not needed for such
> multimaster-like setup -- either there are no conflicts and we are
> fine, or there is a conflict, in which case we get a deadlock.
Can we do something with this deadlock? Will placing an upper limit on time of InDoubt state fix the issue? I understand that aborting automatically is kind of dangerous...
Also, currently hanging 2pc transaction can cause a lot of headache for DBA. Can we have some kind of protection for the case when one node is gone permanently during transaction?
Thanks!
Best regards, Andrey Borodin.
^ permalink raw reply [nested|flat] 12+ messages in thread
* Re: Global snapshots
2018-09-24 08:58 Re: Global snapshots Andrey Borodin <[email protected]>
@ 2018-09-26 15:02 ` Arseny Sher <[email protected]>
2018-11-29 15:21 ` Re: Global snapshots Dmitry Dolgov <[email protected]>
0 siblings, 1 reply; 12+ messages in thread
From: Arseny Sher @ 2018-09-26 15:02 UTC (permalink / raw)
To: Andrey Borodin <[email protected]>; +Cc: Stas Kelvich <[email protected]>; pgsql-hackers
Hello,
Andrey Borodin <[email protected]> writes:
> I like the idea that with this patch set universally all postgres
> instances are bound into single distributed DB, even if they never
> heard about each other before :) This is just amazing. Or do I get
> something wrong?
Yeah, in a sense of xact visibility we can view it like this.
> I've got few questions:
> 1. If we coordinate HA-clusters with replicas, can replicas
> participate if their part of transaction is read-only?
Ok, there are several things to consider. Clock-SI as described in the
paper technically boils down to three things. First two assume CSN-based
implementation of MVCC where local time acts as CSN/snapshot source, and
they impose the following additional rules:
1) When xact expands on some node and imports its snapshot, it must be
blocked until clocks on this node will show time >= snapshot being
imported: node never processes xacts with snap 'from the future'
2) When we choose CSN to commit xact with, we must read clocks on
all the nodes who participated in it and set CSN to max among read
values.
These rules ensure *integrity* of the snapshot in the face of clock
skews regardless of which node we access: that is, snapshots are stable
(no non-repeatable reads) and no xact is considered half committed: they
prevent situation when the snapshot sees some xact on one node as
committed and on another node as still running.
(Actually, this is only true under the assumption that any distributed
xact is committed at all nodes instantly at the same time; this is
obviously not true, see 3rd point below.)
If we are speaking about e.g. traditional usage of hot standy, when
client in one xact accesses either primary, or some standby, but not
several nodes at once, we just don't need this stuff because usual MVCC
in Postgres already provides you consistent snapshot. Same is true for
multimaster-like setups, when each node accepts writes, but client still
accesses single node; if there is a write conflict (e.g. same row
updated on different nodes), one of xacts must be aborted; the snapshot
is still good.
However, if you really intend to read in one xact data from multiple
nodes (e.g. read primary and then replica), then yes, these problems
arise and Clock-SI helps with them. However, honestly it is hard for me
to make up a reason why would you want to do that: reading local data is
always more efficient than visiting several nodes. It would make sense
if we could read primary and replica in parallel, but that currently is
impossible in core Postgres. More straightforward application of the
patchset is sharding, when data is splitted and you might need to go to
several nodes in one xact to collect needed data.
Also, this patchset adds core algorithm and makes use of it only in
postgres_fdw; you would need to adapt replication (import/export global
snapshot API) to make it work there.
3) The third rule of Clock-SI deals with the following problem.
Distributed (writing to several nodes) xact doesn't commit (i.e.
becomes visible) instantly at all nodes. That means that there is a
time hole in which we can see xact as committed on some node and still
running on another. To mitigate this, Clock-SI adds kind of two-phase
commit on visibility: additional state InDoubt which blocks all
attempts to read this xact changes until xact's fate (commit/abort) is
determined.
Unlike the previous problem, this issue exists in all replicated
setups. Even if we just have primary streaming data to one hot standby,
xacts are not committed on them instantly and we might observe xact as
committed on primary, then quickly switch to standby and find the data
we have just seen disappeared. remote_apply mode partially alleviates
this problem (apparently to the degree comfortable for most application
developers) by switching positions: with it xact always commits on
replicas earlier than on master. At least this guarantees that the guy
who wrote the xact will definitely see it on replica unless it drops the
connection to master before commit ack. Still the problem is not fully
solved: only addition of InDoubt state can fix this.
While Clock-SI (and this patchset) certainly addresses the issue as it
becomes even more serious in sharded setups (it makes possible to see
/parts/ of transactions), there is nothing CSN or clock specific
here. In theory, we could implement the same two-phase commit on
visiblity without switching to timestamp-based CSN MVCC.
Aside from the paper, you can have a look at Clock-SI explanation in
these slides [1] from PGCon.
> 2. How does InDoubt transaction behave when we add or subtract leap seconds?
Good question! In Clock-SI, time can be arbitrary desynchronized and
might go forward with arbitrary speed (e.g. clocks can be stopped), but
it must never go backwards. So if leap second correction is implemented
by doubling the duration of certain second (as it usually seems to be),
we are fine.
> Also, I could not understand some notes from Arseny:
>
>> 25 июля 2018 г., в 16:35, Arseny Sher <[email protected]> написал(а):
>>
>> * One drawback of these patches is that only REPEATABLE READ is
>> supported. For READ COMMITTED, we must export every new snapshot
>> generated on coordinator to all nodes, which is fairly easy to
>> do. SERIALIZABLE will definitely require chattering between nodes,
>> but that's much less demanded isolevel (e.g. we still don't support
>> it on replicas).
>
> If all shards are executing transaction in SERIALIZABLE, what anomalies does it permit?
>
> If you have transactions on server A and server B, there are
> transactions 1 and 2, transaction A1 is serialized before A2, but B1
> is after B2, right?
>
> Maybe we can somehow abort 1 or 2?
Yes, your explanation is concise and correct.
To put it in another way, ensuring SERIALIZABLE in MVCC requires
tracking reads, and there is no infrastructure for doing it
globally. Classical write skew is possible: you have node A holding x
and node B holding y, initially x = y = 30 and there is a constraint x +
y > 0.
Two concurrent xacts start:
T1: x = x - 42;
T2: y = y - 42;
They don't see each other, so both commit successfully and the
constraint is violated. We need to transfer info about reads between
nodes to know when we need to abort someone.
>>
>> * Another somewhat serious issue is that there is a risk of recency
>> guarantee violation. If client starts transaction at node with
>> lagging clocks, its snapshot might not include some recently
>> committed transactions; if client works with different nodes, she
>> might not even see her own changes. CockroachDB describes at [1] how
>> they and Google Spanner overcome this problem. In short, both set
>> hard limit on maximum allowed clock skew. Spanner uses atomic
>> clocks, so this skew is small and they just wait it at the end of
>> each transaction before acknowledging the client. In CockroachDB, if
>> tuple is not visible but we are unsure whether it is truly invisible
>> or it's just the skew (the difference between snapshot and tuple's
>> csn is less than the skew), transaction is restarted with advanced
>> snapshot. This process is not infinite because the upper border
>> (initial snapshot + max skew) stays the same; this is correct as we
>> just want to ensure that our xact sees all the committed ones before
>> it started. We can implement the same thing.
> I think that this situation is also covered in Clock-SI since
> transactions will not exit InDoubt state before we can see them. But
> I'm not sure, chances are that I get something wrong, I'll think more
> about it. I'd be happy to hear comments from Stas about this.
InDoubt state protects from seeing xact who is not yet committed
everywhere, but it doesn't protect from starting xact on node with
lagging clocks, obtaining plainly old snapshot. We won't see any
half-committed data (InDoubt covers us here) with it, but some recently
committed xacts might not get into our old snapshot entirely.
>> * 003_bank_shared.pl test is removed. In current shape (loading one
>> node) it is useless, and if we bombard both nodes, deadlock surely
>> appears. In general, global snaphots are not needed for such
>> multimaster-like setup -- either there are no conflicts and we are
>> fine, or there is a conflict, in which case we get a deadlock.
> Can we do something with this deadlock? Will placing an upper limit on
> time of InDoubt state fix the issue? I understand that aborting
> automatically is kind of dangerous...
Sure, this is just a generalization of basic deadlock problem to
distributed system. To deal with it, someone must periodically collect
locks across the nodes, build graph, check for loops and punish (abort)
one of its chain creators, if loop exists. InDoubt (and global snapshots
in general) is unrelated to this, we hanged on usual row-level lock in
this test. BTW, our pg_shardman extension has primitive deadlock
detector [2], I suppose Citus [3] also has one.
> Also, currently hanging 2pc transaction can cause a lot of headache
> for DBA. Can we have some kind of protection for the case when one
> node is gone permanently during transaction?
Oh, subject of automatic 2PC xacts resolution is also matter of another
(probably many-miles) thread, largely unrelated to global
snapshots/visibility. In general, the problem of distributed transaction
commit which doesn't block while majority of nodes is alive requires
implementing distributed consensus algorithm like Paxos or Raft. You
might also find thread [4] interesting.
Thank you for your interest in the topic!
[1] https://yadi.sk/i/qgmFeICvuRwYNA
[2] https://github.com/postgrespro/pg_shardman/blob/broadcast/pg_shardman--0.0.3.sql#L2497
[3] https://www.citusdata.com/
[4] https://www.postgresql.org/message-id/flat/CAFjFpRc5Eo%3DGqgQBa1F%2B_VQ-q_76B-d5-Pt0DWANT2QS24WE7w%4...
--
Arseny Sher
Postgres Professional: http://www.postgrespro.com
The Russian Postgres Company
^ permalink raw reply [nested|flat] 12+ messages in thread
* Re: Global snapshots
2018-09-24 08:58 Re: Global snapshots Andrey Borodin <[email protected]>
2018-09-26 15:02 ` Re: Global snapshots Arseny Sher <[email protected]>
@ 2018-11-29 15:21 ` Dmitry Dolgov <[email protected]>
2018-11-30 13:00 ` Re: Global snapshots Stas Kelvich <[email protected]>
0 siblings, 1 reply; 12+ messages in thread
From: Dmitry Dolgov @ 2018-11-29 15:21 UTC (permalink / raw)
To: [email protected]; +Cc: [email protected]; Stas Kelvich <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers
> On Wed, Jul 25, 2018 at 1:35 PM Arseny Sher <[email protected]> wrote:
>
> Hello,
>
> I have looked through the patches and found them pretty accurate. I'd
> fixed a lot of small issues here and there; updated patchset is
> attached.
Hi,
Thank you for working on this patch. Unfortunately, the patch has some
conflicts, could you please rebase it? Also I wonder if you or Stas can shed
some light about this:
> On Wed, May 16, 2018 at 2:02 PM Stas Kelvich <[email protected]> wrote:
> > On 15 May 2018, at 15:53, Robert Haas <[email protected]> wrote:
> >
> > I guess it seems to me that you
> > have some further research to do along the lines you've described:
> >
> > 1. Can we hold back xmin only when necessary and to the extent
> > necessary instead of all the time?
> > 2. Can we use something like an STO analog, maybe as an optional
> > feature, rather than actually holding back xmin?
>
> Yes, to both questions. I'll implement that and share results.
Is there any resulting patch where the ideas how to implement this are outlined?
^ permalink raw reply [nested|flat] 12+ messages in thread
* Re: Global snapshots
2018-09-24 08:58 Re: Global snapshots Andrey Borodin <[email protected]>
2018-09-26 15:02 ` Re: Global snapshots Arseny Sher <[email protected]>
2018-11-29 15:21 ` Re: Global snapshots Dmitry Dolgov <[email protected]>
@ 2018-11-30 13:00 ` Stas Kelvich <[email protected]>
2019-01-31 15:42 ` Re: Global snapshots Andres Freund <[email protected]>
0 siblings, 1 reply; 12+ messages in thread
From: Stas Kelvich @ 2018-11-30 13:00 UTC (permalink / raw)
To: Dmitry Dolgov <[email protected]>; +Cc: Arseny Sher <[email protected]>; [email protected]; Robert Haas <[email protected]>; pgsql-hackers
> On 29 Nov 2018, at 18:21, Dmitry Dolgov <[email protected]> wrote:
>
>> On Wed, Jul 25, 2018 at 1:35 PM Arseny Sher <[email protected]> wrote:
>>
>> Hello,
>>
>> I have looked through the patches and found them pretty accurate. I'd
>> fixed a lot of small issues here and there; updated patchset is
>> attached.
>
> Hi,
>
> Thank you for working on this patch. Unfortunately, the patch has some
> conflicts, could you please rebase it?
Rebased onto current master (dcfdf56e89a). Also I corrected few formatting issues
and worked around new pgbench return codes policy in tests.
> Also I wonder if you or Stas can shed
> some light about this:
>
>> On Wed, May 16, 2018 at 2:02 PM Stas Kelvich <[email protected]> wrote:
>>> On 15 May 2018, at 15:53, Robert Haas <[email protected]> wrote:
>>>
>>> I guess it seems to me that you
>>> have some further research to do along the lines you've described:
>>>
>>> 1. Can we hold back xmin only when necessary and to the extent
>>> necessary instead of all the time?
>>> 2. Can we use something like an STO analog, maybe as an optional
>>> feature, rather than actually holding back xmin?
>>
>> Yes, to both questions. I'll implement that and share results.
>
> Is there any resulting patch where the ideas how to implement this are outlined?
Not yet. I’m going to continue work on this in January. And probably try to
force some of nearby committers to make a line by line review.
--
Stas Kelvich
Postgres Professional: http://www.postgrespro.com
Russian Postgres Company
Attachments:
[application/octet-stream] 0001-GlobalCSNLog-SLRU-v3.patch (24.0K, ../../[email protected]/2-0001-GlobalCSNLog-SLRU-v3.patch)
download | inline diff:
From 5b00126f9e73e49859e143ae974e0955299ef674 Mon Sep 17 00:00:00 2001
From: Stas Kelvich <[email protected]>
Date: Wed, 25 Apr 2018 16:05:46 +0300
Subject: [PATCH 1/3] GlobalCSNLog SLRU v3
---
src/backend/access/transam/Makefile | 3 +-
src/backend/access/transam/global_csn_log.c | 439 ++++++++++++++++++++
src/backend/access/transam/twophase.c | 1 +
src/backend/access/transam/varsup.c | 2 +
src/backend/access/transam/xlog.c | 12 +
src/backend/storage/ipc/ipci.c | 3 +
src/backend/storage/ipc/procarray.c | 3 +
src/backend/storage/lmgr/lwlocknames.txt | 1 +
src/backend/utils/misc/guc.c | 9 +
src/backend/utils/probes.d | 2 +
src/bin/initdb/initdb.c | 3 +-
src/include/access/global_csn_log.h | 30 ++
src/include/storage/lwlock.h | 1 +
src/include/utils/snapshot.h | 3 +
14 files changed, 510 insertions(+), 2 deletions(-)
create mode 100644 src/backend/access/transam/global_csn_log.c
create mode 100644 src/include/access/global_csn_log.h
diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefile
index 16fbe47269..03aa360ea3 100644
--- a/src/backend/access/transam/Makefile
+++ b/src/backend/access/transam/Makefile
@@ -12,7 +12,8 @@ subdir = src/backend/access/transam
top_builddir = ../../../..
include $(top_builddir)/src/Makefile.global
-OBJS = clog.o commit_ts.o generic_xlog.o multixact.o parallel.o rmgr.o slru.o \
+OBJS = clog.o commit_ts.o global_csn_log.o generic_xlog.o \
+ multixact.o parallel.o rmgr.o slru.o \
subtrans.o timeline.o transam.o twophase.o twophase_rmgr.o varsup.o \
xact.o xlog.o xlogarchive.o xlogfuncs.o \
xloginsert.o xlogreader.o xlogutils.o
diff --git a/src/backend/access/transam/global_csn_log.c b/src/backend/access/transam/global_csn_log.c
new file mode 100644
index 0000000000..d9d66528e4
--- /dev/null
+++ b/src/backend/access/transam/global_csn_log.c
@@ -0,0 +1,439 @@
+/*-----------------------------------------------------------------------------
+ *
+ * global_csn_log.c
+ * Track global commit sequence numbers of finished transactions
+ *
+ * Implementation of cross-node transaction isolation relies on commit sequence
+ * number (CSN) based visibility rules. This module provides SLRU to store
+ * CSN for each transaction. This mapping need to be kept only for xid's
+ * greater then oldestXid, but that can require arbitrary large amounts of
+ * memory in case of long-lived transactions. Because of same lifetime and
+ * persistancy requirements this module is quite similar to subtrans.c
+ *
+ * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/backend/access/transam/global_csn_log.c
+ *
+ *-----------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/global_csn_log.h"
+#include "access/slru.h"
+#include "access/subtrans.h"
+#include "access/transam.h"
+#include "miscadmin.h"
+#include "pg_trace.h"
+#include "utils/snapmgr.h"
+
+bool track_global_snapshots;
+
+/*
+ * Defines for GlobalCSNLog page sizes. A page is the same BLCKSZ as is used
+ * everywhere else in Postgres.
+ *
+ * Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF,
+ * GlobalCSNLog page numbering also wraps around at
+ * 0xFFFFFFFF/GLOBAL_CSN_LOG_XACTS_PER_PAGE, and GlobalCSNLog segment numbering at
+ * 0xFFFFFFFF/CLOG_XACTS_PER_PAGE/SLRU_PAGES_PER_SEGMENT. We need take no
+ * explicit notice of that fact in this module, except when comparing segment
+ * and page numbers in TruncateGlobalCSNLog (see GlobalCSNLogPagePrecedes).
+ */
+
+/* We store the commit GlobalCSN for each xid */
+#define GCSNLOG_XACTS_PER_PAGE (BLCKSZ / sizeof(GlobalCSN))
+
+#define TransactionIdToPage(xid) ((xid) / (TransactionId) GCSNLOG_XACTS_PER_PAGE)
+#define TransactionIdToPgIndex(xid) ((xid) % (TransactionId) GCSNLOG_XACTS_PER_PAGE)
+
+/*
+ * Link to shared-memory data structures for CLOG control
+ */
+static SlruCtlData GlobalCSNLogCtlData;
+#define GlobalCsnlogCtl (&GlobalCSNLogCtlData)
+
+static int ZeroGlobalCSNLogPage(int pageno);
+static bool GlobalCSNLogPagePrecedes(int page1, int page2);
+static void GlobalCSNLogSetPageStatus(TransactionId xid, int nsubxids,
+ TransactionId *subxids,
+ GlobalCSN csn, int pageno);
+static void GlobalCSNLogSetCSNInSlot(TransactionId xid, GlobalCSN csn,
+ int slotno);
+
+/*
+ * GlobalCSNLogSetCSN
+ *
+ * Record GlobalCSN of transaction and its subtransaction tree.
+ *
+ * xid is a single xid to set status for. This will typically be the top level
+ * transactionid for a top level commit or abort. It can also be a
+ * subtransaction when we record transaction aborts.
+ *
+ * subxids is an array of xids of length nsubxids, representing subtransactions
+ * in the tree of xid. In various cases nsubxids may be zero.
+ *
+ * csn is the commit sequence number of the transaction. It should be
+ * AbortedGlobalCSN for abort cases.
+ */
+void
+GlobalCSNLogSetCSN(TransactionId xid, int nsubxids,
+ TransactionId *subxids, GlobalCSN csn)
+{
+ int pageno;
+ int i = 0;
+ int offset = 0;
+
+ /* Callers of GlobalCSNLogSetCSN() must check GUC params */
+ Assert(track_global_snapshots);
+
+ Assert(TransactionIdIsValid(xid));
+
+ pageno = TransactionIdToPage(xid); /* get page of parent */
+ for (;;)
+ {
+ int num_on_page = 0;
+
+ while (i < nsubxids && TransactionIdToPage(subxids[i]) == pageno)
+ {
+ num_on_page++;
+ i++;
+ }
+
+ GlobalCSNLogSetPageStatus(xid,
+ num_on_page, subxids + offset,
+ csn, pageno);
+ if (i >= nsubxids)
+ break;
+
+ offset = i;
+ pageno = TransactionIdToPage(subxids[offset]);
+ xid = InvalidTransactionId;
+ }
+}
+
+/*
+ * Record the final state of transaction entries in the csn log for
+ * all entries on a single page. Atomic only on this page.
+ *
+ * Otherwise API is same as TransactionIdSetTreeStatus()
+ */
+static void
+GlobalCSNLogSetPageStatus(TransactionId xid, int nsubxids,
+ TransactionId *subxids,
+ GlobalCSN csn, int pageno)
+{
+ int slotno;
+ int i;
+
+ LWLockAcquire(GlobalCSNLogControlLock, LW_EXCLUSIVE);
+
+ slotno = SimpleLruReadPage(GlobalCsnlogCtl, pageno, true, xid);
+
+ /* Subtransactions first, if needed ... */
+ for (i = 0; i < nsubxids; i++)
+ {
+ Assert(GlobalCsnlogCtl->shared->page_number[slotno] == TransactionIdToPage(subxids[i]));
+ GlobalCSNLogSetCSNInSlot(subxids[i], csn, slotno);
+ }
+
+ /* ... then the main transaction */
+ if (TransactionIdIsValid(xid))
+ GlobalCSNLogSetCSNInSlot(xid, csn, slotno);
+
+ GlobalCsnlogCtl->shared->page_dirty[slotno] = true;
+
+ LWLockRelease(GlobalCSNLogControlLock);
+}
+
+/*
+ * Sets the commit status of a single transaction.
+ */
+static void
+GlobalCSNLogSetCSNInSlot(TransactionId xid, GlobalCSN csn, int slotno)
+{
+ int entryno = TransactionIdToPgIndex(xid);
+ GlobalCSN *ptr;
+
+ Assert(LWLockHeldByMe(GlobalCSNLogControlLock));
+
+ ptr = (GlobalCSN *) (GlobalCsnlogCtl->shared->page_buffer[slotno] + entryno * sizeof(XLogRecPtr));
+
+ *ptr = csn;
+}
+
+/*
+ * Interrogate the state of a transaction in the log.
+ *
+ * NB: this is a low-level routine and is NOT the preferred entry point
+ * for most uses; TransactionIdGetGlobalCSN() in global_snapshot.c is the
+ * intended caller.
+ */
+GlobalCSN
+GlobalCSNLogGetCSN(TransactionId xid)
+{
+ int pageno = TransactionIdToPage(xid);
+ int entryno = TransactionIdToPgIndex(xid);
+ int slotno;
+ GlobalCSN *ptr;
+ GlobalCSN global_csn;
+
+ /* Callers of GlobalCSNLogGetCSN() must check GUC params */
+ Assert(track_global_snapshots);
+
+ /* Can't ask about stuff that might not be around anymore */
+ Assert(TransactionIdFollowsOrEquals(xid, TransactionXmin));
+
+ /* lock is acquired by SimpleLruReadPage_ReadOnly */
+
+ slotno = SimpleLruReadPage_ReadOnly(GlobalCsnlogCtl, pageno, xid);
+ ptr = (GlobalCSN *) (GlobalCsnlogCtl->shared->page_buffer[slotno] + entryno * sizeof(XLogRecPtr));
+ global_csn = *ptr;
+
+ LWLockRelease(GlobalCSNLogControlLock);
+
+ return global_csn;
+}
+
+/*
+ * Number of shared GlobalCSNLog buffers.
+ */
+static Size
+GlobalCSNLogShmemBuffers(void)
+{
+ return Min(32, Max(4, NBuffers / 512));
+}
+
+/*
+ * Reserve shared memory for GlobalCsnlogCtl.
+ */
+Size
+GlobalCSNLogShmemSize(void)
+{
+ if (!track_global_snapshots)
+ return 0;
+
+ return SimpleLruShmemSize(GlobalCSNLogShmemBuffers(), 0);
+}
+
+/*
+ * Initialization of shared memory for GlobalCSNLog.
+ */
+void
+GlobalCSNLogShmemInit(void)
+{
+ if (!track_global_snapshots)
+ return;
+
+ GlobalCsnlogCtl->PagePrecedes = GlobalCSNLogPagePrecedes;
+ SimpleLruInit(GlobalCsnlogCtl, "GlobalCSNLog Ctl", GlobalCSNLogShmemBuffers(), 0,
+ GlobalCSNLogControlLock, "pg_global_csn", LWTRANCHE_GLOBAL_CSN_LOG_BUFFERS);
+}
+
+/*
+ * This func must be called ONCE on system install. It creates the initial
+ * GlobalCSNLog segment. The pg_global_csn directory is assumed to have been
+ * created by initdb, and GlobalCSNLogShmemInit must have been called already.
+ */
+void
+BootStrapGlobalCSNLog(void)
+{
+ int slotno;
+
+ if (!track_global_snapshots)
+ return;
+
+ LWLockAcquire(GlobalCSNLogControlLock, LW_EXCLUSIVE);
+
+ /* Create and zero the first page of the commit log */
+ slotno = ZeroGlobalCSNLogPage(0);
+
+ /* Make sure it's written out */
+ SimpleLruWritePage(GlobalCsnlogCtl, slotno);
+ Assert(!GlobalCsnlogCtl->shared->page_dirty[slotno]);
+
+ LWLockRelease(GlobalCSNLogControlLock);
+}
+
+/*
+ * Initialize (or reinitialize) a page of GlobalCSNLog to zeroes.
+ *
+ * The page is not actually written, just set up in shared memory.
+ * The slot number of the new page is returned.
+ *
+ * Control lock must be held at entry, and will be held at exit.
+ */
+static int
+ZeroGlobalCSNLogPage(int pageno)
+{
+ Assert(LWLockHeldByMe(GlobalCSNLogControlLock));
+ return SimpleLruZeroPage(GlobalCsnlogCtl, pageno);
+}
+
+/*
+ * This must be called ONCE during postmaster or standalone-backend startup,
+ * after StartupXLOG has initialized ShmemVariableCache->nextXid.
+ *
+ * oldestActiveXID is the oldest XID of any prepared transaction, or nextXid
+ * if there are none.
+ */
+void
+StartupGlobalCSNLog(TransactionId oldestActiveXID)
+{
+ int startPage;
+ int endPage;
+
+ if (!track_global_snapshots)
+ return;
+
+ /*
+ * Since we don't expect pg_global_csn to be valid across crashes, we
+ * initialize the currently-active page(s) to zeroes during startup.
+ * Whenever we advance into a new page, ExtendGlobalCSNLog will likewise
+ * zero the new page without regard to whatever was previously on disk.
+ */
+ LWLockAcquire(GlobalCSNLogControlLock, LW_EXCLUSIVE);
+
+ startPage = TransactionIdToPage(oldestActiveXID);
+ endPage = TransactionIdToPage(ShmemVariableCache->nextXid);
+
+ while (startPage != endPage)
+ {
+ (void) ZeroGlobalCSNLogPage(startPage);
+ startPage++;
+ /* must account for wraparound */
+ if (startPage > TransactionIdToPage(MaxTransactionId))
+ startPage = 0;
+ }
+ (void) ZeroGlobalCSNLogPage(startPage);
+
+ LWLockRelease(GlobalCSNLogControlLock);
+}
+
+/*
+ * This must be called ONCE during postmaster or standalone-backend shutdown
+ */
+void
+ShutdownGlobalCSNLog(void)
+{
+ if (!track_global_snapshots)
+ return;
+
+ /*
+ * Flush dirty GlobalCSNLog pages to disk.
+ *
+ * This is not actually necessary from a correctness point of view. We do
+ * it merely as a debugging aid.
+ */
+ TRACE_POSTGRESQL_GLOBALCSNLOG_CHECKPOINT_START(false);
+ SimpleLruFlush(GlobalCsnlogCtl, false);
+ TRACE_POSTGRESQL_GLOBALCSNLOG_CHECKPOINT_DONE(false);
+}
+
+/*
+ * Perform a checkpoint --- either during shutdown, or on-the-fly
+ */
+void
+CheckPointGlobalCSNLog(void)
+{
+ if (!track_global_snapshots)
+ return;
+
+ /*
+ * Flush dirty GlobalCSNLog pages to disk.
+ *
+ * This is not actually necessary from a correctness point of view. We do
+ * it merely to improve the odds that writing of dirty pages is done by
+ * the checkpoint process and not by backends.
+ */
+ TRACE_POSTGRESQL_GLOBALCSNLOG_CHECKPOINT_START(true);
+ SimpleLruFlush(GlobalCsnlogCtl, true);
+ TRACE_POSTGRESQL_GLOBALCSNLOG_CHECKPOINT_DONE(true);
+}
+
+/*
+ * Make sure that GlobalCSNLog has room for a newly-allocated XID.
+ *
+ * NB: this is called while holding XidGenLock. We want it to be very fast
+ * most of the time; even when it's not so fast, no actual I/O need happen
+ * unless we're forced to write out a dirty clog or xlog page to make room
+ * in shared memory.
+ */
+void
+ExtendGlobalCSNLog(TransactionId newestXact)
+{
+ int pageno;
+
+ if (!track_global_snapshots)
+ return;
+
+ /*
+ * No work except at first XID of a page. But beware: just after
+ * wraparound, the first XID of page zero is FirstNormalTransactionId.
+ */
+ if (TransactionIdToPgIndex(newestXact) != 0 &&
+ !TransactionIdEquals(newestXact, FirstNormalTransactionId))
+ return;
+
+ pageno = TransactionIdToPage(newestXact);
+
+ LWLockAcquire(GlobalCSNLogControlLock, LW_EXCLUSIVE);
+
+ /* Zero the page and make an XLOG entry about it */
+ ZeroGlobalCSNLogPage(pageno);
+
+ LWLockRelease(GlobalCSNLogControlLock);
+}
+
+/*
+ * Remove all GlobalCSNLog segments before the one holding the passed
+ * transaction ID.
+ *
+ * This is normally called during checkpoint, with oldestXact being the
+ * oldest TransactionXmin of any running transaction.
+ */
+void
+TruncateGlobalCSNLog(TransactionId oldestXact)
+{
+ int cutoffPage;
+
+ if (!track_global_snapshots)
+ return;
+
+ /*
+ * The cutoff point is the start of the segment containing oldestXact. We
+ * pass the *page* containing oldestXact to SimpleLruTruncate. We step
+ * back one transaction to avoid passing a cutoff page that hasn't been
+ * created yet in the rare case that oldestXact would be the first item on
+ * a page and oldestXact == next XID. In that case, if we didn't subtract
+ * one, we'd trigger SimpleLruTruncate's wraparound detection.
+ */
+ TransactionIdRetreat(oldestXact);
+ cutoffPage = TransactionIdToPage(oldestXact);
+
+ SimpleLruTruncate(GlobalCsnlogCtl, cutoffPage);
+}
+
+/*
+ * Decide which of two GlobalCSNLog page numbers is "older" for truncation
+ * purposes.
+ *
+ * We need to use comparison of TransactionIds here in order to do the right
+ * thing with wraparound XID arithmetic. However, if we are asked about
+ * page number zero, we don't want to hand InvalidTransactionId to
+ * TransactionIdPrecedes: it'll get weird about permanent xact IDs. So,
+ * offset both xids by FirstNormalTransactionId to avoid that.
+ */
+static bool
+GlobalCSNLogPagePrecedes(int page1, int page2)
+{
+ TransactionId xid1;
+ TransactionId xid2;
+
+ xid1 = ((TransactionId) page1) * GCSNLOG_XACTS_PER_PAGE;
+ xid1 += FirstNormalTransactionId;
+ xid2 = ((TransactionId) page2) * GCSNLOG_XACTS_PER_PAGE;
+ xid2 += FirstNormalTransactionId;
+
+ return TransactionIdPrecedes(xid1, xid2);
+}
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index e65dccc6a2..96d2d104d9 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -77,6 +77,7 @@
#include <unistd.h>
#include "access/commit_ts.h"
+#include "access/global_csn_log.h"
#include "access/htup_details.h"
#include "access/subtrans.h"
#include "access/transam.h"
diff --git a/src/backend/access/transam/varsup.c b/src/backend/access/transam/varsup.c
index a5eb29e01a..9dc2eb86ff 100644
--- a/src/backend/access/transam/varsup.c
+++ b/src/backend/access/transam/varsup.c
@@ -15,6 +15,7 @@
#include "access/clog.h"
#include "access/commit_ts.h"
+#include "access/global_csn_log.h"
#include "access/subtrans.h"
#include "access/transam.h"
#include "access/xact.h"
@@ -169,6 +170,7 @@ GetNewTransactionId(bool isSubXact)
* Extend pg_subtrans and pg_commit_ts too.
*/
ExtendCLOG(xid);
+ ExtendGlobalCSNLog(xid);
ExtendCommitTs(xid);
ExtendSUBTRANS(xid);
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index c80b14ed97..f26cc952bd 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -24,6 +24,7 @@
#include "access/clog.h"
#include "access/commit_ts.h"
+#include "access/global_csn_log.h"
#include "access/multixact.h"
#include "access/rewriteheap.h"
#include "access/subtrans.h"
@@ -5271,6 +5272,7 @@ BootStrapXLOG(void)
/* Bootstrap the commit log, too */
BootStrapCLOG();
+ BootStrapGlobalCSNLog();
BootStrapCommitTs();
BootStrapSUBTRANS();
BootStrapMultiXact();
@@ -6897,6 +6899,7 @@ StartupXLOG(void)
* maintained during recovery and need not be started yet.
*/
StartupCLOG();
+ StartupGlobalCSNLog(oldestActiveXID);
StartupSUBTRANS(oldestActiveXID);
/*
@@ -7695,6 +7698,7 @@ StartupXLOG(void)
if (standbyState == STANDBY_DISABLED)
{
StartupCLOG();
+ StartupGlobalCSNLog(oldestActiveXID);
StartupSUBTRANS(oldestActiveXID);
}
@@ -8358,6 +8362,7 @@ ShutdownXLOG(int code, Datum arg)
CreateCheckPoint(CHECKPOINT_IS_SHUTDOWN | CHECKPOINT_IMMEDIATE);
}
ShutdownCLOG();
+ ShutdownGlobalCSNLog();
ShutdownCommitTs();
ShutdownSUBTRANS();
ShutdownMultiXact();
@@ -8935,7 +8940,10 @@ CreateCheckPoint(int flags)
* StartupSUBTRANS hasn't been called yet.
*/
if (!RecoveryInProgress())
+ {
TruncateSUBTRANS(GetOldestXmin(NULL, PROCARRAY_FLAGS_DEFAULT));
+ TruncateGlobalCSNLog(GetOldestXmin(NULL, PROCARRAY_FLAGS_DEFAULT));
+ }
/* Real work is done, but log and update stats before releasing lock. */
LogCheckpointEnd(false);
@@ -9011,6 +9019,7 @@ static void
CheckPointGuts(XLogRecPtr checkPointRedo, int flags)
{
CheckPointCLOG();
+ CheckPointGlobalCSNLog();
CheckPointCommitTs();
CheckPointSUBTRANS();
CheckPointMultiXact();
@@ -9294,7 +9303,10 @@ CreateRestartPoint(int flags)
* this because StartupSUBTRANS hasn't been called yet.
*/
if (EnableHotStandby)
+ {
TruncateSUBTRANS(GetOldestXmin(NULL, PROCARRAY_FLAGS_DEFAULT));
+ TruncateGlobalCSNLog(GetOldestXmin(NULL, PROCARRAY_FLAGS_DEFAULT));
+ }
/* Real work is done, but log and update before releasing lock. */
LogCheckpointEnd(true);
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c
index 0c86a581c0..2af468fc6a 100644
--- a/src/backend/storage/ipc/ipci.c
+++ b/src/backend/storage/ipc/ipci.c
@@ -16,6 +16,7 @@
#include "access/clog.h"
#include "access/commit_ts.h"
+#include "access/global_csn_log.h"
#include "access/heapam.h"
#include "access/multixact.h"
#include "access/nbtree.h"
@@ -127,6 +128,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
size = add_size(size, ProcGlobalShmemSize());
size = add_size(size, XLOGShmemSize());
size = add_size(size, CLOGShmemSize());
+ size = add_size(size, GlobalCSNLogShmemSize());
size = add_size(size, CommitTsShmemSize());
size = add_size(size, SUBTRANSShmemSize());
size = add_size(size, TwoPhaseShmemSize());
@@ -219,6 +221,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
*/
XLOGShmemInit();
CLOGShmemInit();
+ GlobalCSNLogShmemInit();
CommitTsShmemInit();
SUBTRANSShmemInit();
MultiXactShmemInit();
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index dc7e875680..6deb810969 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -46,6 +46,7 @@
#include <signal.h>
#include "access/clog.h"
+#include "access/global_csn_log.h"
#include "access/subtrans.h"
#include "access/transam.h"
#include "access/twophase.h"
@@ -836,6 +837,7 @@ ProcArrayApplyRecoveryInfo(RunningTransactions running)
while (TransactionIdPrecedes(latestObservedXid, running->nextXid))
{
ExtendSUBTRANS(latestObservedXid);
+ ExtendGlobalCSNLog(latestObservedXid);
TransactionIdAdvance(latestObservedXid);
}
TransactionIdRetreat(latestObservedXid); /* = running->nextXid - 1 */
@@ -3240,6 +3242,7 @@ RecordKnownAssignedTransactionIds(TransactionId xid)
while (TransactionIdPrecedes(next_expected_xid, xid))
{
TransactionIdAdvance(next_expected_xid);
+ ExtendGlobalCSNLog(next_expected_xid);
ExtendSUBTRANS(next_expected_xid);
}
Assert(next_expected_xid == xid);
diff --git a/src/backend/storage/lmgr/lwlocknames.txt b/src/backend/storage/lmgr/lwlocknames.txt
index e6025ecedb..9615058f29 100644
--- a/src/backend/storage/lmgr/lwlocknames.txt
+++ b/src/backend/storage/lmgr/lwlocknames.txt
@@ -50,3 +50,4 @@ OldSnapshotTimeMapLock 42
BackendRandomLock 43
LogicalRepWorkerLock 44
CLogTruncationLock 45
+GlobalCSNLogControlLock 46
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 03594e77fe..0eb04c35cb 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -1053,6 +1053,15 @@ static struct config_bool ConfigureNamesBool[] =
false,
NULL, NULL, NULL
},
+ {
+ {"track_global_snapshots", PGC_POSTMASTER, RESOURCES_MEM,
+ gettext_noop("Enable global snapshot tracking."),
+ gettext_noop("Used to achieve REPEATEBLE READ isolation level for postgres_fdw transactions.")
+ },
+ &track_global_snapshots,
+ true, /* XXX: set true to simplify tesing. XXX2: Seems that RESOURCES_MEM isn't the best catagory */
+ NULL, NULL, NULL
+ },
{
{"ssl", PGC_SIGHUP, CONN_AUTH_SSL,
gettext_noop("Enables SSL connections."),
diff --git a/src/backend/utils/probes.d b/src/backend/utils/probes.d
index ad06e8e2ea..5ebe2ad888 100644
--- a/src/backend/utils/probes.d
+++ b/src/backend/utils/probes.d
@@ -77,6 +77,8 @@ provider postgresql {
probe clog__checkpoint__done(bool);
probe subtrans__checkpoint__start(bool);
probe subtrans__checkpoint__done(bool);
+ probe globalcsnlog__checkpoint__start(bool);
+ probe globalcsnlog__checkpoint__done(bool);
probe multixact__checkpoint__start(bool);
probe multixact__checkpoint__done(bool);
probe twophase__checkpoint__start();
diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c
index 211a96380e..33e16c6870 100644
--- a/src/bin/initdb/initdb.c
+++ b/src/bin/initdb/initdb.c
@@ -221,7 +221,8 @@ static const char *const subdirs[] = {
"pg_xact",
"pg_logical",
"pg_logical/snapshots",
- "pg_logical/mappings"
+ "pg_logical/mappings",
+ "pg_global_csn"
};
diff --git a/src/include/access/global_csn_log.h b/src/include/access/global_csn_log.h
new file mode 100644
index 0000000000..417c26c8a3
--- /dev/null
+++ b/src/include/access/global_csn_log.h
@@ -0,0 +1,30 @@
+/*
+ * global_csn_log.h
+ *
+ * Commit-Sequence-Number log.
+ *
+ * Portions Copyright (c) 1996-2014, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/access/global_csn_log.h
+ */
+#ifndef CSNLOG_H
+#define CSNLOG_H
+
+#include "access/xlog.h"
+#include "utils/snapshot.h"
+
+extern void GlobalCSNLogSetCSN(TransactionId xid, int nsubxids,
+ TransactionId *subxids, GlobalCSN csn);
+extern GlobalCSN GlobalCSNLogGetCSN(TransactionId xid);
+
+extern Size GlobalCSNLogShmemSize(void);
+extern void GlobalCSNLogShmemInit(void);
+extern void BootStrapGlobalCSNLog(void);
+extern void StartupGlobalCSNLog(TransactionId oldestActiveXID);
+extern void ShutdownGlobalCSNLog(void);
+extern void CheckPointGlobalCSNLog(void);
+extern void ExtendGlobalCSNLog(TransactionId newestXact);
+extern void TruncateGlobalCSNLog(TransactionId oldestXact);
+
+#endif /* CSNLOG_H */
\ No newline at end of file
diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.h
index b2dcb73287..4b978f65b5 100644
--- a/src/include/storage/lwlock.h
+++ b/src/include/storage/lwlock.h
@@ -198,6 +198,7 @@ typedef enum BuiltinTrancheIds
LWTRANCHE_CLOG_BUFFERS = NUM_INDIVIDUAL_LWLOCKS,
LWTRANCHE_COMMITTS_BUFFERS,
LWTRANCHE_SUBTRANS_BUFFERS,
+ LWTRANCHE_GLOBAL_CSN_LOG_BUFFERS,
LWTRANCHE_MXACTOFFSET_BUFFERS,
LWTRANCHE_MXACTMEMBER_BUFFERS,
LWTRANCHE_ASYNC_BUFFERS,
diff --git a/src/include/utils/snapshot.h b/src/include/utils/snapshot.h
index a8a5a8f4c0..318d41e6f7 100644
--- a/src/include/utils/snapshot.h
+++ b/src/include/utils/snapshot.h
@@ -24,6 +24,9 @@ typedef struct SnapshotData *Snapshot;
#define InvalidSnapshot ((Snapshot) NULL)
+typedef uint64 GlobalCSN;
+extern bool track_global_snapshots;
+
/*
* We use SnapshotData structures to represent both "regular" (MVCC)
* snapshots and "special" snapshots that have non-MVCC semantics.
--
2.19.2
[application/octet-stream] 0002-Global-snapshots-v3.patch (64.2K, ../../[email protected]/3-0002-Global-snapshots-v3.patch)
download | inline diff:
From b35a69befd92c8ecab5322a68fa8afb0413bb43f Mon Sep 17 00:00:00 2001
From: Stas Kelvich <[email protected]>
Date: Wed, 25 Apr 2018 16:22:30 +0300
Subject: [PATCH 2/3] Global snapshots v3
---
src/backend/access/transam/Makefile | 2 +-
src/backend/access/transam/global_snapshot.c | 754 ++++++++++++++++++
src/backend/access/transam/twophase.c | 156 ++++
src/backend/access/transam/xact.c | 29 +
src/backend/access/transam/xlog.c | 2 +
src/backend/storage/ipc/ipci.c | 3 +
src/backend/storage/ipc/procarray.c | 92 ++-
src/backend/storage/lmgr/lwlocknames.txt | 1 +
src/backend/storage/lmgr/proc.c | 5 +
src/backend/utils/misc/guc.c | 13 +-
src/backend/utils/misc/postgresql.conf.sample | 2 +
src/backend/utils/time/snapmgr.c | 103 +++
src/backend/utils/time/tqual.c | 65 +-
src/include/access/global_snapshot.h | 72 ++
src/include/access/twophase.h | 1 +
src/include/catalog/pg_proc.dat | 14 +
src/include/datatype/timestamp.h | 3 +
src/include/fmgr.h | 1 +
src/include/portability/instr_time.h | 10 +
src/include/storage/proc.h | 15 +
src/include/storage/procarray.h | 8 +
src/include/utils/snapmgr.h | 3 +
src/include/utils/snapshot.h | 8 +
23 files changed, 1354 insertions(+), 8 deletions(-)
create mode 100644 src/backend/access/transam/global_snapshot.c
create mode 100644 src/include/access/global_snapshot.h
diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefile
index 03aa360ea3..8ef677cada 100644
--- a/src/backend/access/transam/Makefile
+++ b/src/backend/access/transam/Makefile
@@ -12,7 +12,7 @@ subdir = src/backend/access/transam
top_builddir = ../../../..
include $(top_builddir)/src/Makefile.global
-OBJS = clog.o commit_ts.o global_csn_log.o generic_xlog.o \
+OBJS = clog.o commit_ts.o global_csn_log.o global_snapshot.o generic_xlog.o \
multixact.o parallel.o rmgr.o slru.o \
subtrans.o timeline.o transam.o twophase.o twophase_rmgr.o varsup.o \
xact.o xlog.o xlogarchive.o xlogfuncs.o \
diff --git a/src/backend/access/transam/global_snapshot.c b/src/backend/access/transam/global_snapshot.c
new file mode 100644
index 0000000000..16b3943a76
--- /dev/null
+++ b/src/backend/access/transam/global_snapshot.c
@@ -0,0 +1,754 @@
+/*-------------------------------------------------------------------------
+ *
+ * global_snapshot.c
+ * Support for cross-node snapshot isolation.
+ *
+ * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/backend/access/transam/global_snapshot.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/global_csn_log.h"
+#include "access/global_snapshot.h"
+#include "access/transam.h"
+#include "access/twophase.h"
+#include "access/xact.h"
+#include "portability/instr_time.h"
+#include "storage/lmgr.h"
+#include "storage/proc.h"
+#include "storage/procarray.h"
+#include "storage/shmem.h"
+#include "storage/spin.h"
+#include "utils/builtins.h"
+#include "utils/guc.h"
+#include "utils/snapmgr.h"
+#include "miscadmin.h"
+
+/* Raise a warning if imported global_csn exceeds ours by this value. */
+#define SNAP_DESYNC_COMPLAIN (1*NSECS_PER_SEC) /* 1 second */
+
+/*
+ * GlobalSnapshotState
+ *
+ * Do not trust local clocks to be strictly monotonical and save last acquired
+ * value so later we can compare next timestamp with it. Accessed through
+ * GlobalSnapshotGenerate() and GlobalSnapshotSync().
+ */
+typedef struct
+{
+ GlobalCSN last_global_csn;
+ volatile slock_t lock;
+} GlobalSnapshotState;
+
+static GlobalSnapshotState *gsState;
+
+
+/*
+ * GUC to delay advance of oldestXid for this amount of time. Also determines
+ * the size GlobalSnapshotXidMap circular buffer.
+ */
+int global_snapshot_defer_time;
+
+/*
+ * Enables this module.
+ */
+extern bool track_global_snapshots;
+
+/*
+ * GlobalSnapshotXidMap
+ *
+ * To be able to install global snapshot that points to past we need to keep
+ * old versions of tuples and therefore delay advance of oldestXid. Here we
+ * keep track of correspondence between snapshot's global_csn and oldestXid
+ * that was set at the time when the snapshot was taken. Much like the
+ * snapshot too old's OldSnapshotControlData does, but with finer granularity
+ * to seconds.
+ *
+ * Different strategies can be employed to hold oldestXid (e.g. we can track
+ * oldest global_csn-based snapshot among cluster nodes and map it oldestXid
+ * on each node) but here implemented one that tries to avoid cross-node
+ * communications which are tricky in case of postgres_fdw.
+ *
+ * On each snapshot acquisition GlobalSnapshotMapXmin() is called and stores
+ * correspondence between current global_csn and oldestXmin in a sparse way:
+ * global_csn is rounded to seconds (and here we use the fact that global_csn
+ * is just a timestamp) and oldestXmin is stored in the circular buffer where
+ * rounded global_csn acts as an offset from current circular buffer head.
+ * Size of the circular buffer is controlled by global_snapshot_defer_time GUC.
+ *
+ * When global snapshot arrives from different node we check that its
+ * global_csn is still in our map, otherwise we'll error out with "snapshot too
+ * old" message. If global_csn is successfully mapped to oldestXid we move
+ * backend's pgxact->xmin to proc->originalXmin and fill pgxact->xmin to
+ * mapped oldestXid. That way GetOldestXmin() can take into account backends
+ * with imported global snapshot and old tuple versions will be preserved.
+ *
+ * Also while calculating oldestXmin for our map in presence of imported
+ * global snapshots we should use proc->originalXmin instead of pgxact->xmin
+ * that was set during import. Otherwise, we can create a feedback loop:
+ * xmin's of imported global snapshots were calculated using our map and new
+ * entries in map going to be calculated based on that xmin's, and there is
+ * a risk to stuck forever with one non-increasing oldestXmin. All other
+ * callers of GetOldestXmin() are using pgxact->xmin so the old tuple versions
+ * are preserved.
+ */
+typedef struct GlobalSnapshotXidMap
+{
+ int head; /* offset of current freshest value */
+ int size; /* total size of circular buffer */
+ GlobalCSN_atomic last_csn_seconds; /* last rounded global_csn that changed
+ * xmin_by_second[] */
+ TransactionId *xmin_by_second; /* circular buffer of oldestXmin's */
+}
+GlobalSnapshotXidMap;
+
+static GlobalSnapshotXidMap *gsXidMap;
+
+
+/* Estimate shared memory space needed */
+Size
+GlobalSnapshotShmemSize(void)
+{
+ Size size = 0;
+
+ if (track_global_snapshots || global_snapshot_defer_time > 0)
+ {
+ size += MAXALIGN(sizeof(GlobalSnapshotState));
+ }
+
+ if (global_snapshot_defer_time > 0)
+ {
+ size += sizeof(GlobalSnapshotXidMap);
+ size += global_snapshot_defer_time*sizeof(TransactionId);
+ size = MAXALIGN(size);
+ }
+
+ return size;
+}
+
+/* Init shared memory structures */
+void
+GlobalSnapshotShmemInit()
+{
+ bool found;
+
+ if (track_global_snapshots || global_snapshot_defer_time > 0)
+ {
+ gsState = ShmemInitStruct("gsState",
+ sizeof(GlobalSnapshotState),
+ &found);
+ if (!found)
+ {
+ gsState->last_global_csn = 0;
+ SpinLockInit(&gsState->lock);
+ }
+ }
+
+ if (global_snapshot_defer_time > 0)
+ {
+ gsXidMap = ShmemInitStruct("gsXidMap",
+ sizeof(GlobalSnapshotXidMap),
+ &found);
+ if (!found)
+ {
+ int i;
+
+ pg_atomic_init_u64(&gsXidMap->last_csn_seconds, 0);
+ gsXidMap->head = 0;
+ gsXidMap->size = global_snapshot_defer_time;
+ gsXidMap->xmin_by_second =
+ ShmemAlloc(sizeof(TransactionId)*gsXidMap->size);
+
+ for (i = 0; i < gsXidMap->size; i++)
+ gsXidMap->xmin_by_second[i] = InvalidTransactionId;
+ }
+ }
+}
+
+/*
+ * GlobalSnapshotStartup
+ *
+ * Set gsXidMap entries to oldestActiveXID during startup.
+ */
+void
+GlobalSnapshotStartup(TransactionId oldestActiveXID)
+{
+ /*
+ * Run only if we have initialized shared memory and gsXidMap
+ * is enabled.
+ */
+ if (IsNormalProcessingMode() && global_snapshot_defer_time > 0)
+ {
+ int i;
+
+ Assert(TransactionIdIsValid(oldestActiveXID));
+ for (i = 0; i < gsXidMap->size; i++)
+ gsXidMap->xmin_by_second[i] = oldestActiveXID;
+ ProcArraySetGlobalSnapshotXmin(oldestActiveXID);
+ }
+}
+
+/*
+ * GlobalSnapshotMapXmin
+ *
+ * Maintain circular buffer of oldestXmins for several seconds in past. This
+ * buffer allows to shift oldestXmin in the past when backend is importing
+ * global transaction. Otherwise old versions of tuples that were needed for
+ * this transaction can be recycled by other processes (vacuum, HOT, etc).
+ *
+ * Locking here is not trivial. Called upon each snapshot creation after
+ * ProcArrayLock is released. Such usage creates several race conditions. It
+ * is possible that backend who got global_csn called GlobalSnapshotMapXmin()
+ * only after other backends managed to get snapshot and complete
+ * GlobalSnapshotMapXmin() call, or even committed. This is safe because
+ *
+ * * We already hold our xmin in MyPgXact, so our snapshot will not be
+ * harmed even though ProcArrayLock is released.
+ *
+ * * snapshot_global_csn is always pessmistically rounded up to the next
+ * second.
+ *
+ * * For performance reasons, xmin value for particular second is filled
+ * only once. Because of that instead of writing to buffer just our
+ * xmin (which is enough for our snapshot), we bump oldestXmin there --
+ * it mitigates the possibility of damaging someone else's snapshot by
+ * writing to the buffer too advanced value in case of slowness of
+ * another backend who generated csn earlier, but didn't manage to
+ * insert it before us.
+ *
+ * * if GlobalSnapshotMapXmin() founds a gap in several seconds between
+ * current call and latest completed call then it should fill that gap
+ * with latest known values instead of new one. Otherwise it is
+ * possible (however highly unlikely) that this gap also happend
+ * between taking snapshot and call to GlobalSnapshotMapXmin() for some
+ * backend. And we are at risk to fill circullar buffer with
+ * oldestXmin's that are bigger then they actually were.
+ */
+void
+GlobalSnapshotMapXmin(GlobalCSN snapshot_global_csn)
+{
+ int offset, gap, i;
+ GlobalCSN csn_seconds;
+ GlobalCSN last_csn_seconds;
+ volatile TransactionId oldest_deferred_xmin;
+ TransactionId current_oldest_xmin, previous_oldest_xmin;
+
+ /* Callers should check config values */
+ Assert(global_snapshot_defer_time > 0);
+ Assert(gsXidMap != NULL);
+
+ /*
+ * Round up global_csn to the next second -- pessimistically and safely.
+ */
+ csn_seconds = (snapshot_global_csn / NSECS_PER_SEC + 1);
+
+ /*
+ * Fast-path check. Avoid taking exclusive GlobalSnapshotXidMapLock lock
+ * if oldestXid was already written to xmin_by_second[] for this rounded
+ * global_csn.
+ */
+ if (pg_atomic_read_u64(&gsXidMap->last_csn_seconds) >= csn_seconds)
+ return;
+
+ /* Ok, we have new entry (or entries) */
+ LWLockAcquire(GlobalSnapshotXidMapLock, LW_EXCLUSIVE);
+
+ /* Re-check last_csn_seconds under lock */
+ last_csn_seconds = pg_atomic_read_u64(&gsXidMap->last_csn_seconds);
+ if (last_csn_seconds >= csn_seconds)
+ {
+ LWLockRelease(GlobalSnapshotXidMapLock);
+ return;
+ }
+ pg_atomic_write_u64(&gsXidMap->last_csn_seconds, csn_seconds);
+
+ /*
+ * Count oldest_xmin.
+ *
+ * It was possible to calculate oldest_xmin during corresponding snapshot
+ * creation, but GetSnapshotData() intentionally reads only PgXact, but not
+ * PgProc. And we need info about originalXmin (see comment to gsXidMap)
+ * which is stored in PgProc because of threats in comments around PgXact
+ * about extending it with new fields. So just calculate oldest_xmin again,
+ * that anyway happens quite rarely.
+ */
+ current_oldest_xmin = GetOldestXmin(NULL, PROCARRAY_NON_IMPORTED_XMIN);
+
+ previous_oldest_xmin = gsXidMap->xmin_by_second[gsXidMap->head];
+
+ Assert(TransactionIdIsNormal(current_oldest_xmin));
+ Assert(TransactionIdIsNormal(previous_oldest_xmin));
+
+ gap = csn_seconds - last_csn_seconds;
+ offset = csn_seconds % gsXidMap->size;
+
+ /* Sanity check before we update head and gap */
+ Assert( gap >= 1 );
+ Assert( (gsXidMap->head + gap) % gsXidMap->size == offset );
+
+ gap = gap > gsXidMap->size ? gsXidMap->size : gap;
+ gsXidMap->head = offset;
+
+ /* Fill new entry with current_oldest_xmin */
+ gsXidMap->xmin_by_second[offset] = current_oldest_xmin;
+
+ /*
+ * If we have gap then fill it with previous_oldest_xmin for reasons
+ * outlined in comment above this function.
+ */
+ for (i = 1; i < gap; i++)
+ {
+ offset = (offset + gsXidMap->size - 1) % gsXidMap->size;
+ gsXidMap->xmin_by_second[offset] = previous_oldest_xmin;
+ }
+
+ oldest_deferred_xmin =
+ gsXidMap->xmin_by_second[ (gsXidMap->head + 1) % gsXidMap->size ];
+
+ LWLockRelease(GlobalSnapshotXidMapLock);
+
+ /*
+ * Advance procArray->global_snapshot_xmin after we released
+ * GlobalSnapshotXidMapLock. Since we gather not xmin but oldestXmin, it
+ * never goes backwards regardless of how slow we can do that.
+ */
+ Assert(TransactionIdFollowsOrEquals(oldest_deferred_xmin,
+ ProcArrayGetGlobalSnapshotXmin()));
+ ProcArraySetGlobalSnapshotXmin(oldest_deferred_xmin);
+}
+
+
+/*
+ * GlobalSnapshotToXmin
+ *
+ * Get oldestXmin that took place when snapshot_global_csn was taken.
+ */
+TransactionId
+GlobalSnapshotToXmin(GlobalCSN snapshot_global_csn)
+{
+ TransactionId xmin;
+ GlobalCSN csn_seconds;
+ volatile GlobalCSN last_csn_seconds;
+
+ /* Callers should check config values */
+ Assert(global_snapshot_defer_time > 0);
+ Assert(gsXidMap != NULL);
+
+ /* Round down to get conservative estimates */
+ csn_seconds = (snapshot_global_csn / NSECS_PER_SEC);
+
+ LWLockAcquire(GlobalSnapshotXidMapLock, LW_SHARED);
+ last_csn_seconds = pg_atomic_read_u64(&gsXidMap->last_csn_seconds);
+ if (csn_seconds > last_csn_seconds)
+ {
+ /* we don't have entry for this global_csn yet, return latest known */
+ xmin = gsXidMap->xmin_by_second[gsXidMap->head];
+ }
+ else if (last_csn_seconds - csn_seconds < gsXidMap->size)
+ {
+ /* we are good, retrieve value from our map */
+ Assert(last_csn_seconds % gsXidMap->size == gsXidMap->head);
+ xmin = gsXidMap->xmin_by_second[csn_seconds % gsXidMap->size];
+ }
+ else
+ {
+ /* requested global_csn is too old, let caller know */
+ xmin = InvalidTransactionId;
+ }
+ LWLockRelease(GlobalSnapshotXidMapLock);
+
+ return xmin;
+}
+
+/*
+ * GlobalSnapshotGenerate
+ *
+ * Generate GlobalCSN which is actually a local time. Also we are forcing
+ * this time to be always increasing. Since now it is not uncommon to have
+ * millions of read transactions per second we are trying to use nanoseconds
+ * if such time resolution is available.
+ */
+GlobalCSN
+GlobalSnapshotGenerate(bool locked)
+{
+ instr_time current_time;
+ GlobalCSN global_csn;
+
+ Assert(track_global_snapshots || global_snapshot_defer_time > 0);
+
+ /*
+ * TODO: create some macro that add small random shift to current time.
+ */
+ INSTR_TIME_SET_CURRENT(current_time);
+ global_csn = (GlobalCSN) INSTR_TIME_GET_NANOSEC(current_time);
+
+ /* TODO: change to atomics? */
+ if (!locked)
+ SpinLockAcquire(&gsState->lock);
+
+ if (global_csn <= gsState->last_global_csn)
+ global_csn = ++gsState->last_global_csn;
+ else
+ gsState->last_global_csn = global_csn;
+
+ if (!locked)
+ SpinLockRelease(&gsState->lock);
+
+ return global_csn;
+}
+
+/*
+ * GlobalSnapshotSync
+ *
+ * Due to time desynchronization on different nodes we can receive global_csn
+ * which is greater than global_csn on this node. To preserve proper isolation
+ * this node needs to wait when such global_csn comes on local clock.
+ *
+ * This should happend relatively rare if nodes have running NTP/PTP/etc.
+ * Complain if wait time is more than SNAP_SYNC_COMPLAIN.
+ */
+void
+GlobalSnapshotSync(GlobalCSN remote_gcsn)
+{
+ GlobalCSN local_gcsn;
+ GlobalCSN delta;
+
+ Assert(track_global_snapshots);
+
+ for(;;)
+ {
+ SpinLockAcquire(&gsState->lock);
+ if (gsState->last_global_csn > remote_gcsn)
+ {
+ /* Everything is fine */
+ SpinLockRelease(&gsState->lock);
+ return;
+ }
+ else if ((local_gcsn = GlobalSnapshotGenerate(true)) >= remote_gcsn)
+ {
+ /*
+ * Everything is fine too, but last_global_csn wasn't updated for
+ * some time.
+ */
+ SpinLockRelease(&gsState->lock);
+ return;
+ }
+ SpinLockRelease(&gsState->lock);
+
+ /* Okay we need to sleep now */
+ delta = remote_gcsn - local_gcsn;
+ if (delta > SNAP_DESYNC_COMPLAIN)
+ ereport(WARNING,
+ (errmsg("remote global snapshot exceeds ours by more than a second"),
+ errhint("Consider running NTPd on servers participating in global transaction")));
+
+ /* TODO: report this sleeptime somewhere? */
+ pg_usleep((long) (delta/NSECS_PER_USEC));
+
+ /*
+ * Loop that checks to ensure that we actually slept for specified
+ * amount of time.
+ */
+ }
+
+ Assert(false); /* Should not happend */
+ return;
+}
+
+/*
+ * TransactionIdGetGlobalCSN
+ *
+ * Get GlobalCSN for specified TransactionId taking care about special xids,
+ * xids beyond TransactionXmin and InDoubt states.
+ */
+GlobalCSN
+TransactionIdGetGlobalCSN(TransactionId xid)
+{
+ GlobalCSN global_csn;
+
+ Assert(track_global_snapshots);
+
+ /* Handle permanent TransactionId's for which we don't have mapping */
+ if (!TransactionIdIsNormal(xid))
+ {
+ if (xid == InvalidTransactionId)
+ return AbortedGlobalCSN;
+ if (xid == FrozenTransactionId || xid == BootstrapTransactionId)
+ return FrozenGlobalCSN;
+ Assert(false); /* Should not happend */
+ }
+
+ /*
+ * For xids which less then TransactionXmin GlobalCSNLog can be already
+ * trimmed but we know that such transaction is definetly not concurrently
+ * running according to any snapshot including timetravel ones. Callers
+ * should check TransactionDidCommit after.
+ */
+ if (TransactionIdPrecedes(xid, TransactionXmin))
+ return FrozenGlobalCSN;
+
+ /* Read GlobalCSN from SLRU */
+ global_csn = GlobalCSNLogGetCSN(xid);
+
+ /*
+ * If we faced InDoubt state then transaction is beeing committed and we
+ * should wait until GlobalCSN will be assigned so that visibility check
+ * could decide whether tuple is in snapshot. See also comments in
+ * GlobalSnapshotPrecommit().
+ */
+ if (GlobalCSNIsInDoubt(global_csn))
+ {
+ XactLockTableWait(xid, NULL, NULL, XLTW_None);
+ global_csn = GlobalCSNLogGetCSN(xid);
+ Assert(GlobalCSNIsNormal(global_csn) ||
+ GlobalCSNIsAborted(global_csn));
+ }
+
+ Assert(GlobalCSNIsNormal(global_csn) ||
+ GlobalCSNIsInProgress(global_csn) ||
+ GlobalCSNIsAborted(global_csn));
+
+ return global_csn;
+}
+
+/*
+ * XidInvisibleInGlobalSnapshot
+ *
+ * Version of XidInMVCCSnapshot for global transactions. For non-imported
+ * global snapshots this should give same results as XidInLocalMVCCSnapshot
+ * (except that aborts will be shown as invisible without going to clog) and to
+ * ensure such behaviour XidInMVCCSnapshot is coated with asserts that checks
+ * identicalness of XidInvisibleInGlobalSnapshot/XidInLocalMVCCSnapshot in
+ * case of ordinary snapshot.
+ */
+bool
+XidInvisibleInGlobalSnapshot(TransactionId xid, Snapshot snapshot)
+{
+ GlobalCSN csn;
+
+ Assert(track_global_snapshots);
+
+ csn = TransactionIdGetGlobalCSN(xid);
+
+ if (GlobalCSNIsNormal(csn))
+ {
+ if (csn < snapshot->global_csn)
+ return false;
+ else
+ return true;
+ }
+ else if (GlobalCSNIsFrozen(csn))
+ {
+ /* It is bootstrap or frozen transaction */
+ return false;
+ }
+ else
+ {
+ /* It is aborted or in-progress */
+ Assert(GlobalCSNIsAborted(csn) || GlobalCSNIsInProgress(csn));
+ if (GlobalCSNIsAborted(csn))
+ Assert(TransactionIdDidAbort(xid));
+ return true;
+ }
+}
+
+
+/*****************************************************************************
+ * Functions to handle distributed commit on transaction coordinator:
+ * GlobalSnapshotPrepareCurrent() / GlobalSnapshotAssignCsnCurrent().
+ * Correspoding functions for remote nodes are defined in twophase.c:
+ * pg_global_snapshot_prepare/pg_global_snapshot_assign.
+ *****************************************************************************/
+
+
+/*
+ * GlobalSnapshotPrepareCurrent
+ *
+ * Set InDoubt state for currently active transaction and return commit's
+ * global snapshot.
+ */
+GlobalCSN
+GlobalSnapshotPrepareCurrent()
+{
+ TransactionId xid = GetCurrentTransactionIdIfAny();
+
+ if (!track_global_snapshots)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("could not prepare transaction for global commit"),
+ errhint("Make sure the configuration parameter \"%s\" is enabled.",
+ "track_global_snapshots")));
+
+ if (TransactionIdIsValid(xid))
+ {
+ TransactionId *subxids;
+ int nsubxids = xactGetCommittedChildren(&subxids);
+ GlobalCSNLogSetCSN(xid, nsubxids,
+ subxids, InDoubtGlobalCSN);
+ }
+
+ /* Nothing to write if we don't heve xid */
+
+ return GlobalSnapshotGenerate(false);
+}
+
+/*
+ * GlobalSnapshotAssignCsnCurrent
+ *
+ * Asign GlobalCSN for currently active transaction. GlobalCSN is supposedly
+ * maximal among of values returned by GlobalSnapshotPrepareCurrent and
+ * pg_global_snapshot_prepare.
+ */
+void
+GlobalSnapshotAssignCsnCurrent(GlobalCSN global_csn)
+{
+ if (!track_global_snapshots)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("could not prepare transaction for global commit"),
+ errhint("Make sure the configuration parameter \"%s\" is enabled.",
+ "track_global_snapshots")));
+
+ if (!GlobalCSNIsNormal(global_csn))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("pg_global_snapshot_assign expects normal global_csn")));
+
+ /* Skip emtpty transactions */
+ if (!TransactionIdIsValid(GetCurrentTransactionIdIfAny()))
+ return;
+
+ /* Set global_csn and defuse ProcArrayEndTransaction from assigning one */
+ pg_atomic_write_u64(&MyProc->assignedGlobalCsn, global_csn);
+}
+
+
+/*****************************************************************************
+ * Functions to handle global and local transactions commit.
+ *
+ * For local transactions GlobalSnapshotPrecommit sets InDoubt state before
+ * ProcArrayEndTransaction is called and transaction data potetntially becomes
+ * visible to other backends. ProcArrayEndTransaction (or ProcArrayRemove in
+ * twophase case) then acquires global_csn under ProcArray lock and stores it
+ * in proc->assignedGlobalCsn. It's important that global_csn for commit is
+ * generated under ProcArray lock, otherwise global and local snapshots won't
+ * be equivalent. Consequent call to GlobalSnapshotCommit will write
+ * proc->assignedGlobalCsn to GlobalCSNLog.
+ *
+ * Same rules applies to global transaction, except that global_csn is already
+ * assigned by GlobalSnapshotAssignCsnCurrent/pg_global_snapshot_assign and
+ * GlobalSnapshotPrecommit is basically no-op.
+ *
+ * GlobalSnapshotAbort is slightly different comparing to commit because abort
+ * can skip InDoubt phase and can be called for transaction subtree.
+ *****************************************************************************/
+
+
+/*
+ * GlobalSnapshotAbort
+ *
+ * Abort transaction in GlobalCsnLog. We can skip InDoubt state for aborts
+ * since no concurrent transactions allowed to see aborted data anyway.
+ */
+void
+GlobalSnapshotAbort(PGPROC *proc, TransactionId xid,
+ int nsubxids, TransactionId *subxids)
+{
+ if (!track_global_snapshots)
+ return;
+
+ GlobalCSNLogSetCSN(xid, nsubxids, subxids, AbortedGlobalCSN);
+
+ /*
+ * Clean assignedGlobalCsn anyway, as it was possibly set in
+ * GlobalSnapshotAssignCsnCurrent.
+ */
+ pg_atomic_write_u64(&proc->assignedGlobalCsn, InProgressGlobalCSN);
+}
+
+/*
+ * GlobalSnapshotPrecommit
+ *
+ * Set InDoubt status for local transaction that we are going to commit.
+ * This step is needed to achieve consistency between local snapshots and
+ * global csn-based snapshots. We don't hold ProcArray lock while writing
+ * csn for transaction in SLRU but instead we set InDoubt status before
+ * transaction is deleted from ProcArray so the readers who will read csn
+ * in the gap between ProcArray removal and GlobalCSN assignment can wait
+ * until GlobalCSN is finally assigned. See also TransactionIdGetGlobalCSN().
+ *
+ * For global transaction this does nothing as InDoubt state was written
+ * earlier.
+ *
+ * This should be called only from parallel group leader before backend is
+ * deleted from ProcArray.
+ */
+void
+GlobalSnapshotPrecommit(PGPROC *proc, TransactionId xid,
+ int nsubxids, TransactionId *subxids)
+{
+ GlobalCSN oldAssignedGlobalCsn = InProgressGlobalCSN;
+ bool in_progress;
+
+ if (!track_global_snapshots)
+ return;
+
+ /* Set InDoubt status if it is local transaction */
+ in_progress = pg_atomic_compare_exchange_u64(&proc->assignedGlobalCsn,
+ &oldAssignedGlobalCsn,
+ InDoubtGlobalCSN);
+ if (in_progress)
+ {
+ Assert(GlobalCSNIsInProgress(oldAssignedGlobalCsn));
+ GlobalCSNLogSetCSN(xid, nsubxids,
+ subxids, InDoubtGlobalCSN);
+ }
+ else
+ {
+ /* Otherwise we should have valid GlobalCSN by this time */
+ Assert(GlobalCSNIsNormal(oldAssignedGlobalCsn));
+ /* Also global transaction should already be in InDoubt state */
+ Assert(GlobalCSNIsInDoubt(GlobalCSNLogGetCSN(xid)));
+ }
+}
+
+/*
+ * GlobalSnapshotCommit
+ *
+ * Write GlobalCSN that were acquired earlier to GlobalCsnLog. Should be
+ * preceded by GlobalSnapshotPrecommit() so readers can wait until we finally
+ * finished writing to SLRU.
+ *
+ * Should be called after ProcArrayEndTransaction, but before releasing
+ * transaction locks, so that TransactionIdGetGlobalCSN can wait on this
+ * lock for GlobalCSN.
+ */
+void
+GlobalSnapshotCommit(PGPROC *proc, TransactionId xid,
+ int nsubxids, TransactionId *subxids)
+{
+ volatile GlobalCSN assigned_global_csn;
+
+ if (!track_global_snapshots)
+ return;
+
+ if (!TransactionIdIsValid(xid))
+ {
+ assigned_global_csn = pg_atomic_read_u64(&proc->assignedGlobalCsn);
+ Assert(GlobalCSNIsInProgress(assigned_global_csn));
+ return;
+ }
+
+ /* Finally write resulting GlobalCSN in SLRU */
+ assigned_global_csn = pg_atomic_read_u64(&proc->assignedGlobalCsn);
+ Assert(GlobalCSNIsNormal(assigned_global_csn));
+ GlobalCSNLogSetCSN(xid, nsubxids,
+ subxids, assigned_global_csn);
+
+ /* Reset for next transaction */
+ pg_atomic_write_u64(&proc->assignedGlobalCsn, InProgressGlobalCSN);
+}
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 96d2d104d9..df34f6caab 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -77,6 +77,7 @@
#include <unistd.h>
#include "access/commit_ts.h"
+#include "access/global_snapshot.h"
#include "access/global_csn_log.h"
#include "access/htup_details.h"
#include "access/subtrans.h"
@@ -1516,8 +1517,34 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
hdr->nabortrels, abortrels,
gid);
+ /*
+ * GlobalSnapshot callbacks that should be called right before we are
+ * going to become visible. Details in comments to this functions.
+ */
+ if (isCommit)
+ GlobalSnapshotPrecommit(proc, xid, hdr->nsubxacts, children);
+ else
+ GlobalSnapshotAbort(proc, xid, hdr->nsubxacts, children);
+
+
ProcArrayRemove(proc, latestXid);
+ /*
+ * Stamp our transaction with GlobalCSN in GlobalCsnLog.
+ * Should be called after ProcArrayEndTransaction, but before releasing
+ * transaction locks, since TransactionIdGetGlobalCSN relies on
+ * XactLockTableWait to await global_csn.
+ */
+ if (isCommit)
+ {
+ GlobalSnapshotCommit(proc, xid, hdr->nsubxacts, children);
+ }
+ else
+ {
+ Assert(GlobalCSNIsInProgress(
+ pg_atomic_read_u64(&proc->assignedGlobalCsn)));
+ }
+
/*
* In case we fail while running the callbacks, mark the gxact invalid so
* no one else will try to commit/rollback, and so it will be recycled if
@@ -2494,3 +2521,132 @@ PrepareRedoRemove(TransactionId xid, bool giveWarning)
return;
}
+
+/*
+ * GlobalSnapshotPrepareTwophase
+ *
+ * Set InDoubt state for currently active transaction and return commit's
+ * global snapshot.
+ *
+ * This function is a counterpart of GlobalSnapshotPrepareCurrent() for
+ * twophase transactions.
+ */
+static GlobalCSN
+GlobalSnapshotPrepareTwophase(const char *gid)
+{
+ GlobalTransaction gxact;
+ PGXACT *pgxact;
+ char *buf;
+ TransactionId xid;
+ xl_xact_parsed_prepare parsed;
+
+ if (!track_global_snapshots)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("could not prepare transaction for global commit"),
+ errhint("Make sure the configuration parameter \"%s\" is enabled.",
+ "track_global_snapshots")));
+
+ /*
+ * Validate the GID, and lock the GXACT to ensure that two backends do not
+ * try to access the same GID at once.
+ */
+ gxact = LockGXact(gid, GetUserId());
+ pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
+ xid = pgxact->xid;
+
+ if (gxact->ondisk)
+ buf = ReadTwoPhaseFile(xid, true);
+ else
+ XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL);
+
+ ParsePrepareRecord(0, buf, &parsed);
+
+ GlobalCSNLogSetCSN(xid, parsed.nsubxacts,
+ parsed.subxacts, InDoubtGlobalCSN);
+
+ /* Unlock our GXACT */
+ LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
+ gxact->locking_backend = InvalidBackendId;
+ LWLockRelease(TwoPhaseStateLock);
+
+ pfree(buf);
+
+ return GlobalSnapshotGenerate(false);
+}
+
+/*
+ * SQL interface to GlobalSnapshotPrepareTwophase()
+ *
+ * TODO: Rewrite this as PREPARE TRANSACTION 'gid' RETURNING SNAPSHOT
+ */
+Datum
+pg_global_snapshot_prepare(PG_FUNCTION_ARGS)
+{
+ const char *gid = text_to_cstring(PG_GETARG_TEXT_PP(0));
+ GlobalCSN global_csn;
+
+ global_csn = GlobalSnapshotPrepareTwophase(gid);
+
+ PG_RETURN_INT64(global_csn);
+}
+
+
+/*
+ * TwoPhaseAssignGlobalCsn
+ *
+ * Asign GlobalCSN for currently active transaction. GlobalCSN is supposedly
+ * maximal among of values returned by GlobalSnapshotPrepareCurrent and
+ * pg_global_snapshot_prepare.
+ *
+ * This function is a counterpart of GlobalSnapshotAssignCsnCurrent() for
+ * twophase transactions.
+ */
+static void
+GlobalSnapshotAssignCsnTwoPhase(const char *gid, GlobalCSN global_csn)
+{
+ GlobalTransaction gxact;
+ PGPROC *proc;
+
+ if (!track_global_snapshots)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("could not prepare transaction for global commit"),
+ errhint("Make sure the configuration parameter \"%s\" is enabled.",
+ "track_global_snapshots")));
+
+ if (!GlobalCSNIsNormal(global_csn))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("pg_global_snapshot_assign expects normal global_csn")));
+
+ /*
+ * Validate the GID, and lock the GXACT to ensure that two backends do not
+ * try to access the same GID at once.
+ */
+ gxact = LockGXact(gid, GetUserId());
+ proc = &ProcGlobal->allProcs[gxact->pgprocno];
+
+ /* Set global_csn and defuse ProcArrayRemove from assigning one. */
+ pg_atomic_write_u64(&proc->assignedGlobalCsn, global_csn);
+
+ /* Unlock our GXACT */
+ LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
+ gxact->locking_backend = InvalidBackendId;
+ LWLockRelease(TwoPhaseStateLock);
+}
+
+/*
+ * SQL interface to GlobalSnapshotAssignCsnTwoPhase()
+ *
+ * TODO: Rewrite this as COMMIT PREPARED 'gid' SNAPSHOT 'global_csn'
+ */
+Datum
+pg_global_snapshot_assign(PG_FUNCTION_ARGS)
+{
+ const char *gid = text_to_cstring(PG_GETARG_TEXT_PP(0));
+ GlobalCSN global_csn = PG_GETARG_INT64(1);
+
+ GlobalSnapshotAssignCsnTwoPhase(gid, global_csn);
+ PG_RETURN_VOID();
+}
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index d967400384..bf5885f878 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -21,6 +21,7 @@
#include <unistd.h>
#include "access/commit_ts.h"
+#include "access/global_snapshot.h"
#include "access/multixact.h"
#include "access/parallel.h"
#include "access/subtrans.h"
@@ -1345,6 +1346,14 @@ RecordTransactionCommit(void)
/* Reset XactLastRecEnd until the next transaction writes something */
XactLastRecEnd = 0;
+
+ /*
+ * Mark our transaction as InDoubt in GlobalCsnLog and get ready for
+ * commit.
+ */
+ if (markXidCommitted)
+ GlobalSnapshotPrecommit(MyProc, xid, nchildren, children);
+
cleanup:
/* Clean up local data */
if (rels)
@@ -1606,6 +1615,11 @@ RecordTransactionAbort(bool isSubXact)
*/
TransactionIdAbortTree(xid, nchildren, children);
+ /*
+ * Mark our transaction as Aborted in GlobalCsnLog.
+ */
+ GlobalSnapshotAbort(MyProc, xid, nchildren, children);
+
END_CRIT_SECTION();
/* Compute latestXid while we have the child XIDs handy */
@@ -2079,6 +2093,21 @@ CommitTransaction(void)
*/
ProcArrayEndTransaction(MyProc, latestXid);
+ /*
+ * Stamp our transaction with GlobalCSN in GlobalCsnLog.
+ * Should be called after ProcArrayEndTransaction, but before releasing
+ * transaction locks.
+ */
+ if (!is_parallel_worker)
+ {
+ TransactionId xid = GetTopTransactionIdIfAny();
+ TransactionId *subxids;
+ int nsubxids;
+
+ nsubxids = xactGetCommittedChildren(&subxids);
+ GlobalSnapshotCommit(MyProc, xid, nsubxids, subxids);
+ }
+
/*
* This is all post-commit cleanup. Note that if an error is raised here,
* it's too late to abort the transaction. This should be just
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index f26cc952bd..5488e27d89 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -6901,6 +6901,7 @@ StartupXLOG(void)
StartupCLOG();
StartupGlobalCSNLog(oldestActiveXID);
StartupSUBTRANS(oldestActiveXID);
+ GlobalSnapshotStartup(oldestActiveXID);
/*
* If we're beginning at a shutdown checkpoint, we know that
@@ -7700,6 +7701,7 @@ StartupXLOG(void)
StartupCLOG();
StartupGlobalCSNLog(oldestActiveXID);
StartupSUBTRANS(oldestActiveXID);
+ GlobalSnapshotStartup(oldestActiveXID);
}
/*
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c
index 2af468fc6a..3a04f6a824 100644
--- a/src/backend/storage/ipc/ipci.c
+++ b/src/backend/storage/ipc/ipci.c
@@ -22,6 +22,7 @@
#include "access/nbtree.h"
#include "access/subtrans.h"
#include "access/twophase.h"
+#include "access/global_snapshot.h"
#include "commands/async.h"
#include "miscadmin.h"
#include "pgstat.h"
@@ -147,6 +148,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
size = add_size(size, WalSndShmemSize());
size = add_size(size, WalRcvShmemSize());
size = add_size(size, ApplyLauncherShmemSize());
+ size = add_size(size, GlobalSnapshotShmemSize());
size = add_size(size, SnapMgrShmemSize());
size = add_size(size, BTreeShmemSize());
size = add_size(size, SyncScanShmemSize());
@@ -273,6 +275,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
SyncScanShmemInit();
AsyncShmemInit();
BackendRandomShmemInit();
+ GlobalSnapshotShmemInit();
#ifdef EXEC_BACKEND
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index 6deb810969..a628f012da 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -47,6 +47,7 @@
#include "access/clog.h"
#include "access/global_csn_log.h"
+#include "access/global_snapshot.h"
#include "access/subtrans.h"
#include "access/transam.h"
#include "access/twophase.h"
@@ -92,6 +93,8 @@ typedef struct ProcArrayStruct
TransactionId replication_slot_xmin;
/* oldest catalog xmin of any replication slot */
TransactionId replication_slot_catalog_xmin;
+ /* xmin of oldest active global snapshot */
+ TransactionId global_snapshot_xmin;
/* indexes into allPgXact[], has PROCARRAY_MAXPROCS entries */
int pgprocnos[FLEXIBLE_ARRAY_MEMBER];
@@ -247,6 +250,7 @@ CreateSharedProcArray(void)
procArray->lastOverflowedXid = InvalidTransactionId;
procArray->replication_slot_xmin = InvalidTransactionId;
procArray->replication_slot_catalog_xmin = InvalidTransactionId;
+ procArray->global_snapshot_xmin = InvalidTransactionId;
}
allProcs = ProcGlobal->allProcs;
@@ -353,6 +357,17 @@ ProcArrayRemove(PGPROC *proc, TransactionId latestXid)
if (TransactionIdPrecedes(ShmemVariableCache->latestCompletedXid,
latestXid))
ShmemVariableCache->latestCompletedXid = latestXid;
+
+ /*
+ * Assign global csn while holding ProcArrayLock for non-global
+ * COMMIT PREPARED. After lock is released consequent
+ * GlobalSnapshotCommit() will write this value to GlobalCsnLog.
+ *
+ * In case of global commit proc->assignedGlobalCsn is already set
+ * by prior AssignGlobalCsn().
+ */
+ if (GlobalCSNIsInDoubt(pg_atomic_read_u64(&proc->assignedGlobalCsn)))
+ pg_atomic_write_u64(&proc->assignedGlobalCsn, GlobalSnapshotGenerate(false));
}
else
{
@@ -433,6 +448,8 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid)
proc->lxid = InvalidLocalTransactionId;
pgxact->xmin = InvalidTransactionId;
+ proc->originalXmin = InvalidTransactionId;
+
/* must be cleared with xid/xmin: */
pgxact->vacuumFlags &= ~PROC_VACUUM_STATE_MASK;
pgxact->delayChkpt = false; /* be sure this is cleared in abort */
@@ -455,6 +472,8 @@ ProcArrayEndTransactionInternal(PGPROC *proc, PGXACT *pgxact,
pgxact->xid = InvalidTransactionId;
proc->lxid = InvalidLocalTransactionId;
pgxact->xmin = InvalidTransactionId;
+ proc->originalXmin = InvalidTransactionId;
+
/* must be cleared with xid/xmin: */
pgxact->vacuumFlags &= ~PROC_VACUUM_STATE_MASK;
pgxact->delayChkpt = false; /* be sure this is cleared in abort */
@@ -468,6 +487,20 @@ ProcArrayEndTransactionInternal(PGPROC *proc, PGXACT *pgxact,
if (TransactionIdPrecedes(ShmemVariableCache->latestCompletedXid,
latestXid))
ShmemVariableCache->latestCompletedXid = latestXid;
+
+ /*
+ * Assign global csn while holding ProcArrayLock for non-global
+ * COMMIT. After lock is released consequent GlobalSnapshotFinish() will
+ * write this value to GlobalCsnLog.
+ *
+ * In case of global commit MyProc->assignedGlobalCsn is already set
+ * by prior AssignGlobalCsn().
+ *
+ * TODO: in case of group commit we can generate one GlobalSnapshot for
+ * whole group to save time on timestamp aquisition.
+ */
+ if (GlobalCSNIsInDoubt(pg_atomic_read_u64(&proc->assignedGlobalCsn)))
+ pg_atomic_write_u64(&proc->assignedGlobalCsn, GlobalSnapshotGenerate(false));
}
/*
@@ -611,6 +644,7 @@ ProcArrayClearTransaction(PGPROC *proc)
pgxact->xid = InvalidTransactionId;
proc->lxid = InvalidLocalTransactionId;
pgxact->xmin = InvalidTransactionId;
+ proc->originalXmin = InvalidTransactionId;
proc->recoveryConflictPending = false;
/* redundant, but just in case */
@@ -1329,6 +1363,7 @@ GetOldestXmin(Relation rel, int flags)
TransactionId replication_slot_xmin = InvalidTransactionId;
TransactionId replication_slot_catalog_xmin = InvalidTransactionId;
+ TransactionId global_snapshot_xmin = InvalidTransactionId;
/*
* If we're not computing a relation specific limit, or if a shared
@@ -1365,8 +1400,9 @@ GetOldestXmin(Relation rel, int flags)
proc->databaseId == MyDatabaseId ||
proc->databaseId == 0) /* always include WalSender */
{
- /* Fetch xid just once - see GetNewTransactionId */
+ /* Fetch both xids just once - see GetNewTransactionId */
TransactionId xid = UINT32_ACCESS_ONCE(pgxact->xid);
+ TransactionId original_xmin = UINT32_ACCESS_ONCE(proc->originalXmin);
/* First consider the transaction's own Xid, if any */
if (TransactionIdIsNormal(xid) &&
@@ -1379,8 +1415,17 @@ GetOldestXmin(Relation rel, int flags)
* We must check both Xid and Xmin because a transaction might
* have an Xmin but not (yet) an Xid; conversely, if it has an
* Xid, that could determine some not-yet-set Xmin.
+ *
+ * In case of oldestXmin calculation for GlobalSnapshotMapXmin()
+ * pgxact->xmin should be changed to proc->originalXmin. Details
+ * in commets to GlobalSnapshotMapXmin.
*/
- xid = UINT32_ACCESS_ONCE(pgxact->xmin);
+ if ((flags & PROCARRAY_NON_IMPORTED_XMIN) &&
+ TransactionIdIsValid(original_xmin))
+ xid = original_xmin;
+ else
+ xid = UINT32_ACCESS_ONCE(pgxact->xmin);
+
if (TransactionIdIsNormal(xid) &&
TransactionIdPrecedes(xid, result))
result = xid;
@@ -1394,6 +1439,7 @@ GetOldestXmin(Relation rel, int flags)
*/
replication_slot_xmin = procArray->replication_slot_xmin;
replication_slot_catalog_xmin = procArray->replication_slot_catalog_xmin;
+ global_snapshot_xmin = procArray->global_snapshot_xmin;
if (RecoveryInProgress())
{
@@ -1435,6 +1481,11 @@ GetOldestXmin(Relation rel, int flags)
result = FirstNormalTransactionId;
}
+ if (!(flags & PROCARRAY_NON_IMPORTED_XMIN) &&
+ TransactionIdIsValid(global_snapshot_xmin) &&
+ NormalTransactionIdPrecedes(global_snapshot_xmin, result))
+ result = global_snapshot_xmin;
+
/*
* Check whether there are replication slots requiring an older xmin.
*/
@@ -1528,8 +1579,10 @@ GetSnapshotData(Snapshot snapshot)
int count = 0;
int subcount = 0;
bool suboverflowed = false;
+ GlobalCSN global_csn = FrozenGlobalCSN;
TransactionId replication_slot_xmin = InvalidTransactionId;
TransactionId replication_slot_catalog_xmin = InvalidTransactionId;
+ TransactionId global_snapshot_xmin = InvalidTransactionId;
Assert(snapshot != NULL);
@@ -1721,10 +1774,18 @@ GetSnapshotData(Snapshot snapshot)
*/
replication_slot_xmin = procArray->replication_slot_xmin;
replication_slot_catalog_xmin = procArray->replication_slot_catalog_xmin;
+ global_snapshot_xmin = procArray->global_snapshot_xmin;
if (!TransactionIdIsValid(MyPgXact->xmin))
MyPgXact->xmin = TransactionXmin = xmin;
+ /*
+ * Take GlobalCSN under ProcArrayLock so the local/global snapshot stays
+ * synchronized.
+ */
+ if (track_global_snapshots)
+ global_csn = GlobalSnapshotGenerate(false);
+
LWLockRelease(ProcArrayLock);
/*
@@ -1740,6 +1801,10 @@ GetSnapshotData(Snapshot snapshot)
if (!TransactionIdIsNormal(RecentGlobalXmin))
RecentGlobalXmin = FirstNormalTransactionId;
+ if (TransactionIdIsValid(global_snapshot_xmin) &&
+ TransactionIdPrecedes(global_snapshot_xmin, RecentGlobalXmin))
+ RecentGlobalXmin = global_snapshot_xmin;
+
/* Check whether there's a replication slot requiring an older xmin. */
if (TransactionIdIsValid(replication_slot_xmin) &&
NormalTransactionIdPrecedes(replication_slot_xmin, RecentGlobalXmin))
@@ -1795,6 +1860,11 @@ GetSnapshotData(Snapshot snapshot)
MaintainOldSnapshotTimeMapping(snapshot->whenTaken, xmin);
}
+ snapshot->imported_global_csn = false;
+ snapshot->global_csn = global_csn;
+ if (global_snapshot_defer_time > 0 && IsUnderPostmaster)
+ GlobalSnapshotMapXmin(snapshot->global_csn);
+
return snapshot;
}
@@ -3032,6 +3102,24 @@ ProcArrayGetReplicationSlotXmin(TransactionId *xmin,
LWLockRelease(ProcArrayLock);
}
+/*
+ * ProcArraySetGlobalSnapshotXmin
+ */
+void
+ProcArraySetGlobalSnapshotXmin(TransactionId xmin)
+{
+ /* We rely on atomic fetch/store of xid */
+ procArray->global_snapshot_xmin = xmin;
+}
+
+/*
+ * ProcArrayGetGlobalSnapshotXmin
+ */
+TransactionId
+ProcArrayGetGlobalSnapshotXmin(void)
+{
+ return procArray->global_snapshot_xmin;
+}
#define XidCacheRemove(i) \
do { \
diff --git a/src/backend/storage/lmgr/lwlocknames.txt b/src/backend/storage/lmgr/lwlocknames.txt
index 9615058f29..8ef731d560 100644
--- a/src/backend/storage/lmgr/lwlocknames.txt
+++ b/src/backend/storage/lmgr/lwlocknames.txt
@@ -51,3 +51,4 @@ BackendRandomLock 43
LogicalRepWorkerLock 44
CLogTruncationLock 45
GlobalCSNLogControlLock 46
+GlobalSnapshotXidMapLock 47
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 33387fb71b..831bcafa97 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -37,6 +37,7 @@
#include "access/transam.h"
#include "access/twophase.h"
+#include "access/global_snapshot.h"
#include "access/xact.h"
#include "miscadmin.h"
#include "pgstat.h"
@@ -425,6 +426,9 @@ InitProcess(void)
MyProc->clogGroupMemberLsn = InvalidXLogRecPtr;
Assert(pg_atomic_read_u32(&MyProc->clogGroupNext) == INVALID_PGPROCNO);
+ MyProc->originalXmin = InvalidTransactionId;
+ pg_atomic_init_u64(&MyProc->assignedGlobalCsn, InProgressGlobalCSN);
+
/*
* Acquire ownership of the PGPROC's latch, so that we can use WaitLatch
* on it. That allows us to repoint the process latch, which so far
@@ -568,6 +572,7 @@ InitAuxiliaryProcess(void)
MyProc->lwWaitMode = 0;
MyProc->waitLock = NULL;
MyProc->waitProcLock = NULL;
+ MyProc->originalXmin = InvalidTransactionId;
#ifdef USE_ASSERT_CHECKING
{
int i;
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 0eb04c35cb..a333cee7c2 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -28,6 +28,7 @@
#include "access/commit_ts.h"
#include "access/gin.h"
+#include "access/global_snapshot.h"
#include "access/rmgr.h"
#include "access/transam.h"
#include "access/twophase.h"
@@ -1059,7 +1060,7 @@ static struct config_bool ConfigureNamesBool[] =
gettext_noop("Used to achieve REPEATEBLE READ isolation level for postgres_fdw transactions.")
},
&track_global_snapshots,
- true, /* XXX: set true to simplify tesing. XXX2: Seems that RESOURCES_MEM isn't the best catagory */
+ false, /* XXX: Seems that RESOURCES_MEM isn't the best catagory */
NULL, NULL, NULL
},
{
@@ -2405,6 +2406,16 @@ static struct config_int ConfigureNamesInt[] =
NULL, NULL, NULL
},
+ {
+ {"global_snapshot_defer_time", PGC_POSTMASTER, REPLICATION_MASTER,
+ gettext_noop("Minimal age of records which allowed to be vacuumed, in seconds."),
+ NULL
+ },
+ &global_snapshot_defer_time,
+ 5, 0, INT_MAX,
+ NULL, NULL, NULL
+ },
+
/*
* See also CheckRequiredParameterValues() if this parameter changes
*/
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 1fa02d2c93..8ece46ae82 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -295,6 +295,8 @@
# and comma-separated list of application_name
# from standby(s); '*' = all
#vacuum_defer_cleanup_age = 0 # number of xacts by which cleanup is delayed
+#global_snapshot_defer_time = 0 # minimal age of records which allowed to be
+ # vacuumed, in seconds
# - Standby Servers -
diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c
index edf59efc29..d4cf0710fc 100644
--- a/src/backend/utils/time/snapmgr.c
+++ b/src/backend/utils/time/snapmgr.c
@@ -48,6 +48,7 @@
#include <sys/stat.h>
#include <unistd.h>
+#include "access/global_snapshot.h"
#include "access/transam.h"
#include "access/xact.h"
#include "access/xlog.h"
@@ -245,6 +246,8 @@ typedef struct SerializedSnapshotData
CommandId curcid;
TimestampTz whenTaken;
XLogRecPtr lsn;
+ GlobalCSN global_csn;
+ bool imported_global_csn;
} SerializedSnapshotData;
Size
@@ -992,7 +995,9 @@ SnapshotResetXmin(void)
pairingheap_first(&RegisteredSnapshots));
if (TransactionIdPrecedes(MyPgXact->xmin, minSnapshot->xmin))
+ {
MyPgXact->xmin = minSnapshot->xmin;
+ }
}
/*
@@ -2081,6 +2086,8 @@ SerializeSnapshot(Snapshot snapshot, char *start_address)
serialized_snapshot.curcid = snapshot->curcid;
serialized_snapshot.whenTaken = snapshot->whenTaken;
serialized_snapshot.lsn = snapshot->lsn;
+ serialized_snapshot.global_csn = snapshot->global_csn;
+ serialized_snapshot.imported_global_csn = snapshot->imported_global_csn;
/*
* Ignore the SubXID array if it has overflowed, unless the snapshot was
@@ -2155,6 +2162,8 @@ RestoreSnapshot(char *start_address)
snapshot->curcid = serialized_snapshot.curcid;
snapshot->whenTaken = serialized_snapshot.whenTaken;
snapshot->lsn = serialized_snapshot.lsn;
+ snapshot->global_csn = serialized_snapshot.global_csn;
+ snapshot->imported_global_csn = serialized_snapshot.imported_global_csn;
/* Copy XIDs, if present. */
if (serialized_snapshot.xcnt > 0)
@@ -2192,3 +2201,97 @@ RestoreTransactionSnapshot(Snapshot snapshot, void *master_pgproc)
{
SetTransactionSnapshot(snapshot, NULL, InvalidPid, master_pgproc);
}
+
+/*
+ * ExportGlobalSnapshot
+ *
+ * Export global_csn so that caller can expand this transaction to other
+ * nodes.
+ *
+ * TODO: it's better to do this through EXPORT/IMPORT SNAPSHOT syntax and
+ * add some additional checks that transaction did not yet acquired xid, but
+ * for current iteration of this patch I don't want to hack on parser.
+ */
+GlobalCSN
+ExportGlobalSnapshot()
+{
+ if (!track_global_snapshots)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("could not export global snapshot"),
+ errhint("Make sure the configuration parameter \"%s\" is enabled.",
+ "track_global_snapshots")));
+
+ return CurrentSnapshot->global_csn;
+}
+
+/* SQL accessor to ExportGlobalSnapshot() */
+Datum
+pg_global_snapshot_export(PG_FUNCTION_ARGS)
+{
+ GlobalCSN global_csn = ExportGlobalSnapshot();
+ PG_RETURN_UINT64(global_csn);
+}
+
+/*
+ * ImportGlobalSnapshot
+ *
+ * Import global_csn and retract this backends xmin to the value that was
+ * actual when we had such global_csn.
+ *
+ * TODO: it's better to do this through EXPORT/IMPORT SNAPSHOT syntax and
+ * add some additional checks that transaction did not yet acquired xid, but
+ * for current iteration of this patch I don't want to hack on parser.
+ */
+void
+ImportGlobalSnapshot(GlobalCSN snap_global_csn)
+{
+ volatile TransactionId xmin;
+
+ if (!track_global_snapshots)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("could not import global snapshot"),
+ errhint("Make sure the configuration parameter \"%s\" is enabled.",
+ "track_global_snapshots")));
+
+ if (global_snapshot_defer_time <= 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("could not import global snapshot"),
+ errhint("Make sure the configuration parameter \"%s\" is positive.",
+ "global_snapshot_defer_time")));
+
+ /*
+ * Call GlobalSnapshotToXmin under ProcArrayLock to avoid situation that
+ * resulting xmin will be evicted from map before we will set it into our
+ * backend's xmin.
+ */
+ LWLockAcquire(ProcArrayLock, LW_SHARED);
+ xmin = GlobalSnapshotToXmin(snap_global_csn);
+ if (!TransactionIdIsValid(xmin))
+ {
+ LWLockRelease(ProcArrayLock);
+ elog(ERROR, "GlobalSnapshotToXmin: global snapshot too old");
+ }
+ MyProc->originalXmin = MyPgXact->xmin;
+ MyPgXact->xmin = TransactionXmin = xmin;
+ LWLockRelease(ProcArrayLock);
+
+ CurrentSnapshot->xmin = xmin; /* defuse SnapshotResetXmin() */
+ CurrentSnapshot->global_csn = snap_global_csn;
+ CurrentSnapshot->imported_global_csn = true;
+ GlobalSnapshotSync(snap_global_csn);
+
+ Assert(TransactionIdPrecedesOrEquals(RecentGlobalXmin, xmin));
+ Assert(TransactionIdPrecedesOrEquals(RecentGlobalDataXmin, xmin));
+}
+
+/* SQL accessor to ImportGlobalSnapshot() */
+Datum
+pg_global_snapshot_import(PG_FUNCTION_ARGS)
+{
+ GlobalCSN global_csn = PG_GETARG_UINT64(0);
+ ImportGlobalSnapshot(global_csn);
+ PG_RETURN_VOID();
+}
diff --git a/src/backend/utils/time/tqual.c b/src/backend/utils/time/tqual.c
index f7c4c9188c..f2fbc77fa8 100644
--- a/src/backend/utils/time/tqual.c
+++ b/src/backend/utils/time/tqual.c
@@ -63,6 +63,7 @@
#include "postgres.h"
+#include "access/global_snapshot.h"
#include "access/htup_details.h"
#include "access/multixact.h"
#include "access/subtrans.h"
@@ -1462,8 +1463,8 @@ HeapTupleIsSurelyDead(HeapTuple htup, TransactionId OldestXmin)
}
/*
- * XidInMVCCSnapshot
- * Is the given XID still-in-progress according to the snapshot?
+ * XidInLocalMVCCSnapshot
+ * Is the given XID still-in-progress according to the local snapshot?
*
* Note: GetSnapshotData never stores either top xid or subxids of our own
* backend into a snapshot, so these xids will not be reported as "running"
@@ -1471,8 +1472,8 @@ HeapTupleIsSurelyDead(HeapTuple htup, TransactionId OldestXmin)
* TransactionIdIsCurrentTransactionId first, except when it's known the
* XID could not be ours anyway.
*/
-bool
-XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
+static bool
+XidInLocalMVCCSnapshot(TransactionId xid, Snapshot snapshot)
{
uint32 i;
@@ -1583,6 +1584,62 @@ XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
return false;
}
+/*
+ * XidInMVCCSnapshot
+ *
+ * Check whether this xid is in snapshot, taking into account fact that
+ * snapshot can be global. When track_global_snapshots is switched off
+ * just call XidInLocalMVCCSnapshot().
+ */
+bool
+XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
+{
+ bool in_snapshot;
+
+ if (snapshot->imported_global_csn)
+ {
+ Assert(track_global_snapshots);
+ /* No point to using snapshot info except CSN */
+ return XidInvisibleInGlobalSnapshot(xid, snapshot);
+ }
+
+ in_snapshot = XidInLocalMVCCSnapshot(xid, snapshot);
+
+ if (!track_global_snapshots)
+ {
+ Assert(GlobalCSNIsFrozen(snapshot->global_csn));
+ return in_snapshot;
+ }
+
+ if (in_snapshot)
+ {
+ /*
+ * This xid may be already in unknown state and in that case
+ * we must wait and recheck.
+ *
+ * TODO: this check can be skipped if we know for sure that there were
+ * no global transactions when this snapshot was taken. That requires
+ * some changes to mechanisms of global snapshots exprot/import (if
+ * backend set xmin then we should have a-priori knowledge that this
+ * transaction going to be global or local -- right now this is not
+ * enforced). Leave that for future and don't complicate this patch.
+ */
+ return XidInvisibleInGlobalSnapshot(xid, snapshot);
+ }
+ else
+ {
+#ifdef USE_ASSERT_CHECKING
+ /* Check that global snapshot gives the same results as local one */
+ if (XidInvisibleInGlobalSnapshot(xid, snapshot))
+ {
+ GlobalCSN gcsn = TransactionIdGetGlobalCSN(xid);
+ Assert(GlobalCSNIsAborted(gcsn));
+ }
+#endif
+ return false;
+ }
+}
+
/*
* Is the tuple really only locked? That is, is it not updated?
*
diff --git a/src/include/access/global_snapshot.h b/src/include/access/global_snapshot.h
new file mode 100644
index 0000000000..246b180cfd
--- /dev/null
+++ b/src/include/access/global_snapshot.h
@@ -0,0 +1,72 @@
+/*-------------------------------------------------------------------------
+ *
+ * global_snapshot.h
+ * Support for cross-node snapshot isolation.
+ *
+ * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/access/global_snapshot.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef GLOBAL_SNAPSHOT_H
+#define GLOBAL_SNAPSHOT_H
+
+#include "port/atomics.h"
+#include "storage/lock.h"
+#include "utils/snapshot.h"
+#include "utils/guc.h"
+
+/*
+ * snapshot.h is used in frontend code so atomic variant of GlobalCSN type
+ * is defined here.
+ */
+typedef pg_atomic_uint64 GlobalCSN_atomic;
+
+#define InProgressGlobalCSN UINT64CONST(0x0)
+#define AbortedGlobalCSN UINT64CONST(0x1)
+#define FrozenGlobalCSN UINT64CONST(0x2)
+#define InDoubtGlobalCSN UINT64CONST(0x3)
+#define FirstNormalGlobalCSN UINT64CONST(0x4)
+
+#define GlobalCSNIsInProgress(csn) ((csn) == InProgressGlobalCSN)
+#define GlobalCSNIsAborted(csn) ((csn) == AbortedGlobalCSN)
+#define GlobalCSNIsFrozen(csn) ((csn) == FrozenGlobalCSN)
+#define GlobalCSNIsInDoubt(csn) ((csn) == InDoubtGlobalCSN)
+#define GlobalCSNIsNormal(csn) ((csn) >= FirstNormalGlobalCSN)
+
+
+extern int global_snapshot_defer_time;
+
+
+extern Size GlobalSnapshotShmemSize(void);
+extern void GlobalSnapshotShmemInit(void);
+extern void GlobalSnapshotStartup(TransactionId oldestActiveXID);
+
+extern void GlobalSnapshotMapXmin(GlobalCSN snapshot_global_csn);
+extern TransactionId GlobalSnapshotToXmin(GlobalCSN snapshot_global_csn);
+
+extern GlobalCSN GlobalSnapshotGenerate(bool locked);
+
+extern bool XidInvisibleInGlobalSnapshot(TransactionId xid, Snapshot snapshot);
+
+extern void GlobalSnapshotSync(GlobalCSN remote_gcsn);
+
+extern GlobalCSN TransactionIdGetGlobalCSN(TransactionId xid);
+
+extern GlobalCSN GlobalSnapshotPrepareGlobal(const char *gid);
+extern void GlobalSnapshotAssignCsnGlobal(const char *gid,
+ GlobalCSN global_csn);
+
+extern GlobalCSN GlobalSnapshotPrepareCurrent(void);
+extern void GlobalSnapshotAssignCsnCurrent(GlobalCSN global_csn);
+
+extern void GlobalSnapshotAbort(PGPROC *proc, TransactionId xid, int nsubxids,
+ TransactionId *subxids);
+extern void GlobalSnapshotPrecommit(PGPROC *proc, TransactionId xid, int nsubxids,
+ TransactionId *subxids);
+extern void GlobalSnapshotCommit(PGPROC *proc, TransactionId xid, int nsubxids,
+ TransactionId *subxids);
+
+#endif /* GLOBAL_SNAPSHOT_H */
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index 0e932daa48..f8b774f393 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -18,6 +18,7 @@
#include "access/xact.h"
#include "datatype/timestamp.h"
#include "storage/lock.h"
+#include "utils/snapshot.h"
/*
* GlobalTransactionData is defined in twophase.c; other places have no
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 034a41eb55..2fc3d35f9e 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -10048,4 +10048,18 @@
proargnames => '{rootrelid,relid,parentrelid,isleaf,level}',
prosrc => 'pg_partition_tree' },
+# global transaction handling
+{ oid => '3430', descr => 'export global transaction snapshot',
+ proname => 'pg_global_snapshot_export', provolatile => 'v', proparallel => 'u',
+ prorettype => 'int8', proargtypes => '', prosrc => 'pg_global_snapshot_export' },
+{ oid => '3431', descr => 'import global transaction snapshot',
+ proname => 'pg_global_snapshot_import', provolatile => 'v', proparallel => 'u',
+ prorettype => 'void', proargtypes => 'int8', prosrc => 'pg_global_snapshot_import' },
+{ oid => '3432', descr => 'prepare distributed transaction for commit, get global_csn',
+ proname => 'pg_global_snapshot_prepare', provolatile => 'v', proparallel => 'u',
+ prorettype => 'int8', proargtypes => 'text', prosrc => 'pg_global_snapshot_prepare' },
+{ oid => '3433', descr => 'assign global_csn to distributed transaction',
+ proname => 'pg_global_snapshot_assign', provolatile => 'v', proparallel => 'u',
+ prorettype => 'void', proargtypes => 'text int8', prosrc => 'pg_global_snapshot_assign' },
+
]
diff --git a/src/include/datatype/timestamp.h b/src/include/datatype/timestamp.h
index f5b6026ef5..75ec93b46b 100644
--- a/src/include/datatype/timestamp.h
+++ b/src/include/datatype/timestamp.h
@@ -93,6 +93,9 @@ typedef struct
#define USECS_PER_MINUTE INT64CONST(60000000)
#define USECS_PER_SEC INT64CONST(1000000)
+#define NSECS_PER_SEC INT64CONST(1000000000)
+#define NSECS_PER_USEC INT64CONST(1000)
+
/*
* We allow numeric timezone offsets up to 15:59:59 either way from Greenwich.
* Currently, the record holders for wackiest offsets in actual use are zones
diff --git a/src/include/fmgr.h b/src/include/fmgr.h
index 101f513ba6..3026e71f83 100644
--- a/src/include/fmgr.h
+++ b/src/include/fmgr.h
@@ -250,6 +250,7 @@ extern struct varlena *pg_detoast_datum_packed(struct varlena *datum);
#define PG_GETARG_FLOAT4(n) DatumGetFloat4(PG_GETARG_DATUM(n))
#define PG_GETARG_FLOAT8(n) DatumGetFloat8(PG_GETARG_DATUM(n))
#define PG_GETARG_INT64(n) DatumGetInt64(PG_GETARG_DATUM(n))
+#define PG_GETARG_UINT64(n) DatumGetUInt64(PG_GETARG_DATUM(n))
/* use this if you want the raw, possibly-toasted input datum: */
#define PG_GETARG_RAW_VARLENA_P(n) ((struct varlena *) PG_GETARG_POINTER(n))
/* use this if you want the input datum de-toasted: */
diff --git a/src/include/portability/instr_time.h b/src/include/portability/instr_time.h
index 60984905b2..9edb79ad6e 100644
--- a/src/include/portability/instr_time.h
+++ b/src/include/portability/instr_time.h
@@ -141,6 +141,9 @@ typedef struct timespec instr_time;
#define INSTR_TIME_GET_MICROSEC(t) \
(((uint64) (t).tv_sec * (uint64) 1000000) + (uint64) ((t).tv_nsec / 1000))
+#define INSTR_TIME_GET_NANOSEC(t) \
+ (((uint64) (t).tv_sec * (uint64) 1000000000) + (uint64) ((t).tv_nsec))
+
#else /* !HAVE_CLOCK_GETTIME */
/* Use gettimeofday() */
@@ -205,6 +208,10 @@ typedef struct timeval instr_time;
#define INSTR_TIME_GET_MICROSEC(t) \
(((uint64) (t).tv_sec * (uint64) 1000000) + (uint64) (t).tv_usec)
+#define INSTR_TIME_GET_NANOSEC(t) \
+ (((uint64) (t).tv_sec * (uint64) 1000000000) + \
+ (uint64) (t).tv_usec * (uint64) 1000)
+
#endif /* HAVE_CLOCK_GETTIME */
#else /* WIN32 */
@@ -237,6 +244,9 @@ typedef LARGE_INTEGER instr_time;
#define INSTR_TIME_GET_MICROSEC(t) \
((uint64) (((double) (t).QuadPart * 1000000.0) / GetTimerFrequency()))
+#define INSTR_TIME_GET_NANOSEC(t) \
+ ((uint64) (((double) (t).QuadPart * 1000000000.0) / GetTimerFrequency()))
+
static inline double
GetTimerFrequency(void)
{
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index cb613c8076..944e0c9cc7 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -15,8 +15,10 @@
#define _PROC_H_
#include "access/clog.h"
+#include "access/global_snapshot.h"
#include "access/xlogdefs.h"
#include "lib/ilist.h"
+#include "utils/snapshot.h"
#include "storage/latch.h"
#include "storage/lock.h"
#include "storage/pg_sema.h"
@@ -57,6 +59,7 @@ struct XidCache
#define PROC_IN_LOGICAL_DECODING 0x10 /* currently doing logical
* decoding outside xact */
#define PROC_RESERVED 0x20 /* reserved for procarray */
+#define PROC_RESERVED2 0x40 /* reserved for procarray */
/* flags reset at EOXact */
#define PROC_VACUUM_STATE_MASK \
@@ -203,6 +206,18 @@ struct PGPROC
PGPROC *lockGroupLeader; /* lock group leader, if I'm a member */
dlist_head lockGroupMembers; /* list of members, if I'm a leader */
dlist_node lockGroupLink; /* my member link, if I'm a member */
+
+ /*
+ * assignedGlobalCsn holds GlobalCSN for this transaction. It is generated
+ * under a ProcArray lock and later is writter to a GlobalCSNLog. This
+ * variable defined as atomic only for case of group commit, in all other
+ * scenarios only backend responsible for this proc entry is working with
+ * this variable.
+ */
+ GlobalCSN_atomic assignedGlobalCsn;
+
+ /* Original xmin of this backend before global snapshot was imported */
+ TransactionId originalXmin;
};
/* NOTE: "typedef struct PGPROC PGPROC" appears in storage/lock.h. */
diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h
index 75bab2985f..e68a87575e 100644
--- a/src/include/storage/procarray.h
+++ b/src/include/storage/procarray.h
@@ -36,6 +36,10 @@
#define PROCARRAY_SLOTS_XMIN 0x20 /* replication slot xmin,
* catalog_xmin */
+
+#define PROCARRAY_NON_IMPORTED_XMIN 0x40 /* use originalXmin instead
+ * of xmin to properly
+ * maintain gsXidMap */
/*
* Only flags in PROCARRAY_PROC_FLAGS_MASK are considered when matching
* PGXACT->vacuumFlags. Other flags are used for different purposes and
@@ -124,4 +128,8 @@ extern void ProcArraySetReplicationSlotXmin(TransactionId xmin,
extern void ProcArrayGetReplicationSlotXmin(TransactionId *xmin,
TransactionId *catalog_xmin);
+extern void ProcArraySetGlobalSnapshotXmin(TransactionId xmin);
+
+extern TransactionId ProcArrayGetGlobalSnapshotXmin(void);
+
#endif /* PROCARRAY_H */
diff --git a/src/include/utils/snapmgr.h b/src/include/utils/snapmgr.h
index 83806f3040..1a066fd8d8 100644
--- a/src/include/utils/snapmgr.h
+++ b/src/include/utils/snapmgr.h
@@ -87,6 +87,9 @@ extern void AtSubCommit_Snapshot(int level);
extern void AtSubAbort_Snapshot(int level);
extern void AtEOXact_Snapshot(bool isCommit, bool resetXmin);
+extern GlobalCSN ExportGlobalSnapshot(void);
+extern void ImportGlobalSnapshot(GlobalCSN snap_global_csn);
+
extern void ImportSnapshot(const char *idstr);
extern bool XactHasExportedSnapshots(void);
extern void DeleteAllExportedSnapshotFiles(void);
diff --git a/src/include/utils/snapshot.h b/src/include/utils/snapshot.h
index 318d41e6f7..1563407824 100644
--- a/src/include/utils/snapshot.h
+++ b/src/include/utils/snapshot.h
@@ -115,6 +115,14 @@ typedef struct SnapshotData
TimestampTz whenTaken; /* timestamp when snapshot was taken */
XLogRecPtr lsn; /* position in the WAL stream when taken */
+
+ /*
+ * GlobalCSN for cross-node snapshot isolation support.
+ * Will be used only if track_global_snapshots is enabled.
+ */
+ GlobalCSN global_csn;
+ /* Did we have our own global_csn or imported one from different node */
+ bool imported_global_csn;
} SnapshotData;
/*
--
2.19.2
[application/octet-stream] 0003-postgres_fdw-support-for-global-snapshots-v3.patch (32.1K, ../../[email protected]/4-0003-postgres_fdw-support-for-global-snapshots-v3.patch)
download | inline diff:
From 657e797c3081c6628f8b2abe077109c7fe6067db Mon Sep 17 00:00:00 2001
From: Stas Kelvich <[email protected]>
Date: Wed, 25 Apr 2018 16:39:09 +0300
Subject: [PATCH 3/3] postgres_fdw support for global snapshots v3
---
contrib/postgres_fdw/Makefile | 9 +
contrib/postgres_fdw/connection.c | 292 ++++++++++++++++--
contrib/postgres_fdw/postgres_fdw.c | 12 +
contrib/postgres_fdw/postgres_fdw.h | 2 +
.../postgres_fdw/t/001_bank_coordinator.pl | 264 ++++++++++++++++
.../postgres_fdw/t/002_bank_participant.pl | 240 ++++++++++++++
src/test/perl/PostgresNode.pm | 35 +++
7 files changed, 827 insertions(+), 27 deletions(-)
create mode 100644 contrib/postgres_fdw/t/001_bank_coordinator.pl
create mode 100644 contrib/postgres_fdw/t/002_bank_participant.pl
diff --git a/contrib/postgres_fdw/Makefile b/contrib/postgres_fdw/Makefile
index 85394b4f1f..02ae067cd0 100644
--- a/contrib/postgres_fdw/Makefile
+++ b/contrib/postgres_fdw/Makefile
@@ -23,3 +23,12 @@ top_builddir = ../..
include $(top_builddir)/src/Makefile.global
include $(top_srcdir)/contrib/contrib-global.mk
endif
+
+# Global makefile will do temp-install for 'check'. Since REGRESS is defined,
+# PGXS (included from contrib-global.mk or directly) will care to add
+# postgres_fdw to it as EXTRA_INSTALL and build pg_regress. It will also
+# actually run pg_regress, so the only thing left is tap tests.
+check: tapcheck
+
+tapcheck: temp-install
+ $(prove_check)
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index a6509932dc..f6c43ad518 100644
--- a/contrib/postgres_fdw/connection.c
+++ b/contrib/postgres_fdw/connection.c
@@ -14,9 +14,11 @@
#include "postgres_fdw.h"
+#include "access/global_snapshot.h"
#include "access/htup_details.h"
#include "catalog/pg_user_mapping.h"
#include "access/xact.h"
+#include "access/xlog.h" /* GetSystemIdentifier() */
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.h"
@@ -24,6 +26,8 @@
#include "utils/hsearch.h"
#include "utils/inval.h"
#include "utils/memutils.h"
+#include "utils/snapmgr.h"
+#include "utils/snapshot.h"
#include "utils/syscache.h"
@@ -65,6 +69,21 @@ typedef struct ConnCacheEntry
*/
static HTAB *ConnectionHash = NULL;
+/*
+ * FdwTransactionState
+ *
+ * Holds number of open remote transactions and shared state
+ * needed for all connection entries.
+ */
+typedef struct FdwTransactionState
+{
+ char *gid;
+ int nparticipants;
+ GlobalCSN global_csn;
+ bool two_phase_commit;
+} FdwTransactionState;
+static FdwTransactionState *fdwTransState;
+
/* for assigning cursor numbers and prepared statement numbers */
static unsigned int cursor_number = 0;
static unsigned int prep_stmt_number = 0;
@@ -72,6 +91,9 @@ static unsigned int prep_stmt_number = 0;
/* tracks whether any work is needed in callback functions */
static bool xact_got_connection = false;
+/* counter of prepared tx made by this backend */
+static int two_phase_xact_count = 0;
+
/* prototypes of private functions */
static PGconn *connect_pg_server(ForeignServer *server, UserMapping *user);
static void disconnect_pg_server(ConnCacheEntry *entry);
@@ -80,6 +102,7 @@ static void configure_remote_session(PGconn *conn);
static void do_sql_command(PGconn *conn, const char *sql);
static void begin_remote_xact(ConnCacheEntry *entry);
static void pgfdw_xact_callback(XactEvent event, void *arg);
+static void deallocate_prepared_stmts(ConnCacheEntry *entry);
static void pgfdw_subxact_callback(SubXactEvent event,
SubTransactionId mySubid,
SubTransactionId parentSubid,
@@ -136,6 +159,15 @@ GetConnection(UserMapping *user, bool will_prep_stmt)
pgfdw_inval_callback, (Datum) 0);
}
+ /* allocate FdwTransactionState */
+ if (fdwTransState == NULL)
+ {
+ MemoryContext oldcxt;
+ oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
+ fdwTransState = palloc0(sizeof(FdwTransactionState));
+ MemoryContextSwitchTo(oldcxt);
+ }
+
/* Set flag that we did GetConnection during the current transaction */
xact_got_connection = true;
@@ -388,7 +420,8 @@ configure_remote_session(PGconn *conn)
}
/*
- * Convenience subroutine to issue a non-data-returning SQL command to remote
+ * Convenience subroutine to issue a non-data-returning SQL command or
+ * statement to remote node.
*/
static void
do_sql_command(PGconn *conn, const char *sql)
@@ -398,7 +431,8 @@ do_sql_command(PGconn *conn, const char *sql)
if (!PQsendQuery(conn, sql))
pgfdw_report_error(ERROR, NULL, conn, false, sql);
res = pgfdw_get_result(conn, sql);
- if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ if (PQresultStatus(res) != PGRES_COMMAND_OK &&
+ PQresultStatus(res) != PGRES_TUPLES_OK)
pgfdw_report_error(ERROR, res, conn, true, sql);
PQclear(res);
}
@@ -426,6 +460,10 @@ begin_remote_xact(ConnCacheEntry *entry)
elog(DEBUG3, "starting remote transaction on connection %p",
entry->conn);
+ if (UseGlobalSnapshots && (!IsolationUsesXactSnapshot() ||
+ IsolationIsSerializable()))
+ elog(ERROR, "Global snapshots support only REPEATABLE READ");
+
if (IsolationIsSerializable())
sql = "START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
else
@@ -434,6 +472,23 @@ begin_remote_xact(ConnCacheEntry *entry)
do_sql_command(entry->conn, sql);
entry->xact_depth = 1;
entry->changing_xact_state = false;
+
+ if (UseGlobalSnapshots)
+ {
+ char import_sql[128];
+
+ /* Export our snapshot */
+ if (fdwTransState->global_csn == 0)
+ fdwTransState->global_csn = ExportGlobalSnapshot();
+
+ snprintf(import_sql, sizeof(import_sql),
+ "SELECT pg_global_snapshot_import("UINT64_FORMAT")",
+ fdwTransState->global_csn);
+
+ do_sql_command(entry->conn, import_sql);
+ }
+
+ fdwTransState->nparticipants += 1;
}
/*
@@ -644,6 +699,94 @@ pgfdw_report_error(int elevel, PGresult *res, PGconn *conn,
PQclear(res);
}
+/* Callback typedef for BroadcastStmt */
+typedef bool (*BroadcastCmdResHandler) (PGresult *result, void *arg);
+
+/* Broadcast sql in parallel to all ConnectionHash entries */
+static bool
+BroadcastStmt(char const * sql, unsigned expectedStatus,
+ BroadcastCmdResHandler handler, void *arg)
+{
+ HASH_SEQ_STATUS scan;
+ ConnCacheEntry *entry;
+ bool allOk = true;
+
+ /* Broadcast sql */
+ hash_seq_init(&scan, ConnectionHash);
+ while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
+ {
+ pgfdw_reject_incomplete_xact_state_change(entry);
+
+ if (entry->xact_depth > 0 && entry->conn != NULL)
+ {
+ if (!PQsendQuery(entry->conn, sql))
+ {
+ PGresult *res = PQgetResult(entry->conn);
+
+ elog(WARNING, "Failed to send command %s", sql);
+ pgfdw_report_error(WARNING, res, entry->conn, true, sql);
+ PQclear(res);
+ }
+ }
+ }
+
+ /* Collect responses */
+ hash_seq_init(&scan, ConnectionHash);
+ while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
+ {
+ if (entry->xact_depth > 0 && entry->conn != NULL)
+ {
+ PGresult *result = PQgetResult(entry->conn);
+
+ if (PQresultStatus(result) != expectedStatus ||
+ (handler && !handler(result, arg)))
+ {
+ elog(WARNING, "Failed command %s: status=%d, expected status=%d", sql, PQresultStatus(result), expectedStatus);
+ pgfdw_report_error(ERROR, result, entry->conn, true, sql);
+ allOk = false;
+ }
+ PQclear(result);
+ PQgetResult(entry->conn); /* consume NULL result */
+ }
+ }
+
+ return allOk;
+}
+
+/* Wrapper for broadcasting commands */
+static bool
+BroadcastCmd(char const *sql)
+{
+ return BroadcastStmt(sql, PGRES_COMMAND_OK, NULL, NULL);
+}
+
+/* Wrapper for broadcasting statements */
+static bool
+BroadcastFunc(char const *sql)
+{
+ return BroadcastStmt(sql, PGRES_TUPLES_OK, NULL, NULL);
+}
+
+/* Callback for selecting maximal csn */
+static bool
+MaxCsnCB(PGresult *result, void *arg)
+{
+ char *resp;
+ GlobalCSN *max_csn = (GlobalCSN *) arg;
+ GlobalCSN csn = 0;
+
+ resp = PQgetvalue(result, 0, 0);
+
+ if (resp == NULL || (*resp) == '\0' ||
+ sscanf(resp, UINT64_FORMAT, &csn) != 1)
+ return false;
+
+ if (*max_csn < csn)
+ *max_csn = csn;
+
+ return true;
+}
+
/*
* pgfdw_xact_callback --- cleanup at main-transaction end.
*/
@@ -657,6 +800,86 @@ pgfdw_xact_callback(XactEvent event, void *arg)
if (!xact_got_connection)
return;
+ /* Handle possible two-phase commit */
+ if (event == XACT_EVENT_PARALLEL_PRE_COMMIT || event == XACT_EVENT_PRE_COMMIT)
+ {
+ bool include_local_tx = false;
+
+ /* Should we take into account this node? */
+ if (TransactionIdIsValid(GetCurrentTransactionIdIfAny()))
+ {
+ include_local_tx = true;
+ fdwTransState->nparticipants += 1;
+ }
+
+ /* Switch to 2PC mode there were more than one participant */
+ if (UseGlobalSnapshots && fdwTransState->nparticipants > 1)
+ fdwTransState->two_phase_commit = true;
+
+ if (fdwTransState->two_phase_commit)
+ {
+ GlobalCSN max_csn = InProgressGlobalCSN,
+ my_csn = InProgressGlobalCSN;
+ bool res;
+ char *sql;
+
+ fdwTransState->gid = psprintf("pgfdw:%lld:%llu:%d:%u:%d:%d",
+ (long long) GetCurrentTimestamp(),
+ (long long) GetSystemIdentifier(),
+ MyProcPid,
+ GetCurrentTransactionIdIfAny(),
+ ++two_phase_xact_count,
+ fdwTransState->nparticipants);
+
+ /* Broadcast PREPARE */
+ sql = psprintf("PREPARE TRANSACTION '%s'", fdwTransState->gid);
+ res = BroadcastCmd(sql);
+ if (!res)
+ goto error;
+
+ /* Broadcast pg_global_snapshot_prepare() */
+ if (include_local_tx)
+ my_csn = GlobalSnapshotPrepareCurrent();
+
+ sql = psprintf("SELECT pg_global_snapshot_prepare('%s')",
+ fdwTransState->gid);
+ res = BroadcastStmt(sql, PGRES_TUPLES_OK, MaxCsnCB, &max_csn);
+ if (!res)
+ goto error;
+
+ /* select maximal global csn */
+ if (include_local_tx && my_csn > max_csn)
+ max_csn = my_csn;
+
+ /* Broadcast pg_global_snapshot_assign() */
+ if (include_local_tx)
+ GlobalSnapshotAssignCsnCurrent(max_csn);
+ sql = psprintf("SELECT pg_global_snapshot_assign('%s',"UINT64_FORMAT")",
+ fdwTransState->gid, max_csn);
+ res = BroadcastFunc(sql);
+
+error:
+ if (!res)
+ {
+ sql = psprintf("ABORT PREPARED '%s'", fdwTransState->gid);
+ BroadcastCmd(sql);
+ elog(ERROR, "Failed to PREPARE transaction on remote node");
+ }
+
+ /*
+ * Do not fall down. Consequent COMMIT event will clean thing up.
+ */
+ return;
+ }
+ }
+
+ /* COMMIT open transaction of we were doing 2PC */
+ if (fdwTransState->two_phase_commit &&
+ (event == XACT_EVENT_PARALLEL_COMMIT || event == XACT_EVENT_COMMIT))
+ {
+ BroadcastCmd(psprintf("COMMIT PREPARED '%s'", fdwTransState->gid));
+ }
+
/*
* Scan all connection cache entries to find open remote transactions, and
* close them.
@@ -664,8 +887,6 @@ pgfdw_xact_callback(XactEvent event, void *arg)
hash_seq_init(&scan, ConnectionHash);
while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
{
- PGresult *res;
-
/* Ignore cache entry if no open connection right now */
if (entry->conn == NULL)
continue;
@@ -682,6 +903,7 @@ pgfdw_xact_callback(XactEvent event, void *arg)
{
case XACT_EVENT_PARALLEL_PRE_COMMIT:
case XACT_EVENT_PRE_COMMIT:
+ Assert(!fdwTransState->two_phase_commit);
/*
* If abort cleanup previously failed for this connection,
@@ -694,28 +916,7 @@ pgfdw_xact_callback(XactEvent event, void *arg)
do_sql_command(entry->conn, "COMMIT TRANSACTION");
entry->changing_xact_state = false;
- /*
- * If there were any errors in subtransactions, and we
- * made prepared statements, do a DEALLOCATE ALL to make
- * sure we get rid of all prepared statements. This is
- * annoying and not terribly bulletproof, but it's
- * probably not worth trying harder.
- *
- * DEALLOCATE ALL only exists in 8.3 and later, so this
- * constrains how old a server postgres_fdw can
- * communicate with. We intentionally ignore errors in
- * the DEALLOCATE, so that we can hobble along to some
- * extent with older servers (leaking prepared statements
- * as we go; but we don't really support update operations
- * pre-8.3 anyway).
- */
- if (entry->have_prep_stmt && entry->have_error)
- {
- res = PQexec(entry->conn, "DEALLOCATE ALL");
- PQclear(res);
- }
- entry->have_prep_stmt = false;
- entry->have_error = false;
+ deallocate_prepared_stmts(entry);
break;
case XACT_EVENT_PRE_PREPARE:
@@ -730,10 +931,15 @@ pgfdw_xact_callback(XactEvent event, void *arg)
*/
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("cannot prepare a transaction that modified remote tables")));
+ errmsg("cannot prepare a transaction that modified remote tables")));
break;
case XACT_EVENT_PARALLEL_COMMIT:
case XACT_EVENT_COMMIT:
+ if (fdwTransState->two_phase_commit)
+ deallocate_prepared_stmts(entry);
+ else /* Pre-commit should have closed the open transaction */
+ elog(ERROR, "missed cleaning up connection during pre-commit");
+ break;
case XACT_EVENT_PREPARE:
/* Pre-commit should have closed the open transaction */
elog(ERROR, "missed cleaning up connection during pre-commit");
@@ -829,6 +1035,38 @@ pgfdw_xact_callback(XactEvent event, void *arg)
/* Also reset cursor numbering for next transaction */
cursor_number = 0;
+
+ /* Reset fdwTransState */
+ memset(fdwTransState, '\0', sizeof(FdwTransactionState));
+}
+
+/*
+ * If there were any errors in subtransactions, and we
+ * made prepared statements, do a DEALLOCATE ALL to make
+ * sure we get rid of all prepared statements. This is
+ * annoying and not terribly bulletproof, but it's
+ * probably not worth trying harder.
+ *
+ * DEALLOCATE ALL only exists in 8.3 and later, so this
+ * constrains how old a server postgres_fdw can
+ * communicate with. We intentionally ignore errors in
+ * the DEALLOCATE, so that we can hobble along to some
+ * extent with older servers (leaking prepared statements
+ * as we go; but we don't really support update operations
+ * pre-8.3 anyway).
+ */
+static void
+deallocate_prepared_stmts(ConnCacheEntry *entry)
+{
+ PGresult *res;
+
+ if (entry->have_prep_stmt && entry->have_error)
+ {
+ res = PQexec(entry->conn, "DEALLOCATE ALL");
+ PQclear(res);
+ }
+ entry->have_prep_stmt = false;
+ entry->have_error = false;
}
/*
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index d22c974f0f..9e160efdef 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -269,6 +269,9 @@ typedef struct
List *already_used; /* expressions already dealt with */
} ec_member_foreign_arg;
+bool UseGlobalSnapshots;
+void _PG_init(void);
+
/*
* SQL functions
*/
@@ -5772,3 +5775,12 @@ find_em_expr_for_rel(EquivalenceClass *ec, RelOptInfo *rel)
/* We didn't find any suitable equivalence class expression */
return NULL;
}
+
+void
+_PG_init(void)
+{
+ DefineCustomBoolVariable("postgres_fdw.use_global_snapshots",
+ "Use global snapshots for FDW transactions", NULL,
+ &UseGlobalSnapshots, false, PGC_USERSET, 0, NULL,
+ NULL, NULL);
+}
diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h
index 70b538e2f9..8cf5b12798 100644
--- a/contrib/postgres_fdw/postgres_fdw.h
+++ b/contrib/postgres_fdw/postgres_fdw.h
@@ -186,4 +186,6 @@ extern const char *get_jointype_name(JoinType jointype);
extern bool is_builtin(Oid objectId);
extern bool is_shippable(Oid objectId, Oid classId, PgFdwRelationInfo *fpinfo);
+extern bool UseGlobalSnapshots;
+
#endif /* POSTGRES_FDW_H */
diff --git a/contrib/postgres_fdw/t/001_bank_coordinator.pl b/contrib/postgres_fdw/t/001_bank_coordinator.pl
new file mode 100644
index 0000000000..1e31f33349
--- /dev/null
+++ b/contrib/postgres_fdw/t/001_bank_coordinator.pl
@@ -0,0 +1,264 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+use Test::More tests => 3;
+
+my $master = get_new_node("master");
+$master->init;
+$master->append_conf('postgresql.conf', qq(
+ max_prepared_transactions = 30
+ log_checkpoints = true
+ postgres_fdw.use_global_snapshots = on
+ track_global_snapshots = on
+ default_transaction_isolation = 'REPEATABLE READ'
+));
+$master->start;
+
+my $shard1 = get_new_node("shard1");
+$shard1->init;
+$shard1->append_conf('postgresql.conf', qq(
+ max_prepared_transactions = 30
+ global_snapshot_defer_time = 15
+ track_global_snapshots = on
+));
+$shard1->start;
+
+my $shard2 = get_new_node("shard2");
+$shard2->init;
+$shard2->append_conf('postgresql.conf', qq(
+ max_prepared_transactions = 30
+ global_snapshot_defer_time = 15
+ track_global_snapshots = on
+));
+$shard2->start;
+
+###############################################################################
+# Prepare nodes
+###############################################################################
+
+$master->safe_psql('postgres', qq[
+ CREATE EXTENSION postgres_fdw;
+ CREATE TABLE accounts(id integer primary key, amount integer);
+ CREATE TABLE global_transactions(tx_time timestamp);
+]);
+
+foreach my $node ($shard1, $shard2)
+{
+ my $port = $node->port;
+ my $host = $node->host;
+
+ $node->safe_psql('postgres',
+ "CREATE TABLE accounts(id integer primary key, amount integer)");
+
+ $master->safe_psql('postgres', qq[
+ CREATE SERVER shard_$port FOREIGN DATA WRAPPER postgres_fdw options(dbname 'postgres', host '$host', port '$port');
+ CREATE FOREIGN TABLE accounts_fdw_$port() inherits (accounts) server shard_$port options(table_name 'accounts');
+ CREATE USER MAPPING for CURRENT_USER SERVER shard_$port;
+ ])
+}
+
+$shard1->safe_psql('postgres', qq[
+ insert into accounts select 2*id-1, 0 from generate_series(1, 10010) as id;
+ CREATE TABLE local_transactions(tx_time timestamp);
+]);
+
+$shard2->safe_psql('postgres', qq[
+ insert into accounts select 2*id, 0 from generate_series(1, 10010) as id;
+ CREATE TABLE local_transactions(tx_time timestamp);
+]);
+
+diag("master: @{[$master->connstr('postgres')]}");
+diag("shard1: @{[$shard1->connstr('postgres')]}");
+diag("shard2: @{[$shard2->connstr('postgres')]}");
+
+###############################################################################
+# pgbench scripts
+###############################################################################
+
+my $bank = File::Temp->new();
+append_to_file($bank, q{
+ \set id random(1, 20000)
+ BEGIN;
+ WITH upd AS (UPDATE accounts SET amount = amount - 1 WHERE id = :id RETURNING *)
+ INSERT into global_transactions SELECT now() FROM upd;
+ UPDATE accounts SET amount = amount + 1 WHERE id = (:id + 1);
+ COMMIT;
+});
+
+my $bank1 = File::Temp->new();
+append_to_file($bank1, q{
+ \set id random(1, 10000)
+ BEGIN;
+ WITH upd AS (UPDATE accounts SET amount = amount - 1 WHERE id = (2*:id + 1) RETURNING *)
+ INSERT into local_transactions SELECT now() FROM upd;
+ UPDATE accounts SET amount = amount + 1 WHERE id = (2*:id + 3);
+ COMMIT;
+});
+
+my $bank2 = File::Temp->new();
+append_to_file($bank2, q{
+ \set id random(1, 10000)
+
+ BEGIN;
+ WITH upd AS (UPDATE accounts SET amount = amount - 1 WHERE id = 2*:id RETURNING *)
+ INSERT into local_transactions SELECT now() FROM upd;
+ UPDATE accounts SET amount = amount + 1 WHERE id = (2*:id + 2);
+ COMMIT;
+});
+
+###############################################################################
+# Helpers
+###############################################################################
+
+sub count_and_delete_rows
+{
+ my ($node, $table) = @_;
+ my $count;
+
+ $count = $node->safe_psql('postgres',"select count(*) from $table");
+ $node->safe_psql('postgres',"delete from $table");
+ diag($node->name, ": completed $count transactions");
+ return $count;
+}
+
+###############################################################################
+# Concurrent global transactions
+###############################################################################
+
+my ($err, $rc);
+my $started;
+my $seconds = 30;
+my $selects;
+my $total = '0';
+my $oldtotal = '0';
+my $isolation_errors = 0;
+
+
+my $pgb_handle;
+
+$pgb_handle = $master->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+
+$started = time();
+$selects = 0;
+while (time() - $started < $seconds)
+{
+ $total = $master->safe_psql('postgres', "select sum(amount) from accounts");
+ if ( ($total ne $oldtotal) and ($total ne '') )
+ {
+ $isolation_errors++;
+ $oldtotal = $total;
+ diag("Isolation error. Total = $total");
+ }
+ if ($total ne '') { $selects++; }
+}
+
+$master->pgbench_await($pgb_handle);
+
+# sanity check
+diag("completed $selects selects");
+die "no actual transactions happend" unless ( $selects > 0 &&
+ count_and_delete_rows($master, 'global_transactions') > 0);
+
+is($isolation_errors, 0, 'isolation between concurrent global transaction');
+
+###############################################################################
+# Concurrent global and local transactions
+###############################################################################
+
+my ($pgb_handle1, $pgb_handle2, $pgb_handle3);
+
+# global txses
+$pgb_handle1 = $master->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+
+# concurrent local
+$pgb_handle2 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank1, 'postgres' );
+$pgb_handle3 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank2, 'postgres' );
+
+$started = time();
+$selects = 0;
+$oldtotal = 0;
+while (time() - $started < $seconds)
+{
+ $total = $master->safe_psql('postgres', "select sum(amount) from accounts");
+ if ( ($total ne $oldtotal) and ($total ne '') )
+ {
+ $isolation_errors++;
+ $oldtotal = $total;
+ diag("Isolation error. Total = $total");
+ }
+ if ($total ne '') { $selects++; }
+}
+
+diag("selects = $selects");
+$master->pgbench_await($pgb_handle1);
+$shard1->pgbench_await($pgb_handle2);
+$shard2->pgbench_await($pgb_handle3);
+
+diag("completed $selects selects");
+die "" unless ( $selects > 0 &&
+ count_and_delete_rows($master, 'global_transactions') > 0 &&
+ count_and_delete_rows($shard1, 'local_transactions') > 0 &&
+ count_and_delete_rows($shard2, 'local_transactions') > 0);
+
+is($isolation_errors, 0, 'isolation between concurrent global and local transactions');
+
+
+###############################################################################
+# Snapshot stability
+###############################################################################
+
+my ($hashes, $hash1, $hash2);
+my $stability_errors = 0;
+
+# global txses
+$pgb_handle1 = $master->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+# concurrent local
+$pgb_handle2 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank1, 'postgres' );
+$pgb_handle3 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank2, 'postgres' );
+
+$selects = 0;
+$started = time();
+while (time() - $started < $seconds)
+{
+ foreach my $node ($master, $shard1, $shard2)
+ {
+ ($hash1, $_, $hash2) = split "\n", $node->safe_psql('postgres', qq[
+ begin isolation level repeatable read;
+ select md5(array_agg((t.*)::text)::text) from (select * from accounts order by id) as t;
+ select pg_sleep(3);
+ select md5(array_agg((t.*)::text)::text) from (select * from accounts order by id) as t;
+ commit;
+ ]);
+
+ if ($hash1 ne $hash2)
+ {
+ diag("oops");
+ $stability_errors++;
+ }
+ elsif ($hash1 eq '' or $hash2 eq '')
+ {
+ die;
+ }
+ else
+ {
+ $selects++;
+ }
+ }
+}
+
+$master->pgbench_await($pgb_handle1);
+$shard1->pgbench_await($pgb_handle2);
+$shard2->pgbench_await($pgb_handle3);
+
+die "" unless ( $selects > 0 &&
+ count_and_delete_rows($master, 'global_transactions') > 0 &&
+ count_and_delete_rows($shard1, 'local_transactions') > 0 &&
+ count_and_delete_rows($shard2, 'local_transactions') > 0);
+
+is($stability_errors, 0, 'snapshot is stable during concurrent global and local transactions');
+
+$master->stop;
+$shard1->stop;
+$shard2->stop;
diff --git a/contrib/postgres_fdw/t/002_bank_participant.pl b/contrib/postgres_fdw/t/002_bank_participant.pl
new file mode 100644
index 0000000000..04a2f1ba85
--- /dev/null
+++ b/contrib/postgres_fdw/t/002_bank_participant.pl
@@ -0,0 +1,240 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+use Test::More tests => 3;
+
+my $shard1 = get_new_node("shard1");
+$shard1->init;
+$shard1->append_conf('postgresql.conf', qq(
+ max_prepared_transactions = 30
+ postgres_fdw.use_global_snapshots = on
+ global_snapshot_defer_time = 15
+ track_global_snapshots = on
+ default_transaction_isolation = 'REPEATABLE READ'
+));
+$shard1->start;
+
+my $shard2 = get_new_node("shard2");
+$shard2->init;
+$shard2->append_conf('postgresql.conf', qq(
+ max_prepared_transactions = 30
+ postgres_fdw.use_global_snapshots = on
+ global_snapshot_defer_time = 15
+ track_global_snapshots = on
+ default_transaction_isolation = 'REPEATABLE READ'
+));
+$shard2->start;
+
+###############################################################################
+# Prepare nodes
+###############################################################################
+
+my @shards = ($shard1, $shard2);
+
+foreach my $node (@shards)
+{
+ $node->safe_psql('postgres', qq[
+ CREATE EXTENSION postgres_fdw;
+ CREATE TABLE accounts(id integer primary key, amount integer);
+ CREATE TABLE accounts_local() inherits(accounts);
+ CREATE TABLE global_transactions(tx_time timestamp);
+ CREATE TABLE local_transactions(tx_time timestamp);
+ ]);
+
+ foreach my $neighbor (@shards)
+ {
+ next if ($neighbor eq $node);
+
+ my $port = $neighbor->port;
+ my $host = $neighbor->host;
+
+ $node->safe_psql('postgres', qq[
+ CREATE SERVER shard_$port FOREIGN DATA WRAPPER postgres_fdw
+ options(dbname 'postgres', host '$host', port '$port');
+ CREATE FOREIGN TABLE accounts_fdw_$port() inherits (accounts)
+ server shard_$port options(table_name 'accounts_local');
+ CREATE USER MAPPING for CURRENT_USER SERVER shard_$port;
+ ]);
+ }
+}
+
+$shard1->psql('postgres', "insert into accounts_local select 2*id-1, 0 from generate_series(1, 10010) as id;");
+$shard2->psql('postgres', "insert into accounts_local select 2*id, 0 from generate_series(1, 10010) as id;");
+
+###############################################################################
+# pgbench scripts
+###############################################################################
+
+my $bank = File::Temp->new();
+append_to_file($bank, q{
+ \set id random(1, 20000)
+ BEGIN;
+ WITH upd AS (UPDATE accounts SET amount = amount - 1 WHERE id = :id RETURNING *)
+ INSERT into global_transactions SELECT now() FROM upd;
+ UPDATE accounts SET amount = amount + 1 WHERE id = (:id + 1);
+ COMMIT;
+});
+
+###############################################################################
+# Helpers
+###############################################################################
+
+sub count_and_delete_rows
+{
+ my ($node, $table) = @_;
+ my $count;
+
+ $count = $node->safe_psql('postgres',"select count(*) from $table");
+ $node->safe_psql('postgres',"delete from $table");
+ diag($node->name, ": completed $count transactions");
+ return $count;
+}
+
+###############################################################################
+# Concurrent global transactions
+###############################################################################
+
+my ($err, $rc);
+my $started;
+my $seconds = 30;
+my $selects;
+my $total = '0';
+my $oldtotal = '0';
+my $isolation_errors = 0;
+my $i;
+
+
+my ($pgb_handle1, $pgb_handle2);
+
+$pgb_handle1 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+$pgb_handle2 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+
+$started = time();
+$selects = 0;
+$i = 0;
+while (time() - $started < $seconds)
+{
+ my $shard = $shard1;
+ foreach my $shard (@shards)
+ {
+ $total = $shard->safe_psql('postgres', "select sum(amount) from accounts");
+ if ( ($total ne $oldtotal) and ($total ne '') )
+ {
+ $isolation_errors++;
+ $oldtotal = $total;
+ diag("$i: Isolation error. Total = $total");
+ }
+ if ($total ne '') { $selects++; }
+ }
+ $i++;
+}
+
+$shard1->pgbench_await($pgb_handle1);
+$shard2->pgbench_await($pgb_handle2);
+
+# sanity check
+diag("completed $selects selects");
+die "no actual transactions happend" unless ( $selects > 0 &&
+ count_and_delete_rows($shard1, 'global_transactions') > 0 &&
+ count_and_delete_rows($shard2, 'global_transactions') > 0);
+
+is($isolation_errors, 0, 'isolation between concurrent global transaction');
+
+###############################################################################
+# And do the same after soft restart
+###############################################################################
+
+$shard1->restart;
+$shard2->restart;
+$shard1->poll_query_until('postgres', "select 't'")
+ or die "Timed out waiting for shard1 to became online";
+$shard2->poll_query_until('postgres', "select 't'")
+ or die "Timed out waiting for shard2 to became online";
+
+$seconds = 15;
+$pgb_handle1 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+$pgb_handle2 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+
+$started = time();
+$selects = 0;
+$i = 0;
+
+while (time() - $started < $seconds)
+{
+ my $shard = $shard1;
+ foreach my $shard (@shards)
+ {
+ $total = $shard->safe_psql('postgres', "select sum(amount) from accounts");
+ if ( ($total ne $oldtotal) and ($total ne '') )
+ {
+ $isolation_errors++;
+ $oldtotal = $total;
+ diag("$i: Isolation error. Total = $total");
+ }
+ if ($total ne '') { $selects++; }
+ }
+ $i++;
+}
+
+$shard1->pgbench_await($pgb_handle1);
+$shard2->pgbench_await($pgb_handle2);
+
+# sanity check
+diag("completed $selects selects");
+die "no actual transactions happend" unless ( $selects > 0 &&
+ count_and_delete_rows($shard1, 'global_transactions') > 0 &&
+ count_and_delete_rows($shard2, 'global_transactions') > 0);
+
+is($isolation_errors, 0, 'isolation between concurrent global transaction after restart');
+
+###############################################################################
+# And do the same after hard restart
+###############################################################################
+
+$shard1->teardown_node;
+$shard2->teardown_node;
+$shard1->start;
+$shard2->start;
+$shard1->poll_query_until('postgres', "select 't'")
+ or die "Timed out waiting for shard1 to became online";
+$shard2->poll_query_until('postgres', "select 't'")
+ or die "Timed out waiting for shard2 to became online";
+
+
+$seconds = 15;
+$pgb_handle1 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+$pgb_handle2 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+
+$started = time();
+$selects = 0;
+$i = 0;
+
+while (time() - $started < $seconds)
+{
+ my $shard = $shard1;
+ foreach my $shard (@shards)
+ {
+ $total = $shard->safe_psql('postgres', "select sum(amount) from accounts");
+ if ( ($total ne $oldtotal) and ($total ne '') )
+ {
+ $isolation_errors++;
+ $oldtotal = $total;
+ diag("$i: Isolation error. Total = $total");
+ }
+ if ($total ne '') { $selects++; }
+ }
+ $i++;
+}
+
+$shard1->pgbench_await($pgb_handle1);
+$shard2->pgbench_await($pgb_handle2);
+
+# sanity check
+diag("completed $selects selects");
+die "no actual transactions happend" unless ( $selects > 0 &&
+ count_and_delete_rows($shard1, 'global_transactions') > 0 &&
+ count_and_delete_rows($shard2, 'global_transactions') > 0);
+
+is($isolation_errors, 0, 'isolation between concurrent global transaction after hard restart');
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index 8a2c6fc122..3ae474be28 100644
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -1840,6 +1840,41 @@ sub pg_recvlogical_upto
}
}
+sub pgbench()
+{
+ my ($self, $node, @args) = @_;
+ my $pgbench_handle = $self->pgbench_async($node, @args);
+ $self->pgbench_await($pgbench_handle);
+}
+
+sub pgbench_async()
+{
+ my ($self, @args) = @_;
+
+ my ($in, $out, $err, $rc);
+ $in = '';
+ $out = '';
+
+ my @pgbench_command = (
+ 'pgbench',
+ -h => $self->host,
+ -p => $self->port,
+ @args
+ );
+ my $handle = IPC::Run::start(\@pgbench_command, $in, $out);
+ return $handle;
+}
+
+sub pgbench_await()
+{
+ my ($self, $pgbench_handle) = @_;
+
+ # During run some pgbench threads can exit (for example due to
+ # serialization error). That will set non-zero returning code.
+ # So don't check return code here and leave it to a caller.
+ my $rc = IPC::Run::finish($pgbench_handle);
+}
+
=pod
=back
--
2.19.2
^ permalink raw reply [nested|flat] 12+ messages in thread
* Re: Global snapshots
2018-09-24 08:58 Re: Global snapshots Andrey Borodin <[email protected]>
2018-09-26 15:02 ` Re: Global snapshots Arseny Sher <[email protected]>
2018-11-29 15:21 ` Re: Global snapshots Dmitry Dolgov <[email protected]>
2018-11-30 13:00 ` Re: Global snapshots Stas Kelvich <[email protected]>
@ 2019-01-31 15:42 ` Andres Freund <[email protected]>
2019-01-31 16:46 ` Re: Global snapshots Stas Kelvich <[email protected]>
0 siblings, 1 reply; 12+ messages in thread
From: Andres Freund @ 2019-01-31 15:42 UTC (permalink / raw)
To: Stas Kelvich <[email protected]>; +Cc: Dmitry Dolgov <[email protected]>; Arseny Sher <[email protected]>; [email protected]; Robert Haas <[email protected]>; pgsql-hackers
Hi,
On 2018-11-30 16:00:17 +0300, Stas Kelvich wrote:
> > On 29 Nov 2018, at 18:21, Dmitry Dolgov <[email protected]> wrote:
> > Is there any resulting patch where the ideas how to implement this are outlined?
>
> Not yet. I’m going to continue work on this in January. And probably try to
> force some of nearby committers to make a line by line review.
This hasn't happened yet, so I think this ought to be marked ad returned
with feedback?
- Andres
^ permalink raw reply [nested|flat] 12+ messages in thread
* Re: Global snapshots
2018-09-24 08:58 Re: Global snapshots Andrey Borodin <[email protected]>
2018-09-26 15:02 ` Re: Global snapshots Arseny Sher <[email protected]>
2018-11-29 15:21 ` Re: Global snapshots Dmitry Dolgov <[email protected]>
2018-11-30 13:00 ` Re: Global snapshots Stas Kelvich <[email protected]>
2019-01-31 15:42 ` Re: Global snapshots Andres Freund <[email protected]>
@ 2019-01-31 16:46 ` Stas Kelvich <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Stas Kelvich @ 2019-01-31 16:46 UTC (permalink / raw)
To: Andres Freund <[email protected]>; +Cc: Dmitry Dolgov <[email protected]>; Arseny Sher <[email protected]>; [email protected]; Robert Haas <[email protected]>; pgsql-hackers
> On 31 Jan 2019, at 18:42, Andres Freund <[email protected]> wrote:
>
> Hi,
>
> On 2018-11-30 16:00:17 +0300, Stas Kelvich wrote:
>>> On 29 Nov 2018, at 18:21, Dmitry Dolgov <[email protected]> wrote:
>>> Is there any resulting patch where the ideas how to implement this are outlined?
>>
>> Not yet. I’m going to continue work on this in January. And probably try to
>> force some of nearby committers to make a line by line review.
>
> This hasn't happened yet, so I think this ought to be marked ad returned
> with feedback?
>
No objections. I don't think this will realistically go in during last CF, so
will open it during next release cycle.
--
Stas Kelvich
Postgres Professional: http://www.postgrespro.com
The Russian Postgres Company
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH 06/10] Make XlogReadTwoPhaseData not use callback but call the function directly
@ 2019-04-18 06:02 Kyotaro Horiguchi <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Kyotaro Horiguchi @ 2019-04-18 06:02 UTC (permalink / raw)
This patch replaces the call to the callback in XlogReadTwoPhaseData
with direct call to the original function. Then invalidate the
parameters callback and private for XLogReaderAllocate.
---
src/backend/access/transam/twophase.c | 8 ++------
src/backend/access/transam/xlogutils.c | 8 +++++---
src/backend/replication/logical/logicalfuncs.c | 8 ++++++--
src/include/access/xlogutils.h | 5 +----
4 files changed, 14 insertions(+), 15 deletions(-)
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index a3573ad0af..caaa09d785 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -1386,8 +1386,7 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
XLogReaderState *xlogreader;
char *errormsg;
- xlogreader = XLogReaderAllocate(wal_segment_size, &read_local_xlog_page,
- NULL);
+ xlogreader = XLogReaderAllocate(wal_segment_size, NULL, NULL);
if (!xlogreader)
ereport(ERROR,
(errcode(ERRCODE_OUT_OF_MEMORY),
@@ -1396,10 +1395,7 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
while (XLogReadRecord(xlogreader, lsn, &record, &errormsg) ==
XLREAD_NEED_DATA)
- xlogreader->read_page(xlogreader,
- xlogreader->loadPagePtr, xlogreader->loadLen,
- xlogreader->currRecPtr, xlogreader->readBuf,
- &xlogreader->readPageTLI);
+ read_local_xlog_page(xlogreader);
if (record == NULL)
ereport(ERROR,
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index c853e1f0e3..fd461f16fc 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -908,10 +908,12 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wa
* loop for now.
*/
void
-read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
- int reqLen, XLogRecPtr targetRecPtr, char *cur_page,
- TimeLineID *pageTLI)
+read_local_xlog_page(XLogReaderState *state)
{
+ XLogRecPtr targetPagePtr = state->loadPagePtr;
+ int reqLen = state->loadLen;
+ char *cur_page = state->readBuf;
+ TimeLineID *pageTLI = &state->readPageTLI;
XLogRecPtr read_upto,
loc;
int count;
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 4d09255504..240a375d8f 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -118,8 +118,12 @@ void
logical_read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
int reqLen, XLogRecPtr targetRecPtr, char *cur_page, TimeLineID *pageTLI)
{
- read_local_xlog_page(state, targetPagePtr, reqLen,
- targetRecPtr, cur_page, pageTLI);
+ Assert(targetPagePtr == state->loadPagePtr &&
+ reqLen == state->loadLen &&
+ targetRecPtr == state->currRecPtr &&
+ cur_page == state->readBuf &&
+ pageTLI == &state->readPageTLI);
+ read_local_xlog_page(state);
}
/*
diff --git a/src/include/access/xlogutils.h b/src/include/access/xlogutils.h
index 5dba86b8b8..7f119837ce 100644
--- a/src/include/access/xlogutils.h
+++ b/src/include/access/xlogutils.h
@@ -47,10 +47,7 @@ extern Buffer XLogReadBufferExtended(RelFileNode rnode, ForkNumber forknum,
extern Relation CreateFakeRelcacheEntry(RelFileNode rnode);
extern void FreeFakeRelcacheEntry(Relation fakerel);
-extern void read_local_xlog_page(XLogReaderState *state,
- XLogRecPtr targetPagePtr, int reqLen,
- XLogRecPtr targetRecPtr, char *cur_page,
- TimeLineID *pageTLI);
+extern void read_local_xlog_page(XLogReaderState *state);
extern void XLogReadDetermineTimeline(XLogReaderState *state,
XLogRecPtr wantPage, uint32 wantLength);
--
2.16.3
----Next_Part(Thu_Apr_18_21_02_57_2019_406)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="0007-Make-logical-rep-stuff-not-use-callback-but-call-the.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH 06/10] Make XlogReadTwoPhaseData not use callback but call the function directly
@ 2019-04-18 06:02 Kyotaro Horiguchi <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Kyotaro Horiguchi @ 2019-04-18 06:02 UTC (permalink / raw)
This patch replaces the call to the callback in XlogReadTwoPhaseData
with direct call to the original function. Then invalidate the
parameters callback and private for XLogReaderAllocate.
---
src/backend/access/transam/twophase.c | 8 ++------
src/backend/access/transam/xlogutils.c | 8 +++++---
src/backend/replication/logical/logicalfuncs.c | 8 ++++++--
src/include/access/xlogutils.h | 5 +----
4 files changed, 14 insertions(+), 15 deletions(-)
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 6feca69126..f6d48368fe 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -1386,8 +1386,7 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
XLogReaderState *xlogreader;
char *errormsg;
- xlogreader = XLogReaderAllocate(wal_segment_size, &read_local_xlog_page,
- NULL);
+ xlogreader = XLogReaderAllocate(wal_segment_size, NULL, NULL);
if (!xlogreader)
ereport(ERROR,
(errcode(ERRCODE_OUT_OF_MEMORY),
@@ -1396,10 +1395,7 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
while (XLogReadRecord(xlogreader, lsn, &record, &errormsg) ==
XLREAD_NEED_DATA)
- xlogreader->read_page(xlogreader,
- xlogreader->loadPagePtr, xlogreader->loadLen,
- xlogreader->currRecPtr, xlogreader->readBuf,
- &xlogreader->readPageTLI);
+ read_local_xlog_page(xlogreader);
if (record == NULL)
ereport(ERROR,
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index c853e1f0e3..fd461f16fc 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -908,10 +908,12 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wa
* loop for now.
*/
void
-read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
- int reqLen, XLogRecPtr targetRecPtr, char *cur_page,
- TimeLineID *pageTLI)
+read_local_xlog_page(XLogReaderState *state)
{
+ XLogRecPtr targetPagePtr = state->loadPagePtr;
+ int reqLen = state->loadLen;
+ char *cur_page = state->readBuf;
+ TimeLineID *pageTLI = &state->readPageTLI;
XLogRecPtr read_upto,
loc;
int count;
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 4d09255504..240a375d8f 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -118,8 +118,12 @@ void
logical_read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
int reqLen, XLogRecPtr targetRecPtr, char *cur_page, TimeLineID *pageTLI)
{
- read_local_xlog_page(state, targetPagePtr, reqLen,
- targetRecPtr, cur_page, pageTLI);
+ Assert(targetPagePtr == state->loadPagePtr &&
+ reqLen == state->loadLen &&
+ targetRecPtr == state->currRecPtr &&
+ cur_page == state->readBuf &&
+ pageTLI == &state->readPageTLI);
+ read_local_xlog_page(state);
}
/*
diff --git a/src/include/access/xlogutils.h b/src/include/access/xlogutils.h
index be0c79d18c..9724ce20b8 100644
--- a/src/include/access/xlogutils.h
+++ b/src/include/access/xlogutils.h
@@ -47,10 +47,7 @@ extern Buffer XLogReadBufferExtended(RelFileNode rnode, ForkNumber forknum,
extern Relation CreateFakeRelcacheEntry(RelFileNode rnode);
extern void FreeFakeRelcacheEntry(Relation fakerel);
-extern void read_local_xlog_page(XLogReaderState *state,
- XLogRecPtr targetPagePtr, int reqLen,
- XLogRecPtr targetRecPtr, char *cur_page,
- TimeLineID *pageTLI);
+extern void read_local_xlog_page(XLogReaderState *state);
extern void XLogReadDetermineTimeline(XLogReaderState *state,
XLogRecPtr wantPage, uint32 wantLength);
--
2.16.3
----Next_Part(Wed_Jul_10_13_18_10_2019_842)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="v4-0007-Make-logical-rep-stuff-not-use-callback-but-call-the.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH 06/10] Make XlogReadTwoPhaseData not use callback but call the function directly
@ 2019-04-18 06:02 Kyotaro Horiguchi <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Kyotaro Horiguchi @ 2019-04-18 06:02 UTC (permalink / raw)
This patch replaces the call to the callback in XlogReadTwoPhaseData
with direct call to the original function. Then invalidate the
parameters callback and private for XLogReaderAllocate.
---
src/backend/access/transam/twophase.c | 8 ++------
src/backend/access/transam/xlogutils.c | 8 +++++---
src/backend/replication/logical/logicalfuncs.c | 8 ++++++--
src/include/access/xlogutils.h | 5 +----
4 files changed, 14 insertions(+), 15 deletions(-)
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index ad0d505e54..23424886dc 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -1386,8 +1386,7 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
XLogReaderState *xlogreader;
char *errormsg;
- xlogreader = XLogReaderAllocate(wal_segment_size, &read_local_xlog_page,
- NULL);
+ xlogreader = XLogReaderAllocate(wal_segment_size, NULL, NULL);
if (!xlogreader)
ereport(ERROR,
(errcode(ERRCODE_OUT_OF_MEMORY),
@@ -1396,10 +1395,7 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
while (XLogReadRecord(xlogreader, lsn, &record, &errormsg) ==
XLREAD_NEED_DATA)
- xlogreader->read_page(xlogreader,
- xlogreader->loadPagePtr, xlogreader->loadLen,
- xlogreader->currRecPtr, xlogreader->readBuf,
- &xlogreader->readPageTLI);
+ read_local_xlog_page(xlogreader);
if (record == NULL)
ereport(ERROR,
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index c853e1f0e3..fd461f16fc 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -908,10 +908,12 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wa
* loop for now.
*/
void
-read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
- int reqLen, XLogRecPtr targetRecPtr, char *cur_page,
- TimeLineID *pageTLI)
+read_local_xlog_page(XLogReaderState *state)
{
+ XLogRecPtr targetPagePtr = state->loadPagePtr;
+ int reqLen = state->loadLen;
+ char *cur_page = state->readBuf;
+ TimeLineID *pageTLI = &state->readPageTLI;
XLogRecPtr read_upto,
loc;
int count;
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 4d09255504..240a375d8f 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -118,8 +118,12 @@ void
logical_read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
int reqLen, XLogRecPtr targetRecPtr, char *cur_page, TimeLineID *pageTLI)
{
- read_local_xlog_page(state, targetPagePtr, reqLen,
- targetRecPtr, cur_page, pageTLI);
+ Assert(targetPagePtr == state->loadPagePtr &&
+ reqLen == state->loadLen &&
+ targetRecPtr == state->currRecPtr &&
+ cur_page == state->readBuf &&
+ pageTLI == &state->readPageTLI);
+ read_local_xlog_page(state);
}
/*
diff --git a/src/include/access/xlogutils.h b/src/include/access/xlogutils.h
index be0c79d18c..9724ce20b8 100644
--- a/src/include/access/xlogutils.h
+++ b/src/include/access/xlogutils.h
@@ -47,10 +47,7 @@ extern Buffer XLogReadBufferExtended(RelFileNode rnode, ForkNumber forknum,
extern Relation CreateFakeRelcacheEntry(RelFileNode rnode);
extern void FreeFakeRelcacheEntry(Relation fakerel);
-extern void read_local_xlog_page(XLogReaderState *state,
- XLogRecPtr targetPagePtr, int reqLen,
- XLogRecPtr targetRecPtr, char *cur_page,
- TimeLineID *pageTLI);
+extern void read_local_xlog_page(XLogReaderState *state);
extern void XLogReadDetermineTimeline(XLogReaderState *state,
XLogRecPtr wantPage, uint32 wantLength);
--
2.16.3
----Next_Part(Fri_May_24_11_56_24_2019_374)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="v3-0007-Make-logical-rep-stuff-not-use-callback-but-call-the.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* [PATCH 06/10] Make XlogReadTwoPhaseData not use callback but call the function directly
@ 2019-04-18 06:02 Kyotaro Horiguchi <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Kyotaro Horiguchi @ 2019-04-18 06:02 UTC (permalink / raw)
This patch replaces the call to the callback in XlogReadTwoPhaseData
with direct call to the original function. Then invalidate the
parameters callback and private for XLogReaderAllocate.
---
src/backend/access/transam/twophase.c | 8 ++------
src/backend/access/transam/xlogutils.c | 8 +++++---
src/backend/replication/logical/logicalfuncs.c | 8 ++++++--
src/include/access/xlogutils.h | 5 +----
4 files changed, 14 insertions(+), 15 deletions(-)
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index a3573ad0af..caaa09d785 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -1386,8 +1386,7 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
XLogReaderState *xlogreader;
char *errormsg;
- xlogreader = XLogReaderAllocate(wal_segment_size, &read_local_xlog_page,
- NULL);
+ xlogreader = XLogReaderAllocate(wal_segment_size, NULL, NULL);
if (!xlogreader)
ereport(ERROR,
(errcode(ERRCODE_OUT_OF_MEMORY),
@@ -1396,10 +1395,7 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
while (XLogReadRecord(xlogreader, lsn, &record, &errormsg) ==
XLREAD_NEED_DATA)
- xlogreader->read_page(xlogreader,
- xlogreader->loadPagePtr, xlogreader->loadLen,
- xlogreader->currRecPtr, xlogreader->readBuf,
- &xlogreader->readPageTLI);
+ read_local_xlog_page(xlogreader);
if (record == NULL)
ereport(ERROR,
diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index c853e1f0e3..fd461f16fc 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -908,10 +908,12 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wa
* loop for now.
*/
void
-read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
- int reqLen, XLogRecPtr targetRecPtr, char *cur_page,
- TimeLineID *pageTLI)
+read_local_xlog_page(XLogReaderState *state)
{
+ XLogRecPtr targetPagePtr = state->loadPagePtr;
+ int reqLen = state->loadLen;
+ char *cur_page = state->readBuf;
+ TimeLineID *pageTLI = &state->readPageTLI;
XLogRecPtr read_upto,
loc;
int count;
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 4d09255504..240a375d8f 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -118,8 +118,12 @@ void
logical_read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
int reqLen, XLogRecPtr targetRecPtr, char *cur_page, TimeLineID *pageTLI)
{
- read_local_xlog_page(state, targetPagePtr, reqLen,
- targetRecPtr, cur_page, pageTLI);
+ Assert(targetPagePtr == state->loadPagePtr &&
+ reqLen == state->loadLen &&
+ targetRecPtr == state->currRecPtr &&
+ cur_page == state->readBuf &&
+ pageTLI == &state->readPageTLI);
+ read_local_xlog_page(state);
}
/*
diff --git a/src/include/access/xlogutils.h b/src/include/access/xlogutils.h
index 5dba86b8b8..7f119837ce 100644
--- a/src/include/access/xlogutils.h
+++ b/src/include/access/xlogutils.h
@@ -47,10 +47,7 @@ extern Buffer XLogReadBufferExtended(RelFileNode rnode, ForkNumber forknum,
extern Relation CreateFakeRelcacheEntry(RelFileNode rnode);
extern void FreeFakeRelcacheEntry(Relation fakerel);
-extern void read_local_xlog_page(XLogReaderState *state,
- XLogRecPtr targetPagePtr, int reqLen,
- XLogRecPtr targetRecPtr, char *cur_page,
- TimeLineID *pageTLI);
+extern void read_local_xlog_page(XLogReaderState *state);
extern void XLogReadDetermineTimeline(XLogReaderState *state,
XLogRecPtr wantPage, uint32 wantLength);
--
2.16.3
----Next_Part(Fri_Apr_26_17_40_34_2019_888)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="v2-0007-Make-logical-rep-stuff-not-use-callback-but-call-the.patch"
^ permalink raw reply [nested|flat] 12+ messages in thread
* Re: psql memory leaks
@ 2023-02-21 19:03 Corey Huinker <[email protected]>
2023-02-22 05:32 ` Re: psql memory leaks Michael Paquier <[email protected]>
0 siblings, 1 reply; 12+ messages in thread
From: Corey Huinker @ 2023-02-21 19:03 UTC (permalink / raw)
To: Kyotaro Horiguchi <[email protected]>; +Cc: [email protected]
On Mon, Feb 20, 2023 at 9:56 PM Kyotaro Horiguchi <[email protected]>
wrote:
> I noticed that \bind is leaking memory for each option.
>
> =# SELECT $1, $2, $3 \ bind 1 2 3 \g
>
> The leaked memory blocks are comming from
> psql_scan_slash_option(). The attached small patch resolves that
> issue. I looked through the function's call sites, but I didn't find
> the same mistake.
>
> regards.
>
>
Good catch. Patch passes make check-world.
^ permalink raw reply [nested|flat] 12+ messages in thread
* Re: psql memory leaks
2023-02-21 19:03 Re: psql memory leaks Corey Huinker <[email protected]>
@ 2023-02-22 05:32 ` Michael Paquier <[email protected]>
0 siblings, 0 replies; 12+ messages in thread
From: Michael Paquier @ 2023-02-22 05:32 UTC (permalink / raw)
To: Corey Huinker <[email protected]>; +Cc: Kyotaro Horiguchi <[email protected]>; [email protected]
On Tue, Feb 21, 2023 at 02:03:43PM -0500, Corey Huinker wrote:
> Good catch. Patch passes make check-world.
Indeed. I was reviewing the whole and there could be a point in
resetting bind_nparams at the end of SendQuery() to keep a correct
track of what's saved in the pset data for the bind parameters, but
not doing so is not a big deal either because we'd just reset it once
the full allocation of the parameters is done and bind_flag is all
about that. The code paths leading to the free seem correct, seen
from here.
--
Michael
Attachments:
[application/pgp-signature] signature.asc (833B, ../../Y%[email protected]/2-signature.asc)
download
^ permalink raw reply [nested|flat] 12+ messages in thread
end of thread, other threads:[~2023-02-22 05:32 UTC | newest]
Thread overview: 12+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2018-09-24 08:58 Re: Global snapshots Andrey Borodin <[email protected]>
2018-09-26 15:02 ` Arseny Sher <[email protected]>
2018-11-29 15:21 ` Dmitry Dolgov <[email protected]>
2018-11-30 13:00 ` Stas Kelvich <[email protected]>
2019-01-31 15:42 ` Andres Freund <[email protected]>
2019-01-31 16:46 ` Stas Kelvich <[email protected]>
2019-04-18 06:02 [PATCH 06/10] Make XlogReadTwoPhaseData not use callback but call the function directly Kyotaro Horiguchi <[email protected]>
2019-04-18 06:02 [PATCH 06/10] Make XlogReadTwoPhaseData not use callback but call the function directly Kyotaro Horiguchi <[email protected]>
2019-04-18 06:02 [PATCH 06/10] Make XlogReadTwoPhaseData not use callback but call the function directly Kyotaro Horiguchi <[email protected]>
2019-04-18 06:02 [PATCH 06/10] Make XlogReadTwoPhaseData not use callback but call the function directly Kyotaro Horiguchi <[email protected]>
2023-02-21 19:03 Re: psql memory leaks Corey Huinker <[email protected]>
2023-02-22 05:32 ` Re: psql memory leaks Michael Paquier <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox