*
* lockfuncs.c
* Functions for SQL access to various lock-manager capabilities.
*
* Copyright (c) 2002-2012, PostgreSQL Global Development Group
*
* IDENTIFICATION
* src/backend/utils/adt/lockfuncs.c
*
* -------------------------------------------------------------------------
*/
#include "postgres.h"
#include "knl/knl_variable.h"
#include "access/transam.h"
#include "access/heapam.h"
#include "access/tableam.h"
#include "catalog/pg_type.h"
#include "catalog/pg_database.h"
#include "catalog/indexing.h"
#include "commands/dbcommands.h"
#include "funcapi.h"
#include "miscadmin.h"
#ifdef PGXC
#include "pgxc/groupmgr.h"
#include "pgxc/pgxc.h"
#include "pgxc/pgxcnode.h"
#include "pgxc/nodemgr.h"
#include "pgxc/poolutils.h"
#include "executor/spi.h"
#include "tcop/utility.h"
#endif
#include "storage/predicate_internals.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "access/heapam.h"
#include "utils/syscache.h"
#include "utils/snapmgr.h"
#include "pgstat.h"
#ifdef DOLPHIN
#include "plugin_postgres.h"
#include "libpq/md5.h"
#include "utils/knl_relcache.h"
#include "plugin_utils/show_common.h"
* user advisory lock like b_database.
* different from openGauss:
* 1. lock using string, not int.
* 2. lokc allow timeout
* using md5sum parse string as int8*16, then cast to int32*4,using locktag_field2-5
*/
* get_lock get advisory lock using string.
* arg1: string length < 64
* arg2: int timeout ,seconds,default 0 ,waiting forever
* return: 1 if success, 0 if timeout, null if other error
*/
#define SET_LOCKTAG_ADVISORY_5(locktag, id1, id2, id3, id4, id5) \
((locktag).locktag_field1 = (id1), (locktag).locktag_field2 = (id2), (locktag).locktag_field3 = (id3), \
(locktag).locktag_field4 = (id4), (locktag).locktag_field5 = (id5), (locktag).locktag_type = LOCKTAG_ADVISORY, \
(locktag).locktag_lockmethodid = USER_LOCKMETHOD)
#define MAX_ADVISORY_LOCK_NAME_LENTH 64
#define DATABASE_ID_LENTH 32
struct HTAB* lockNameHash = NULL;
pthread_mutex_t gNameHashLock;
typedef struct LockName {
char name[MAX_ADVISORY_LOCK_NAME_LENTH+DATABASE_ID_LENTH+1];
int64 nlocks;
} LockName;
PG_FUNCTION_INFO_V1_PUBLIC(GetAdvisoryLockWithtimeTextFormat);
extern "C" DLL_PUBLIC Datum GetAdvisoryLockWithtimeTextFormat(PG_FUNCTION_ARGS);
PG_FUNCTION_INFO_V1_PUBLIC(GetAdvisoryLockWithtime);
extern "C" DLL_PUBLIC Datum GetAdvisoryLockWithtime(PG_FUNCTION_ARGS);
PG_FUNCTION_INFO_V1_PUBLIC(GetAdvisoryLock);
extern "C" DLL_PUBLIC Datum GetAdvisoryLock(PG_FUNCTION_ARGS);
PG_FUNCTION_INFO_V1_PUBLIC(ReleaseAdvisoryLock);
extern "C" DLL_PUBLIC Datum ReleaseAdvisoryLock(PG_FUNCTION_ARGS);
PG_FUNCTION_INFO_V1_PUBLIC(IsFreeAdvisoryLock);
extern "C" DLL_PUBLIC Datum IsFreeAdvisoryLock(PG_FUNCTION_ARGS);
PG_FUNCTION_INFO_V1_PUBLIC(IsUsedAdvisoryLock);
extern "C" DLL_PUBLIC Datum IsUsedAdvisoryLock(PG_FUNCTION_ARGS);
PG_FUNCTION_INFO_V1_PUBLIC(ReleaseAllAdvisoryLock);
extern "C" DLL_PUBLIC Datum ReleaseAllAdvisoryLock(PG_FUNCTION_ARGS);
PG_FUNCTION_INFO_V1_PUBLIC(GetAllAdvisoryLock);
extern "C" DLL_PUBLIC Datum GetAllAdvisoryLock(PG_FUNCTION_ARGS);
PG_FUNCTION_INFO_V1_PUBLIC(ClearInvalidLockName);
extern "C" DLL_PUBLIC Datum ClearInvalidLockName(PG_FUNCTION_ARGS);
void InitLockNameHash()
{
HASHCTL info = {0};
info.keysize = MAX_ADVISORY_LOCK_NAME_LENTH+DATABASE_ID_LENTH+1;
info.entrysize = sizeof(LockName);
info.hash = string_hash;
info.hcxt = g_instance.builtin_proc_context;
lockNameHash = hash_create("Lock Name Hash", 100, &info, HASH_ELEM | HASH_FUNCTION | HASH_CONTEXT);
}
uint32 *AdvisoryLockGetIdFromName(char *name, uint32 *lockid)
{
uint8 sum[16];
if (strlen(name) > MAX_ADVISORY_LOCK_NAME_LENTH) {
ereport(ERROR, (errcode(ERRCODE_NAME_TOO_LONG), errmsg("max length of advisory lock is 64")));
}
if (strlen(name) == 0) {
ereport(ERROR, (errcode(ERRCODE_INVALID_NAME), errmsg("Incorrect user-level lock name '%s'", name)));
}
pg_md5_binary(name, strlen(name), sum);
for (int i = 0; i < 4; i++) {
int index = i * 4;
for (int j = 0; j < 4; j++) {
uint32 t = sum[index + j];
lockid[i] = lockid[i] << 8;
lockid[i] |= t;
}
}
return lockid;
}
bool IsFreeAdvisoryLockInner(char *lockname)
{
LOCKTAG tag;
uint32 lockid[4] = {0};
AdvisoryLockGetIdFromName(lockname, lockid);
SET_LOCKTAG_ADVISORY_5(tag, u_sess->proc_cxt.MyDatabaseId, lockid[0], lockid[1], lockid[2], lockid[3]);
uint32 hashcode = LockTagHashCode(&tag);
LWLock *partitionLock = NULL;
partitionLock = LockHashPartitionLock(hashcode);
LWLockAcquire(partitionLock, LW_SHARED);
bool foundCachedEntry = false;
hash_search_with_hash_value(t_thrd.storage_cxt.LockMethodLockHash, &tag, hashcode, HASH_FIND, &foundCachedEntry);
LWLockRelease(partitionLock);
return !foundCachedEntry;
}
Datum ClearInvalidLockNameInner()
{
HASH_SEQ_STATUS status;
LockName *lockName = NULL;
int64 sumLocks = 0;
AutoMutexLock nameHashLock(&gNameHashLock);
nameHashLock.lock();
hash_seq_init(&status, lockNameHash);
char dataBaseId[DATABASE_ID_LENTH+1] = {0};
errno_t rc = snprintf_s((char*)dataBaseId, DATABASE_ID_LENTH+1, DATABASE_ID_LENTH, "%u", u_sess->proc_cxt.MyDatabaseId);
securec_check_ss(rc, "\0", "\0");
while ((lockName = (LockName *)hash_seq_search(&status)) != NULL) {
if (pg_strncasecmp(dataBaseId, lockName->name, strlen((char*)dataBaseId)) != 0)
continue;
if (IsFreeAdvisoryLockInner((char*)(lockName->name)+strlen((char*)dataBaseId))) {
sumLocks += lockName->nlocks;
lockName->nlocks = 0;
hash_search(lockNameHash, (void *)&(lockName->name), HASH_REMOVE, NULL);
}
}
nameHashLock.unLock();
PG_RETURN_INT64(sumLocks);
}
void SetLockNameHash(const char* name, bool nlockInitialFlag = false, bool releaseFlag = false)
{
char tempName[MAX_ADVISORY_LOCK_NAME_LENTH+1+DATABASE_ID_LENTH] = {0};
errno_t rc = snprintf_s(tempName, sizeof(tempName), sizeof(tempName) - 1, "%u%s", u_sess->proc_cxt.MyDatabaseId, name);
securec_check_ss(rc, "\0", "\0");
LockName *result = NULL;
bool found = false;
Assert(name != NULL);
AutoMutexLock nameHashLock(&gNameHashLock);
nameHashLock.lock();
result = (LockName *)hash_search(lockNameHash, (void *)&tempName, HASH_ENTER, &found);
if (releaseFlag && found) {
result->nlocks--;
if (result->nlocks == 0)
hash_search(lockNameHash, (void *)&(result->name), HASH_REMOVE, NULL);
} else if (!releaseFlag) {
auto oldcxt = MemoryContextSwitchTo(u_sess->self_mem_cxt);
GetSessionContext()->lockNameList = lappend(GetSessionContext()->lockNameList, result);
MemoryContextSwitchTo(oldcxt);
if (found && !nlockInitialFlag)
result->nlocks++;
else
result->nlocks = 1;
}
nameHashLock.unLock();
}
Datum IsUsedAdvisoryLockInner(char *lockname)
{
LOCKTAG tag;
uint32 lockid[4] = {0};
if (g_instance.attr.attr_common.enable_thread_pool) {
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("Unsupport using advisory lock in threadpool")));
}
AdvisoryLockGetIdFromName(lockname, lockid);
SET_LOCKTAG_ADVISORY_5(tag, u_sess->proc_cxt.MyDatabaseId, lockid[0], lockid[1], lockid[2], lockid[3]);
uint32 hashcode = LockTagHashCode(&tag);
LWLock *partitionLock = NULL;
partitionLock = LockHashPartitionLock(hashcode);
LWLockAcquire(partitionLock, LW_SHARED);
bool foundCachedEntry = false;
LOCK *lock =
(LOCK *)hash_search_with_hash_value(t_thrd.storage_cxt.LockMethodLockHash, &tag, hashcode, HASH_FIND, &foundCachedEntry);
if (!foundCachedEntry) {
LWLockRelease(partitionLock);
return (Datum)0;
}
PROCLOCK *proclock;
SHM_QUEUE *procLocks;
PGPROC *proc;
uint64 sessionId = 0;
procLocks = &(lock->procLocks);
proclock = (PROCLOCK *)SHMQueueNext(procLocks, procLocks, offsetof(PROCLOCK, lockLink));
while (proclock != NULL) {
proc = proclock->tag.myProc;
sessionId = proc->sessionid;
proclock = (PROCLOCK *)SHMQueueNext(procLocks, &proclock->lockLink, offsetof(PROCLOCK, lockLink));
break;
}
LWLockRelease(partitionLock);
if (sessionId == 0)
return (Datum)0;
return UInt64GetDatum(sessionId);
}
int GetAdvisoryLockWithTimeInner(char *lockname, double timeout)
{
TimestampTz startTime;
LOCKTAG tag;
uint32 lockid[4] = {0};
uint64 currentSessionId = IS_THREAD_POOL_WORKER ? u_sess->session_id : t_thrd.proc_cxt.MyProcPid;
if (g_instance.attr.attr_common.enable_thread_pool) {
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("Unsupport using advisory lock in threadpool")));
}
AdvisoryLockGetIdFromName(lockname, lockid);
SET_LOCKTAG_ADVISORY_5(tag, u_sess->proc_cxt.MyDatabaseId, lockid[0], lockid[1], lockid[2], lockid[3]);
if (t_thrd.shemem_ptr_cxt.mySessionTimeEntry) {
startTime = GetCurrentTimestamp();
} else {
ereport(WARNING, (errmsg("TimeEntry not inited")));
return -1;
}
ClearInvalidLockNameInner();
bool nlockInitialFlag = (DatumGetUInt64(IsUsedAdvisoryLockInner(lockname)) != currentSessionId);
if (timeout <= 0) {
LockAcquireResult res = LockAcquire(&tag, ExclusiveLock, true, false, false);
if (res == LOCKACQUIRE_NOT_AVAIL) {
return 0;
} else {
SetLockNameHash(lockname, nlockInitialFlag);
return 1;
}
} else {
do {
LockAcquireResult res = LockAcquire(&tag, ExclusiveLock, true, true, false);
if (res == LOCKACQUIRE_NOT_AVAIL) {
pg_usleep(USECS_PER_MSEC);
ProcessInterrupts();
} else {
SetLockNameHash(lockname, nlockInitialFlag);
return 1;
}
} while (ComputeTimeStamp(startTime) / MSECS_PER_SEC < timeout);
}
return 0;
}
int64 LockNameHashRelease()
{
LockName *lockName = NULL;
int64 sumLocks = 0;
ListCell* lc = NULL;
AutoMutexLock nameHashLock(&gNameHashLock);
nameHashLock.lock();
foreach (lc, GetSessionContext()->lockNameList) {
lockName = (LockName*)lfirst(lc);
sumLocks += lockName->nlocks;
lockName->nlocks = 0;
hash_search(lockNameHash, (void *)&(lockName->name), HASH_REMOVE, NULL);
}
nameHashLock.unLock();
return sumLocks;
}
#endif
#define NUM_LOCKTAG_ID 17
const char* const LockTagTypeNames[] = {"relation",
"extend",
"partition",
"partition_seq",
"page",
"tuple",
"transactionid",
"virtualxid",
"object",
"cstore_freespace",
"userlock",
"advisory",
"filenode",
"subtransactionid",
"tuple_uid",
"tablespace",
"plpytyhon_gil"};
static const char* const PredicateLockTagTypeNames[] = {"relation", "page", "tuple"};
typedef struct {
LockData* lockData;
int currIdx;
PredicateLockData* predLockData;
int predLockIdx;
} PG_Lock_Status;
#ifdef PGXC
* These enums are defined to make calls to pgxc_advisory_lock more readable.
*/
typedef enum { SESSION_LOCK, TRANSACTION_LOCK } LockLevel;
typedef enum { WAIT, DONT_WAIT } TryType;
static bool pgxc_advisory_lock(int64 key64, int32 key1, int32 key2, bool iskeybig, LOCKMODE lockmode,
LockLevel locklevel, TryType locktry, Name databaseName = NULL);
#endif
#define NUM_LOCK_STATUS_COLUMNS 20
* VXIDGetDatum - Construct a text representation of a VXID
*
* This is currently only used in pg_lock_status, so we put it here.
*/
static Datum VXIDGetDatum(BackendId bid, LocalTransactionId lxid)
{
* The representation is "<bid>/<lxid>", decimal and unsigned decimal
* respectively. Note that elog.c also knows how to format a vxid.
*/
char vxidstr[64];
errno_t ss_rc = snprintf_s(vxidstr, sizeof(vxidstr), sizeof(vxidstr) - 1, "%d/" XID_FMT, bid, lxid);
securec_check_ss(ss_rc, "\0", "\0");
return CStringGetTextDatum(vxidstr);
}
char* LocktagToString(const LOCKTAG locktag)
{
StringInfoData tag;
initStringInfo(&tag);
appendStringInfo(&tag, "%x:%x:%x:%x:%x:%x", locktag.locktag_field1, locktag.locktag_field2,
locktag.locktag_field3, locktag.locktag_field4, locktag.locktag_field5, locktag.locktag_type);
return tag.data;
}
static void GetLocktagInfo(const LockInstanceData* instance, Datum values[])
{
char* blocklocktag = LocktagToString(instance->locktag);
values[NUM_LOCKTAG_ID] = CStringGetTextDatum(blocklocktag);
pfree_ext(blocklocktag);
}
static char* LockPredTagToString(const PREDICATELOCKTARGETTAG* predTag)
{
LOCKTAG tag;
Oid dbId = GET_PREDICATELOCKTARGETTAG_DB(*predTag);
Oid relId = GET_PREDICATELOCKTARGETTAG_RELATION(*predTag);
switch (GET_PREDICATELOCKTARGETTAG_TYPE(*predTag)) {
case PREDLOCKTAG_PAGE:
{
BlockNumber pageId = GET_PREDICATELOCKTARGETTAG_PAGE(*predTag);
SET_LOCKTAG_PAGE(tag, dbId, relId, 0, pageId);
}
break;
case PREDLOCKTAG_TUPLE:
{
BlockNumber pageId = GET_PREDICATELOCKTARGETTAG_PAGE(*predTag);
OffsetNumber itemId = GET_PREDICATELOCKTARGETTAG_OFFSET(*predTag);
SET_LOCKTAG_TUPLE(tag, dbId, relId, 0, pageId, itemId);
}
break;
case PREDLOCKTAG_RELATION:
SET_LOCKTAG_RELATION(tag, dbId, relId);
break;
default:
ereport(ERROR,
(errcode(ERRCODE_INTERNAL_ERROR), errmsg("invalid predlock locktype")));
break;
}
return LocktagToString(tag);
}
* pg_lock_status - produce a view with one row per held or awaited lock mode
*/
Datum pg_lock_status(PG_FUNCTION_ARGS)
{
FuncCallContext* funcctx = NULL;
PG_Lock_Status* mystatus = NULL;
LockData* lockData = NULL;
PredicateLockData* predLockData = NULL;
if (SRF_IS_FIRSTCALL()) {
TupleDesc tupdesc;
MemoryContext oldcontext;
funcctx = SRF_FIRSTCALL_INIT();
* switch to memory context appropriate for multiple function calls
*/
oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
tupdesc = CreateTemplateTupleDesc(NUM_LOCK_STATUS_COLUMNS, false);
TupleDescInitEntry(tupdesc, (AttrNumber)1, "locktype", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)2, "database", OIDOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)3, "relation", OIDOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)4, "page", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)5, "tuple", INT2OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)6, "bucket", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)7, "virtualxid", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)8, "transactionid", XIDOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)9, "classid", OIDOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)10, "objid", OIDOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)11, "objsubid", INT2OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)12, "virtualtransaction", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)13, "pid", INT8OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)14, "sessionid", INT8OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)15, "mode", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)16, "granted", BOOLOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)17, "fastpath", BOOLOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)18, "locktag", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)19, "global_sessionid", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)20, "waitstart", TIMESTAMPTZOID, -1, 0);
funcctx->tuple_desc = BlessTupleDesc(tupdesc);
* Collect all the locking information that we will format and send
* out as a result set.
*/
mystatus = (PG_Lock_Status*)palloc(sizeof(PG_Lock_Status));
funcctx->user_fctx = (void*)mystatus;
mystatus->lockData = GetLockStatusData();
mystatus->currIdx = 0;
mystatus->predLockData = GetPredicateLockStatusData();
mystatus->predLockIdx = 0;
MemoryContextSwitchTo(oldcontext);
}
funcctx = SRF_PERCALL_SETUP();
mystatus = (PG_Lock_Status*)funcctx->user_fctx;
lockData = mystatus->lockData;
while (mystatus->currIdx < lockData->nelements) {
bool granted = false;
LOCKMODE mode = 0;
const char* locktypename = NULL;
char tnbuf[32];
Datum values[NUM_LOCK_STATUS_COLUMNS];
bool nulls[NUM_LOCK_STATUS_COLUMNS];
HeapTuple tuple;
Datum result;
LockInstanceData* instance = NULL;
instance = &(lockData->locks[mystatus->currIdx]);
* Look to see if there are any held lock modes in this PROCLOCK. If
* so, report, and destructively modify lockData so we don't report
* again.
*/
granted = false;
if (instance->holdMask) {
for (mode = 0; mode < MAX_LOCKMODES; mode++) {
if (instance->holdMask & LOCKBIT_ON(mode)) {
granted = true;
instance->holdMask &= LOCKBIT_OFF(mode);
break;
}
}
}
* If no (more) held modes to report, see if PROC is waiting for a
* lock on this lock.
*/
if (!granted) {
if (instance->waitLockMode != NoLock) {
mode = instance->waitLockMode;
* We are now done with this PROCLOCK, so advance pointer to
* continue with next one on next call.
*/
mystatus->currIdx++;
} else {
* Okay, we've displayed all the locks associated with this
* PROCLOCK, proceed to the next one.
*/
mystatus->currIdx++;
continue;
}
}
* Form tuple with appropriate data.
*/
errno_t rc = memset_s(values, sizeof(values), 0, sizeof(values));
securec_check(rc, "\0", "\0");
rc = memset_s(nulls, sizeof(nulls), false, sizeof(nulls));
securec_check(rc, "\0", "\0");
if (instance->locktag.locktag_type <= LOCKTAG_LAST_TYPE)
locktypename = LockTagTypeNames[instance->locktag.locktag_type];
else {
errno_t ss_rc =
snprintf_s(tnbuf, sizeof(tnbuf), sizeof(tnbuf) - 1, "unknown %d", (int)instance->locktag.locktag_type);
securec_check_ss(ss_rc, "\0", "\0");
locktypename = tnbuf;
}
values[0] = CStringGetTextDatum(locktypename);
switch ((LockTagType)instance->locktag.locktag_type) {
case LOCKTAG_RELATION:
case LOCKTAG_CSTORE_FREESPACE:
values[1] = ObjectIdGetDatum(instance->locktag.locktag_field1);
values[2] = ObjectIdGetDatum(instance->locktag.locktag_field2);
nulls[3] = true;
nulls[4] = true;
nulls[5] = true;
nulls[6] = true;
nulls[7] = true;
nulls[8] = true;
nulls[9] = true;
nulls[10] = true;
break;
case LOCKTAG_RELFILENODE:
values[1] = ObjectIdGetDatum(instance->locktag.locktag_field1);
values[2] = ObjectIdGetDatum(instance->locktag.locktag_field2);
values[3] = ObjectIdGetDatum(instance->locktag.locktag_field3);
nulls[4] = true;
nulls[5] = true;
nulls[6] = true;
nulls[7] = true;
nulls[8] = true;
nulls[9] = true;
nulls[10] = true;
break;
case LOCKTAG_RELATION_EXTEND:
values[1] = ObjectIdGetDatum(instance->locktag.locktag_field1);
values[2] = ObjectIdGetDatum(instance->locktag.locktag_field2);
values[5] = UInt16GetDatum(instance->locktag.locktag_field5);
nulls[3] = true;
nulls[4] = true;
nulls[6] = true;
nulls[7] = true;
nulls[8] = true;
nulls[9] = true;
nulls[10] = true;
break;
case LOCKTAG_PAGE:
values[1] = ObjectIdGetDatum(instance->locktag.locktag_field1);
values[2] = ObjectIdGetDatum(instance->locktag.locktag_field2);
values[3] = UInt32GetDatum(instance->locktag.locktag_field3);
values[5] = UInt16GetDatum(instance->locktag.locktag_field5);
nulls[4] = true;
nulls[6] = true;
nulls[7] = true;
nulls[8] = true;
nulls[9] = true;
nulls[10] = true;
break;
case LOCKTAG_TUPLE:
values[1] = ObjectIdGetDatum(instance->locktag.locktag_field1);
values[2] = ObjectIdGetDatum(instance->locktag.locktag_field2);
values[3] = UInt32GetDatum(instance->locktag.locktag_field3);
values[4] = UInt32GetDatum(instance->locktag.locktag_field4);
values[5] = UInt16GetDatum(instance->locktag.locktag_field5);
nulls[6] = true;
nulls[7] = true;
nulls[8] = true;
nulls[9] = true;
nulls[10] = true;
break;
case LOCKTAG_UID:
values[1] = ObjectIdGetDatum(instance->locktag.locktag_field1);
values[2] = ObjectIdGetDatum(instance->locktag.locktag_field2);
nulls[3] = true;
nulls[4] = true;
nulls[5] = true;
nulls[6] = true;
values[7] = TransactionIdGetDatum((uint64)instance->locktag.locktag_field3 << 32 |
((uint64)instance->locktag.locktag_field4));
nulls[8] = true;
nulls[9] = true;
nulls[10] = true;
break;
case LOCKTAG_TRANSACTION:
values[7] = TransactionIdGetDatum((TransactionId)instance->locktag.locktag_field1 |
((TransactionId)instance->locktag.locktag_field2 << 32));
nulls[1] = true;
nulls[2] = true;
nulls[3] = true;
nulls[4] = true;
nulls[5] = true;
nulls[6] = true;
nulls[8] = true;
nulls[9] = true;
nulls[10] = true;
break;
case LOCKTAG_VIRTUALTRANSACTION:
values[6] = VXIDGetDatum(instance->locktag.locktag_field1,
(TransactionId)instance->locktag.locktag_field2 |
((TransactionId)instance->locktag.locktag_field3 << 32));
nulls[1] = true;
nulls[2] = true;
nulls[3] = true;
nulls[4] = true;
nulls[5] = true;
nulls[7] = true;
nulls[8] = true;
nulls[9] = true;
nulls[10] = true;
break;
case LOCKTAG_TABLESPACE:
nulls[1] = ObjectIdGetDatum(instance->locktag.locktag_field2);
nulls[2] = true;
nulls[3] = true;
nulls[4] = true;
nulls[5] = true;
nulls[6] = true;
nulls[7] = true;
nulls[8] = true;
nulls[9] = true;
nulls[10] = true;
break;
case LOCKTAG_OBJECT:
case LOCKTAG_USERLOCK:
case LOCKTAG_ADVISORY:
default:
values[1] = ObjectIdGetDatum(instance->locktag.locktag_field1);
values[8] = ObjectIdGetDatum(instance->locktag.locktag_field2);
values[9] = ObjectIdGetDatum(instance->locktag.locktag_field3);
values[10] = Int32GetDatum(instance->locktag.locktag_field4);
nulls[2] = true;
nulls[3] = true;
nulls[4] = true;
nulls[5] = true;
nulls[6] = true;
nulls[7] = true;
break;
}
values[11] = VXIDGetDatum(instance->backend, instance->lxid);
if (instance->pid != 0)
values[12] = Int64GetDatum(instance->pid);
else
nulls[12] = true;
if (instance->sessionid != 0)
values[13] = Int64GetDatum(instance->sessionid);
else
nulls[13] = true;
values[14] = CStringGetTextDatum(GetLockmodeName(instance->locktag.locktag_lockmethodid, mode));
values[15] = BoolGetDatum(granted);
values[16] = BoolGetDatum(instance->fastpath);
GetLocktagInfo(instance, values);
char* gId = GetGlobalSessionStr(instance->globalSessionId);
values[18] = CStringGetTextDatum(gId);
pfree(gId);
if (!granted && instance->waitStart != 0) {
values[19] = TimestampTzGetDatum(instance->waitStart);
} else {
nulls[19] = true;
}
tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
result = HeapTupleGetDatum(tuple);
SRF_RETURN_NEXT(funcctx, result);
}
* Have returned all regular locks. Now start on the SIREAD predicate
* locks.
*/
predLockData = mystatus->predLockData;
if (mystatus->predLockIdx < predLockData->nelements) {
PredicateLockTargetType lockType;
PREDICATELOCKTARGETTAG* predTag = &(predLockData->locktags[mystatus->predLockIdx]);
SERIALIZABLEXACT* xact = &(predLockData->xacts[mystatus->predLockIdx]);
Datum values[NUM_LOCK_STATUS_COLUMNS];
bool nulls[NUM_LOCK_STATUS_COLUMNS];
HeapTuple tuple;
Datum result;
char* blocklocktag = NULL;
mystatus->predLockIdx++;
* Form tuple with appropriate data.
*/
errno_t rc = EOK;
rc = memset_s(values, sizeof(values), 0, sizeof(values));
securec_check(rc, "\0", "\0");
rc = memset_s(nulls, sizeof(nulls), false, sizeof(nulls));
securec_check(rc, "\0", "\0");
lockType = GET_PREDICATELOCKTARGETTAG_TYPE(*predTag);
values[0] = CStringGetTextDatum(PredicateLockTagTypeNames[lockType]);
values[1] = GET_PREDICATELOCKTARGETTAG_DB(*predTag);
values[2] = GET_PREDICATELOCKTARGETTAG_RELATION(*predTag);
if (lockType == PREDLOCKTAG_TUPLE)
values[4] = GET_PREDICATELOCKTARGETTAG_OFFSET(*predTag);
else
nulls[4] = true;
if ((lockType == PREDLOCKTAG_TUPLE) || (lockType == PREDLOCKTAG_PAGE))
values[3] = GET_PREDICATELOCKTARGETTAG_PAGE(*predTag);
else
nulls[3] = true;
nulls[5] = true;
nulls[6] = true;
nulls[7] = true;
nulls[8] = true;
nulls[9] = true;
nulls[10] = true;
values[11] = VXIDGetDatum(xact->vxid.backendId, xact->vxid.localTransactionId);
if (xact->pid != 0)
values[12] = Int64GetDatum(xact->pid);
else
nulls[12] = true;
nulls[13] = true;
* Lock mode. Currently all predicate locks are SIReadLocks, which are
* always held (never waiting) and have no fast path
*/
values[14] = CStringGetTextDatum("SIReadLock");
values[15] = BoolGetDatum(true);
values[16] = BoolGetDatum(false);
blocklocktag = LockPredTagToString(predTag);
values[NUM_LOCKTAG_ID] = CStringGetTextDatum(blocklocktag);
pfree_ext(blocklocktag);
nulls[18] = true;
nulls[19] = true;
tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
result = HeapTupleGetDatum(tuple);
SRF_RETURN_NEXT(funcctx, result);
}
SRF_RETURN_DONE(funcctx);
}
* Functions for manipulating advisory locks
*
* We make use of the locktag fields as follows:
*
* field1: u_sess->proc_cxt.MyDatabaseId ... ensures locks are local to each database
* field2: first of 2 int4 keys, or high-order half of an int8 key
* field3: second of 2 int4 keys, or low-order half of an int8 key
* field4: 1 if using an int8 key, 2 if using 2 int4 keys
*/
#define SET_LOCKTAG_INT64(tag, key64) \
SET_LOCKTAG_ADVISORY(tag, u_sess->proc_cxt.MyDatabaseId, (uint32)((key64) >> 32), (uint32)(key64), 1)
#define SET_LOCKTAG_INT32(tag, key1, key2) SET_LOCKTAG_ADVISORY(tag, u_sess->proc_cxt.MyDatabaseId, key1, key2, 2)
#define SET_LOCKTAG_INT32_DB(tag, databaseOid, key1, key2) SET_LOCKTAG_ADVISORY(tag, databaseOid, key1, key2, 2)
static void CheckIfAnySchemaInRedistribution()
{
Relation rel = heap_open(NamespaceRelationId, AccessShareLock);
TableScanDesc scan = tableam_scan_begin(rel, SnapshotNow, 0, NULL);
bool isNull = false;
HeapTuple tuple;
Datum datum;
while ((tuple = (HeapTuple) tableam_scan_getnexttuple(scan, ForwardScanDirection)) != NULL) {
datum = heap_getattr(tuple, Anum_pg_namespace_in_redistribution, RelationGetDescr(rel), &isNull);
if (isNull) {
continue;
}
if (DatumGetChar(datum) == 'y') {
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("Please check if another schema is in redistribution in the same database.")));
}
}
tableam_scan_end(scan);
heap_close(rel, NoLock);
}
static void UpdateSchemaInRedistribution(Name schemaName, bool isLock)
{
Relation rel = heap_open(NamespaceRelationId, RowExclusiveLock);
HeapTuple tuple = SearchSysCache1(NAMESPACENAME, CStringGetDatum(schemaName->data));
if (!HeapTupleIsValid(tuple)) {
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_SCHEMA),
errmsg("schema \"%s\" does not exist", schemaName->data)));
}
Oid schemaOid = HeapTupleGetOid(tuple);
if (schemaOid < FirstNormalObjectId && schemaOid != PG_PUBLIC_NAMESPACE) {
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("system schema \"%s\" does not support transfer", schemaName->data)));
}
Datum newRecord[Natts_pg_namespace];
bool newRecordNulls[Natts_pg_namespace];
bool newRecordRepl[Natts_pg_namespace];
errno_t rc = memset_s(newRecord, sizeof(newRecord), 0, sizeof(newRecord));
securec_check(rc, "\0", "\0");
rc = memset_s(newRecordNulls, sizeof(newRecordNulls), false, sizeof(newRecordNulls));
securec_check(rc, "\0", "\0");
rc = memset_s(newRecordRepl, sizeof(newRecordRepl), false, sizeof(newRecordRepl));
securec_check(rc, "\0", "\0");
if (isLock) {
newRecord[Anum_pg_namespace_in_redistribution - 1] = CharGetDatum('n');
} else {
newRecord[Anum_pg_namespace_in_redistribution - 1] = CharGetDatum('y');
}
newRecordRepl[Anum_pg_namespace_in_redistribution - 1] = true;
HeapTuple newtuple = heap_modify_tuple(tuple, RelationGetDescr(rel), newRecord, newRecordNulls, newRecordRepl);
simple_heap_update(rel, &tuple->t_self, newtuple);
CatalogUpdateIndexes(rel, newtuple);
heap_freetuple_ext(newtuple);
ReleaseSysCache(tuple);
heap_close(rel, NoLock);
}
#ifdef PGXC
static void PGXCSendTransfer(Name schemaName, bool isLock)
{
char updateSql[CHAR_BUF_SIZE] = {0};
int rc;
if (isLock) {
rc = snprintf_s(updateSql, CHAR_BUF_SIZE, CHAR_BUF_SIZE - 1,
"select pg_catalog.pgxc_unlock_for_transfer('%s'::name)", schemaName->data);
} else {
rc = snprintf_s(updateSql, CHAR_BUF_SIZE, CHAR_BUF_SIZE - 1,
"select pg_catalog.pgxc_lock_for_transfer('%s'::name)", schemaName->data);
}
securec_check_ss(rc, "\0", "\0");
ExecUtilityStmtOnNodes(updateSql, NULL, false, false, EXEC_ON_COORDS, false);
}
#define MAXINT8LEN 25
* pgxc_advisory_lock - Core function that implements the algorithm needed to
* propogate the advisory lock function calls to all Coordinators.
* The idea is to make the advisory locks cluster-aware, so that a user having
* a lock from Coordinator 1 will make the user from Coordinator 2 to wait for
* the same lock.
*
* Return true if all locks are returned successfully. False otherwise.
* Effectively this function returns false only if dontWait is true. Otherwise
* it either returns true, or waits on a resource, or throws an exception
* returned by the lock function calls in case of unexpected or fatal errors.
*
* Currently used only for session level locks; not used for transaction level
* locks.
*/
static bool pgxc_advisory_lock(int64 key64, int32 key1, int32 key2, bool iskeybig, LOCKMODE lockmode,
LockLevel locklevel, TryType locktry, Name databaseName)
{
LOCKTAG locktag;
Oid *coOids = NULL, *dnOids = NULL;
int numdnodes, numcoords;
StringInfoData lock_cmd, unlock_cmd, lock_funcname, unlock_funcname, args;
char str_key[MAXINT8LEN + 1];
int i, prev;
bool abort_locking = false;
Datum lock_status;
bool sessionLock = (locklevel == SESSION_LOCK);
bool dontWait = (locktry == DONT_WAIT);
bool sp_database = (databaseName != NULL);
Oid databaseOid = u_sess->proc_cxt.MyDatabaseId;
if (sp_database)
databaseOid = get_database_oid(databaseName->data, false);
if (iskeybig)
SET_LOCKTAG_INT64(locktag, key64);
else
SET_LOCKTAG_INT32_DB(locktag, databaseOid, key1, key2);
* Before get cn/dn oids, we should refresh NumCoords/NumDataNodes and co_handles/dn_handles in u_sess in case of
* can not process SIGUSR1 of "pgxc_pool_reload" command immediately.
*/
#ifdef ENABLE_MULTIPLE_NODES
ReloadPoolerWithoutTransaction();
#endif
PgxcNodeGetOids(&coOids, &dnOids, &numcoords, &numdnodes, false);
if (numcoords <= 1) {
LockAcquireResult res;
res = LockAcquire(&locktag, lockmode, sessionLock, dontWait);
return (res == LOCKACQUIRE_OK || res == LOCKACQUIRE_ALREADY_HELD);
}
* If there is already a lock held by us, just increment and return; we
* already did all necessary steps when we locked for the first time.
*/
if (LockIncrementIfExists(&locktag, lockmode, sessionLock) == true)
return true;
initStringInfo(&lock_funcname);
appendStringInfo(&lock_funcname,
"pg_%sadvisory_%slock%s",
(dontWait ? "try_" : ""),
(sessionLock ? "" : "xact_"),
((lockmode == ShareLock) ? "_shared" : ""));
initStringInfo(&unlock_funcname);
appendStringInfo(&unlock_funcname, "pg_advisory_unlock%s", ((lockmode == ShareLock) ? "_shared" : ""));
initStringInfo(&args);
if (iskeybig) {
pg_lltoa(key64, str_key);
appendStringInfo(&args, "%s", str_key);
} else {
pg_ltoa(key1, str_key);
appendStringInfo(&args, "%s, ", str_key);
pg_ltoa(key2, str_key);
appendStringInfo(&args,
"%s%s%s%s",
str_key,
(sp_database ? ", \'" : ""),
(sp_database ? databaseName->data : ""),
(sp_database ? "\'" : ""));
}
initStringInfo(&lock_cmd);
appendStringInfo(&lock_cmd, "SELECT pg_catalog.%s(%s)", lock_funcname.data, args.data);
initStringInfo(&unlock_cmd);
appendStringInfo(&unlock_cmd, "SELECT pg_catalog.%s(%s)", unlock_funcname.data, args.data);
* Go on locking on each Coordinator. Keep on unlocking the previous one
* after a lock is held on next Coordinator. Don't unlock the local
* Coordinator. After finishing all Coordinators, ultimately only the local
* Coordinator would be locked, but still we will have scanned all
* Coordinators to make sure no one else has already grabbed the lock. The
* reason for unlocking all remote locks is because the session level locks
* don't get unlocked until explicitly unlocked or the session quits. After
* the user session quits without explicitly unlocking, the coord-to-coord
* pooler connection stays and so does the remote Coordinator lock.
*/
prev = -1;
for (i = 0; i <= numcoords && !abort_locking; i++, prev++) {
if (i < numcoords) {
if (i == u_sess->pgxc_cxt.PGXCNodeId - 1)
lock_status = LockAcquire(&locktag, lockmode, sessionLock, dontWait);
else
lock_status = pgxc_execute_on_nodes(1, &coOids[i], lock_cmd.data);
if (dontWait == true && DatumGetBool(lock_status) == false) {
abort_locking = true;
* If we have gone past the local Coordinator node, it implies
* that we have obtained a local lock. But now that we are
* aborting, we need to release the local lock first.
*/
if (i > u_sess->pgxc_cxt.PGXCNodeId - 1)
(void)LockRelease(&locktag, lockmode, sessionLock);
}
}
* If we are dealing with session locks, unlock the previous lock, but
* only if it is a remote Coordinator. If it is a local one, we want to
* keep that lock. Remember, the final status should be that there is
* only *one* lock held, and that is the local lock.
*/
if (sessionLock && prev >= 0 && prev != u_sess->pgxc_cxt.PGXCNodeId - 1)
pgxc_execute_on_nodes(1, &coOids[prev], unlock_cmd.data);
}
return (!abort_locking);
}
#endif
* pg_advisory_lock(int8) - acquire exclusive lock on an int8 key
*/
Datum pg_advisory_lock_int8(PG_FUNCTION_ARGS)
{
int64 key = PG_GETARG_INT64(0);
LOCKTAG tag;
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
(void)pgxc_advisory_lock(key, 0, 0, true, ExclusiveLock, SESSION_LOCK, WAIT);
PG_RETURN_VOID();
}
#endif
SET_LOCKTAG_INT64(tag, key);
(void)LockAcquire(&tag, ExclusiveLock, true, false);
PG_RETURN_VOID();
}
* pg_advisory_xact_lock(int8) - acquire xact scoped
* exclusive lock on an int8 key
*/
Datum pg_advisory_xact_lock_int8(PG_FUNCTION_ARGS)
{
int64 key = PG_GETARG_INT64(0);
LOCKTAG tag;
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
(void)pgxc_advisory_lock(key, 0, 0, true, ExclusiveLock, TRANSACTION_LOCK, WAIT);
PG_RETURN_VOID();
}
#endif
SET_LOCKTAG_INT64(tag, key);
(void)LockAcquire(&tag, ExclusiveLock, false, false);
PG_RETURN_VOID();
}
* pg_advisory_lock_shared(int8) - acquire share lock on an int8 key
*/
Datum pg_advisory_lock_shared_int8(PG_FUNCTION_ARGS)
{
int64 key = PG_GETARG_INT64(0);
LOCKTAG tag;
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
(void)pgxc_advisory_lock(key, 0, 0, true, ShareLock, SESSION_LOCK, WAIT);
PG_RETURN_VOID();
}
#endif
SET_LOCKTAG_INT64(tag, key);
(void)LockAcquire(&tag, ShareLock, true, false);
PG_RETURN_VOID();
}
* pg_advisory_xact_lock_shared(int8) - acquire xact scoped
* share lock on an int8 key
*/
Datum pg_advisory_xact_lock_shared_int8(PG_FUNCTION_ARGS)
{
int64 key = PG_GETARG_INT64(0);
LOCKTAG tag;
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
(void)pgxc_advisory_lock(key, 0, 0, true, ShareLock, TRANSACTION_LOCK, WAIT);
PG_RETURN_VOID();
}
#endif
SET_LOCKTAG_INT64(tag, key);
(void)LockAcquire(&tag, ShareLock, false, false);
PG_RETURN_VOID();
}
* pg_try_advisory_lock(int8) - acquire exclusive lock on an int8 key, no wait
*
* Returns true if successful, false if lock not available
*/
Datum pg_try_advisory_lock_int8(PG_FUNCTION_ARGS)
{
int64 key = PG_GETARG_INT64(0);
LOCKTAG tag;
LockAcquireResult res;
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord())
PG_RETURN_BOOL(pgxc_advisory_lock(key, 0, 0, true, ExclusiveLock, SESSION_LOCK, DONT_WAIT));
#endif
SET_LOCKTAG_INT64(tag, key);
res = LockAcquire(&tag, ExclusiveLock, true, true);
PG_RETURN_BOOL(res != LOCKACQUIRE_NOT_AVAIL);
}
* pg_try_advisory_xact_lock(int8) - acquire xact scoped
* exclusive lock on an int8 key, no wait
*
* Returns true if successful, false if lock not available
*/
Datum pg_try_advisory_xact_lock_int8(PG_FUNCTION_ARGS)
{
int64 key = PG_GETARG_INT64(0);
LOCKTAG tag;
LockAcquireResult res;
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord())
PG_RETURN_BOOL(pgxc_advisory_lock(key, 0, 0, true, ExclusiveLock, TRANSACTION_LOCK, DONT_WAIT));
#endif
SET_LOCKTAG_INT64(tag, key);
res = LockAcquire(&tag, ExclusiveLock, false, true);
PG_RETURN_BOOL(res != LOCKACQUIRE_NOT_AVAIL);
}
* pg_try_advisory_lock_shared(int8) - acquire share lock on an int8 key, no wait
*
* Returns true if successful, false if lock not available
*/
Datum pg_try_advisory_lock_shared_int8(PG_FUNCTION_ARGS)
{
int64 key = PG_GETARG_INT64(0);
LOCKTAG tag;
LockAcquireResult res;
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord())
PG_RETURN_BOOL(pgxc_advisory_lock(key, 0, 0, true, ShareLock, SESSION_LOCK, DONT_WAIT));
#endif
SET_LOCKTAG_INT64(tag, key);
res = LockAcquire(&tag, ShareLock, true, true);
PG_RETURN_BOOL(res != LOCKACQUIRE_NOT_AVAIL);
}
* pg_try_advisory_xact_lock_shared(int8) - acquire xact scoped
* share lock on an int8 key, no wait
*
* Returns true if successful, false if lock not available
*/
Datum pg_try_advisory_xact_lock_shared_int8(PG_FUNCTION_ARGS)
{
int64 key = PG_GETARG_INT64(0);
LOCKTAG tag;
LockAcquireResult res;
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord())
PG_RETURN_BOOL(pgxc_advisory_lock(key, 0, 0, true, ShareLock, TRANSACTION_LOCK, DONT_WAIT));
#endif
SET_LOCKTAG_INT64(tag, key);
res = LockAcquire(&tag, ShareLock, false, true);
PG_RETURN_BOOL(res != LOCKACQUIRE_NOT_AVAIL);
}
* pg_advisory_unlock(int8) - release exclusive lock on an int8 key
*
* Returns true if successful, false if lock was not held
*/
Datum pg_advisory_unlock_int8(PG_FUNCTION_ARGS)
{
int64 key = PG_GETARG_INT64(0);
LOCKTAG tag;
bool res = false;
SET_LOCKTAG_INT64(tag, key);
res = LockRelease(&tag, ExclusiveLock, true);
PG_RETURN_BOOL(res);
}
* pg_advisory_unlock_shared(int8) - release share lock on an int8 key
*
* Returns true if successful, false if lock was not held
*/
Datum pg_advisory_unlock_shared_int8(PG_FUNCTION_ARGS)
{
int64 key = PG_GETARG_INT64(0);
LOCKTAG tag;
bool res = false;
SET_LOCKTAG_INT64(tag, key);
res = LockRelease(&tag, ShareLock, true);
PG_RETURN_BOOL(res);
}
* pg_advisory_lock(int4, int4) - acquire exclusive lock on 2 int4 keys
*/
Datum pg_advisory_lock_int4(PG_FUNCTION_ARGS)
{
int32 key1 = PG_GETARG_INT32(0);
int32 key2 = PG_GETARG_INT32(1);
LOCKTAG tag;
if (key1 == XC_LOCK_FOR_BACKUP_KEY_1 && key2 == XC_LOCK_FOR_BACKUP_KEY_2 && !superuser())
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("Only system admin can lock the cluster.")));
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
(void)pgxc_advisory_lock(0, key1, key2, false, ExclusiveLock, SESSION_LOCK, WAIT);
elog(INFO, "please do not close this session until you are done adding the new node");
PG_RETURN_VOID();
}
#endif
SET_LOCKTAG_INT32(tag, key1, key2);
(void)LockAcquire(&tag, ExclusiveLock, true, false);
PG_RETURN_VOID();
}
* pg_advisory_lock(int4, int4, Name) - acquire exclusive lock on 2 int4 keys for specific database
*/
Datum pg_advisory_lock_sp_db_int4(PG_FUNCTION_ARGS)
{
int32 key1 = PG_GETARG_INT32(0);
int32 key2 = PG_GETARG_INT32(1);
Name databaseName = PG_GETARG_NAME(2);
LOCKTAG tag;
Oid databaseOid = u_sess->proc_cxt.MyDatabaseId;
if (key1 == XC_LOCK_FOR_BACKUP_KEY_1 && key2 == XC_LOCK_FOR_BACKUP_KEY_2 && !superuser())
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("Only system admin can lock the cluster.")));
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
(void)pgxc_advisory_lock(0, key1, key2, false, ExclusiveLock, SESSION_LOCK, WAIT, databaseName);
elog(INFO, "please do not close this session until you are done adding the new node");
PG_RETURN_VOID();
}
#endif
if (databaseName != NULL)
databaseOid = get_database_oid(databaseName->data, false);
SET_LOCKTAG_INT32_DB(tag, databaseOid, key1, key2);
(void)LockAcquire(&tag, ExclusiveLock, true, false);
PG_RETURN_VOID();
}
* pg_advisory_xact_lock(int4, int4) - acquire xact scoped
* exclusive lock on 2 int4 keys
*/
Datum pg_advisory_xact_lock_int4(PG_FUNCTION_ARGS)
{
int32 key1 = PG_GETARG_INT32(0);
int32 key2 = PG_GETARG_INT32(1);
LOCKTAG tag;
if (key1 == XC_LOCK_FOR_BACKUP_KEY_1 && key2 == XC_LOCK_FOR_BACKUP_KEY_2 && !superuser())
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("Only system admin can lock the cluster.")));
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
(void)pgxc_advisory_lock(0, key1, key2, false, ExclusiveLock, TRANSACTION_LOCK, WAIT);
PG_RETURN_VOID();
}
#endif
SET_LOCKTAG_INT32(tag, key1, key2);
(void)LockAcquire(&tag, ExclusiveLock, false, false);
PG_RETURN_VOID();
}
* pg_advisory_lock_shared(int4, int4) - acquire share lock on 2 int4 keys
*/
Datum pg_advisory_lock_shared_int4(PG_FUNCTION_ARGS)
{
int32 key1 = PG_GETARG_INT32(0);
int32 key2 = PG_GETARG_INT32(1);
LOCKTAG tag;
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
(void)pgxc_advisory_lock(0, key1, key2, false, ShareLock, SESSION_LOCK, WAIT);
PG_RETURN_VOID();
}
#endif
SET_LOCKTAG_INT32(tag, key1, key2);
(void)LockAcquire(&tag, ShareLock, true, false);
PG_RETURN_VOID();
}
* pg_advisory_xact_lock_shared(int4, int4) - acquire xact scoped
* share lock on 2 int4 keys
*/
Datum pg_advisory_xact_lock_shared_int4(PG_FUNCTION_ARGS)
{
int32 key1 = PG_GETARG_INT32(0);
int32 key2 = PG_GETARG_INT32(1);
LOCKTAG tag;
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
(void)pgxc_advisory_lock(0, key1, key2, false, ShareLock, TRANSACTION_LOCK, WAIT);
PG_RETURN_VOID();
}
#endif
SET_LOCKTAG_INT32(tag, key1, key2);
(void)LockAcquire(&tag, ShareLock, false, false);
PG_RETURN_VOID();
}
* pg_try_advisory_lock(int4, int4) - acquire exclusive lock on 2 int4 keys, no wait
*
* Returns true if successful, false if lock not available
*/
Datum pg_try_advisory_lock_int4(PG_FUNCTION_ARGS)
{
int32 key1 = PG_GETARG_INT32(0);
int32 key2 = PG_GETARG_INT32(1);
LOCKTAG tag;
LockAcquireResult res;
if (key1 == XC_LOCK_FOR_BACKUP_KEY_1 && key2 == XC_LOCK_FOR_BACKUP_KEY_2 && !superuser())
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("Only system admin can lock the cluster.")));
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord())
PG_RETURN_BOOL(pgxc_advisory_lock(0, key1, key2, false, ExclusiveLock, SESSION_LOCK, DONT_WAIT));
#endif
SET_LOCKTAG_INT32(tag, key1, key2);
res = LockAcquire(&tag, ExclusiveLock, true, true);
PG_RETURN_BOOL(res != LOCKACQUIRE_NOT_AVAIL);
}
* pg_try_advisory_xact_lock(int4, int4) - acquire xact scoped
* exclusive lock on 2 int4 keys, no wait
*
* Returns true if successful, false if lock not available
*/
Datum pg_try_advisory_xact_lock_int4(PG_FUNCTION_ARGS)
{
int32 key1 = PG_GETARG_INT32(0);
int32 key2 = PG_GETARG_INT32(1);
LOCKTAG tag;
LockAcquireResult res;
if (key1 == XC_LOCK_FOR_BACKUP_KEY_1 && key2 == XC_LOCK_FOR_BACKUP_KEY_2 && !superuser())
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("Only system admin can lock the cluster.")));
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord())
PG_RETURN_BOOL(pgxc_advisory_lock(0, key1, key2, false, ExclusiveLock, TRANSACTION_LOCK, DONT_WAIT));
#endif
SET_LOCKTAG_INT32(tag, key1, key2);
res = LockAcquire(&tag, ExclusiveLock, false, true);
PG_RETURN_BOOL(res != LOCKACQUIRE_NOT_AVAIL);
}
* pg_try_advisory_lock_shared(int4, int4) - acquire share lock on 2 int4 keys, no wait
*
* Returns true if successful, false if lock not available
*/
Datum pg_try_advisory_lock_shared_int4(PG_FUNCTION_ARGS)
{
int32 key1 = PG_GETARG_INT32(0);
int32 key2 = PG_GETARG_INT32(1);
LOCKTAG tag;
LockAcquireResult res;
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord())
PG_RETURN_BOOL(pgxc_advisory_lock(0, key1, key2, false, ShareLock, SESSION_LOCK, DONT_WAIT));
#endif
SET_LOCKTAG_INT32(tag, key1, key2);
res = LockAcquire(&tag, ShareLock, true, true);
PG_RETURN_BOOL(res != LOCKACQUIRE_NOT_AVAIL);
}
* pg_try_advisory_xact_lock_shared(int4, int4) - acquire xact scoped
* share lock on 2 int4 keys, no wait
*
* Returns true if successful, false if lock not available
*/
Datum pg_try_advisory_xact_lock_shared_int4(PG_FUNCTION_ARGS)
{
int32 key1 = PG_GETARG_INT32(0);
int32 key2 = PG_GETARG_INT32(1);
LOCKTAG tag;
LockAcquireResult res;
#ifdef PGXC
if (IS_PGXC_COORDINATOR && !IsConnFromCoord())
PG_RETURN_BOOL(pgxc_advisory_lock(0, key1, key2, false, ShareLock, TRANSACTION_LOCK, DONT_WAIT));
#endif
SET_LOCKTAG_INT32(tag, key1, key2);
res = LockAcquire(&tag, ShareLock, false, true);
PG_RETURN_BOOL(res != LOCKACQUIRE_NOT_AVAIL);
}
* pg_advisory_unlock(int4, int4) - release exclusive lock on 2 int4 keys
*
* Returns true if successful, false if lock was not held
*/
Datum pg_advisory_unlock_int4(PG_FUNCTION_ARGS)
{
int32 key1 = PG_GETARG_INT32(0);
int32 key2 = PG_GETARG_INT32(1);
LOCKTAG tag;
bool res = false;
SET_LOCKTAG_INT32(tag, key1, key2);
res = LockRelease(&tag, ExclusiveLock, true);
PG_RETURN_BOOL(res);
}
* pg_advisory_unlock(int4, int4, Name) - release exclusive lock on 2 int4 keys for specific database
*
* Returns true if successful, false if lock was not held
*/
Datum pg_advisory_unlock_sp_db_int4(PG_FUNCTION_ARGS)
{
int32 key1 = PG_GETARG_INT32(0);
int32 key2 = PG_GETARG_INT32(1);
Name databaseName = PG_GETARG_NAME(2);
LOCKTAG tag;
bool res = false;
Oid databaseOid = u_sess->proc_cxt.MyDatabaseId;
if (databaseName != NULL)
databaseOid = get_database_oid(databaseName->data, false);
SET_LOCKTAG_INT32_DB(tag, databaseOid, key1, key2);
res = LockRelease(&tag, ExclusiveLock, true);
PG_RETURN_BOOL(res);
}
* pg_advisory_unlock_shared(int4, int4) - release share lock on 2 int4 keys
*
* Returns true if successful, false if lock was not held
*/
Datum pg_advisory_unlock_shared_int4(PG_FUNCTION_ARGS)
{
int32 key1 = PG_GETARG_INT32(0);
int32 key2 = PG_GETARG_INT32(1);
LOCKTAG tag;
bool res = false;
SET_LOCKTAG_INT32(tag, key1, key2);
res = LockRelease(&tag, ShareLock, true);
PG_RETURN_BOOL(res);
}
* pg_advisory_unlock_all() - release all advisory locks
*/
Datum pg_advisory_unlock_all(PG_FUNCTION_ARGS)
{
LockReleaseSession(USER_LOCKMETHOD);
PG_RETURN_VOID();
}
#ifdef DOLPHIN
Datum GetAdvisoryLockWithtimeTextFormat(PG_FUNCTION_ARGS)
{
char *lockname = text_to_cstring(PG_GETARG_TEXT_PP(0));
char *timeoutStr = text_to_cstring(PG_GETARG_TEXT_PP(1));
char *ptr = NULL;
double timeout = strtod(timeoutStr, &ptr);
if (strlen(ptr) > 0) {
ereport(WARNING, (errmsg("truncated incorrect value: %s", ptr)));
}
int ret = GetAdvisoryLockWithTimeInner(lockname, timeout);
if (ret == -1) {
PG_RETURN_NULL();
}
PG_RETURN_INT64(ret);
}
Datum GetAdvisoryLockWithtime(PG_FUNCTION_ARGS)
{
char *lockname = text_to_cstring(PG_GETARG_TEXT_PP(0));
double timeout = PG_GETARG_FLOAT8(1);
int ret = GetAdvisoryLockWithTimeInner(lockname, timeout);
if (ret == -1) {
PG_RETURN_NULL();
}
PG_RETURN_INT32(ret);
}
Datum GetAdvisoryLock(PG_FUNCTION_ARGS)
{
char *lockname = text_to_cstring(PG_GETARG_TEXT_PP(0));
LOCKTAG tag;
uint32 lockid[4] = {0};
uint64 currentSessionId = IS_THREAD_POOL_WORKER ? u_sess->session_id : t_thrd.proc_cxt.MyProcPid;
if (g_instance.attr.attr_common.enable_thread_pool) {
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("Unsupport using advisory lock in threadpool")));
}
AdvisoryLockGetIdFromName(lockname, lockid);
SET_LOCKTAG_ADVISORY_5(tag, u_sess->proc_cxt.MyDatabaseId, lockid[0], lockid[1], lockid[2], lockid[3]);
ClearInvalidLockNameInner();
bool nlockInitialFlag = (DatumGetUInt64(IsUsedAdvisoryLockInner(lockname)) != currentSessionId);
LockAcquireResult res = LockAcquire(&tag, ExclusiveLock, true, false, false);
if (res == LOCKACQUIRE_NOT_AVAIL) {
PG_RETURN_INT32(0);
} else {
SetLockNameHash(lockname, nlockInitialFlag);
PG_RETURN_INT32(1);
}
}
Datum ReleaseAdvisoryLock(PG_FUNCTION_ARGS)
{
char *lockname = text_to_cstring(PG_GETARG_TEXT_PP(0));
LOCKTAG tag;
if (g_instance.attr.attr_common.enable_thread_pool) {
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("Unsupport using advisory lock in threadpool")));
}
if (IsFreeAdvisoryLockInner(lockname))
PG_RETURN_NULL();
uint32 lockid[4] = {0};
AdvisoryLockGetIdFromName(lockname, lockid);
SET_LOCKTAG_ADVISORY_5(tag, u_sess->proc_cxt.MyDatabaseId, lockid[0], lockid[1], lockid[2], lockid[3]);
bool res = LockRelease(&tag, ExclusiveLock, true);
if (res) {
SetLockNameHash(lockname, false, true);
PG_RETURN_INT32(1);
}
else
PG_RETURN_INT32(0);
}
Datum IsFreeAdvisoryLock(PG_FUNCTION_ARGS)
{
char *lockname = text_to_cstring(PG_GETARG_TEXT_PP(0));
if (g_instance.attr.attr_common.enable_thread_pool) {
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("Unsupport using advisory lock in threadpool")));
}
if (IsFreeAdvisoryLockInner(lockname))
PG_RETURN_INT32(1);
PG_RETURN_INT32(0);
}
Datum IsUsedAdvisoryLock(PG_FUNCTION_ARGS)
{
char *lockname = text_to_cstring(PG_GETARG_TEXT_PP(0));
Datum sessionid = IsUsedAdvisoryLockInner(lockname);
if (sessionid == 0)
PG_RETURN_NULL();
else
return sessionid;
}
Datum ReleaseAllAdvisoryLock(PG_FUNCTION_ARGS)
{
if (g_instance.attr.attr_common.enable_thread_pool) {
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("Unsupport using advisory lock in threadpool")));
}
ClearInvalidLockNameInner();
LockReleaseSession(USER_LOCKMETHOD);
auto ret = LockNameHashRelease();
PG_RETURN_INT64(ret);
}
Datum GetAllAdvisoryLock(PG_FUNCTION_ARGS)
{
FuncCallContext* funcctx = NULL;
HASH_SEQ_STATUS* status = NULL;
if (SRF_IS_FIRSTCALL()) {
TupleDesc tupdesc;
MemoryContext oldcontext;
funcctx = SRF_FIRSTCALL_INIT();
* switch to memory context appropriate for multiple function calls
*/
oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
tupdesc = CreateTemplateTupleDesc(2, false);
TupleDescInitEntry(tupdesc, (AttrNumber)1, "lockname", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)2, "sessionid", INT8OID, -1, 0);
funcctx->tuple_desc = BlessTupleDesc(tupdesc);
status = (HASH_SEQ_STATUS*)palloc0(sizeof(HASH_SEQ_STATUS));
hash_seq_init(status, lockNameHash);
funcctx->user_fctx = (void*)status;
MemoryContextSwitchTo(oldcontext);
}
funcctx = SRF_PERCALL_SETUP();
LockName *lockName = NULL;
status = (HASH_SEQ_STATUS*)(funcctx->user_fctx);
char dataBaseId[DATABASE_ID_LENTH+1] = {0};
errno_t rc = snprintf_s((char*)dataBaseId, DATABASE_ID_LENTH+1, DATABASE_ID_LENTH, "%u", u_sess->proc_cxt.MyDatabaseId);
securec_check_ss(rc, "\0", "\0");
while ((lockName = (LockName *)hash_seq_search(status)) != NULL) {
if (pg_strncasecmp(dataBaseId, lockName->name, strlen((char*)dataBaseId)) != 0)
continue;
Datum values[2];
bool nulls[2];
HeapTuple tuple;
Datum result;
* Form tuple with appropriate data.
*/
errno_t rc = memset_s(values, sizeof(values), 0, sizeof(values));
securec_check(rc, "\0", "\0");
rc = memset_s(nulls, sizeof(nulls), false, sizeof(nulls));
securec_check(rc, "\0", "\0");
values[0] = CStringGetTextDatum((char*)(lockName->name)+strlen((char*)dataBaseId));
fcinfo->arg[0] = values[0];
Datum datum1 = IsUsedAdvisoryLock(fcinfo);
if (datum1 == 0) {
fcinfo->isnull = false;
AutoMutexLock nameHashLock(&gNameHashLock);
nameHashLock.lock();
hash_search(lockNameHash, (void *)&(lockName->name), HASH_REMOVE, NULL);
nameHashLock.unLock();
continue;
}
values[1] = datum1;
tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
result = HeapTupleGetDatum(tuple);
SRF_RETURN_NEXT(funcctx, result);
}
SRF_RETURN_DONE(funcctx);
}
Datum ClearInvalidLockName(PG_FUNCTION_ARGS)
{
return ClearInvalidLockNameInner();
}
typedef struct relidcacheent {
Oid reloid;
Relation reldesc;
} RelIdCacheEnt;
typedef struct {
PG_Lock_Status* lockStatus;
char* sqlMode;
} Kept_State_Data;
PG_FUNCTION_INFO_V1_PUBLIC(pg_open_tables);
extern "C" DLL_PUBLIC Datum pg_open_tables(PG_FUNCTION_ARGS);
#define OPEN_TABLES_COL_NUM 6
typedef enum {
OID_COL = 0,
RELNAME_COL,
RELNAMESPACE_COL,
REFCNT_COL,
LOCKCNT_COL,
ACCESSEXCLUSIVE_LOCKCNT_COL
} OpenTablesColIdxs;
void getLockCnt(PG_Lock_Status* mystatus, Oid rid, Datum* lockCnt, Datum* accessExclusive_LockCnt)
{
*lockCnt = 0;
*accessExclusive_LockCnt = 0;
while (mystatus->currIdx < mystatus->lockData->nelements) {
bool granted = false;
LOCKMODE mode = 0;
LockInstanceData *instance = NULL;
instance = &(mystatus->lockData->locks[mystatus->currIdx]);
if (instance->locktag.locktag_field2 != rid) {
mystatus->currIdx++;
continue;
}
if (instance->holdMask) {
for (mode = 0; mode < MAX_LOCKMODES; mode++) {
if (instance->holdMask & LOCKBIT_ON(mode)) {
granted = true;
(*lockCnt)++;
instance->holdMask &= LOCKBIT_OFF(mode);
break;
}
}
if (mode == AccessExclusiveLock) {
(*accessExclusive_LockCnt)++;
}
}
* If no (more) held modes to report, see if PROC is waiting for a
* lock on this lock.
*/
if (!granted) {
if (instance->waitLockMode != NoLock) {
mode = instance->waitLockMode;
(*lockCnt)++;
* We are now done with this PROCLOCK, so advance pointer to
* continue with next one on next call.
*/
mystatus->currIdx++;
} else {
* Okay, we've displayed all the locks associated with this
* PROCLOCK, proceed to the next one.
*/
mystatus->currIdx++;
}
}
}
}
Datum pg_open_tables(PG_FUNCTION_ARGS)
{
char *schemaname = NULL;
if (PG_NARGS() == 1 && !PG_ARGISNULL(0))
schemaname = text_to_cstring(PG_GETARG_TEXT_P(0));
TupleDesc tupdesc = NULL;
FuncCallContext* funcctx = NULL;
Kept_State_Data* kept = NULL;
PG_Lock_Status* mystatus = NULL;
if (SRF_IS_FIRSTCALL()) {
MemoryContext oldcontext;
funcctx = SRF_FIRSTCALL_INIT();
* switch to memory context appropriate for multiple function calls
*/
oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
tupdesc = CreateTemplateTupleDesc(OPEN_TABLES_COL_NUM, false);
TupleDescInitEntry(tupdesc, (AttrNumber)OID_COL + 1, "oid", OIDOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)RELNAME_COL + 1, "relname", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)RELNAMESPACE_COL + 1, "relnamespace", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)REFCNT_COL + 1, "refcnt", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)LOCKCNT_COL + 1, "lockcnt", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ACCESSEXCLUSIVE_LOCKCNT_COL + 1,
"accessexclusive_lockcnt", INT4OID, -1, 0);
funcctx->tuple_desc = BlessTupleDesc(tupdesc);
kept = (Kept_State_Data*)palloc(sizeof(Kept_State_Data));
funcctx->user_fctx = (void*)kept;
* Collect all the locking information that we will format and send
* out as a result set.
*/
mystatus = (PG_Lock_Status*)palloc(sizeof(PG_Lock_Status));
kept->lockStatus = mystatus;
mystatus->lockData = GetLockStatusData();
mystatus->currIdx = 0;
mystatus->predLockData = NULL;
mystatus->predLockIdx = 0;
SPI_STACK_LOG("connect", NULL, NULL);
if (SPI_connect() < 0) {
ereport(ERROR,
(errcode(ERRCODE_CONNECTION_EXCEPTION),
errmsg("internal error while showing open tables: SPI_connect failed")));
}
kept->sqlMode = GetSqlMode();
SetSqlMode("ansi_quotes");
PG_TRY();
{
if (schemaname != NULL) {
char query[CHAR_BUF_SIZE] = {0};
int rc = snprintf_s(query, CHAR_BUF_SIZE, CHAR_BUF_SIZE - 1,
"select c.oid from pg_catalog.pg_class c, pg_catalog.pg_namespace n"
" where c.relnamespace = n.oid"
" and c.relkind in ('r', 'v')"
" and c.relpersistence not in ('t', 'g')"
" and n.nspname = '%s';", schemaname);
securec_check_ss(rc, "", "");
if (SPI_OK_SELECT != SPI_execute(query, true, 0)) {
ereport(ERROR, (errcode(ERRCODE_DATA_EXCEPTION),
errmsg("internal error while showing open tables: SPI_execute failed")));
}
} else {
if (SPI_OK_SELECT != SPI_execute("select oid from pg_catalog.pg_class"
" where relkind in ('r', 'v') and relpersistence not in ('t', 'g');", true, 0)) {
ereport(ERROR, (errcode(ERRCODE_DATA_EXCEPTION),
errmsg("internal error while showing open tables: SPI_execute failed")));
}
}
}
PG_CATCH();
{
SetSqlMode((const char*)kept->sqlMode);
PG_RE_THROW();
}
PG_END_TRY();
funcctx->max_calls = SPI_processed;
funcctx->call_cntr = 0;
MemoryContextSwitchTo(oldcontext);
}
funcctx = SRF_PERCALL_SETUP();
kept = (Kept_State_Data*)funcctx->user_fctx;
mystatus = kept->lockStatus;
mystatus->currIdx = 0;
bool setSqlMode = true;
HeapTuple tuple = NULL;
Relation rel = NULL;
PG_TRY();
{
if (funcctx->call_cntr < funcctx->max_calls) {
Datum values[6];
bool nulls[6];
* Form tuple with appropriate data.
*/
errno_t rc = memset_s(values, sizeof(values), 0, sizeof(values));
securec_check(rc, "\0", "\0");
rc = memset_s(nulls, sizeof(nulls), false, sizeof(nulls));
securec_check(rc, "\0", "\0");
bool isNull;
Oid rid = SPI_getbinval(SPI_tuptable->vals[funcctx->call_cntr], SPI_tuptable->tupdesc, 1, &isNull);
if (!isNull) {
RelationIdCacheLookup(rid, rel);
if (rel == NULL) {
values[OID_COL] = rid;
nulls[RELNAME_COL] = true;
nulls[RELNAMESPACE_COL] = true;
nulls[REFCNT_COL] = true;
nulls[LOCKCNT_COL] = true;
nulls[ACCESSEXCLUSIVE_LOCKCNT_COL] = true;
} else {
values[OID_COL] = ObjectIdGetDatum(rel->rd_id);
values[RELNAME_COL] = CStringGetTextDatum(RelationGetRelationName(rel));
values[RELNAMESPACE_COL] = CStringGetTextDatum(get_namespace_name(RelationGetNamespace(rel)));
values[REFCNT_COL] = rel->rd_refcnt;
getLockCnt(mystatus, rid, &values[LOCKCNT_COL], &values[ACCESSEXCLUSIVE_LOCKCNT_COL]);
}
tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
t_thrd.log_cxt.PG_exception_stack = save_exception_stack;
t_thrd.log_cxt.error_context_stack = save_context_stack;
gstrace_tryblock_exit(false, oldTryCounter);
SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
}
} else {
setSqlMode = false;
SetSqlMode((const char *)kept->sqlMode);
SPI_STACK_LOG("finish", NULL, NULL);
if (SPI_finish() != SPI_OK_FINISH) {
ereport(ERROR, (errcode(ERRCODE_SPI_FINISH_FAILURE),
errmsg("internal error while showing open tables: SPI_finish failed")));
}
pfree_ext(kept->lockStatus);
pfree_ext(funcctx->user_fctx);
t_thrd.log_cxt.PG_exception_stack = save_exception_stack;
t_thrd.log_cxt.error_context_stack = save_context_stack;
gstrace_tryblock_exit(false, oldTryCounter);
SRF_RETURN_DONE(funcctx);
}
}
PG_CATCH();
{
if (setSqlMode) {
SetSqlMode((const char *)kept->sqlMode);
}
PG_RE_THROW();
}
PG_END_TRY();
return (Datum)0;
}
#endif
#ifdef PGXC
* pgxc_lock_for_backup
*
* Lock the cluster for taking backup
* To lock the cluster, try to acquire a session level advisory lock exclusivly
* By lock we mean to disallow any statements that change
* the portions of the catalog which are backed up by pg_dump/pg_dumpall
* Returns true or fails with an error message.
*/
Datum pgxc_lock_for_backup(PG_FUNCTION_ARGS)
{
bool lockAcquired = false;
int prepared_xact_count = 0;
if (!superuser())
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("only system admin can lock the cluster for backup")));
* The system cannot be locked for backup if there is an uncommitted
* prepared transaction, the reason is as follows:
* Utility statements are divided into two groups, one is allowed group
* and the other is disallowed group. A statement is put in allowed group
* if it does not make changes to the catalog or makes such changes which
* are not backed up by pg_dump or pg_dumpall, otherwise it is put in
* disallowed group. Every time a disallowed statement is issued we try to
* hold an advisory lock in shared mode and if the lock can be acquired
* only then the statement is allowed.
* In case of prepared transactions suppose the lock is not released at
* prepare transaction 'txn_id'
* Consider the following scenario:
*
* begin;
* create table abc_def(a int, b int);
* insert into abc_def values(1,2),(3,4);
* prepare transaction 'abc';
*
* Now assume that the server is restarted for any reason.
* When prepared transactions are saved on disk, session level locks are
* ignored and hence when the prepared transactions are reterieved and all
* the other locks are reclaimed, but session level advisory locks are
* not reclaimed.
* Hence we made the following decisions
* a) Transaction level advisory locks should be used for DDLs which are
* automatically released at prepare transaction 'txn_id'
* b) If there is any uncomitted prepared transaction, it is assumed
* that it must be issuing a statement that belongs to disallowed
* group and hence the request to hold the advisory lock exclusively
* is denied.
*/
SPI_STACK_LOG("connect", NULL, NULL);
if (SPI_connect() < 0) {
ereport(ERROR,
(errcode(ERRCODE_CONNECTION_EXCEPTION), errmsg("internal error while locking the cluster for backup")));
}
SPI_execute("select gid from pg_catalog.pg_prepared_xacts limit 1", true, 0);
prepared_xact_count = SPI_processed;
SPI_STACK_LOG("finish", NULL, NULL);
SPI_finish();
if (prepared_xact_count > 0) {
ereport(ERROR,
(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
errmsg("cannot lock cluster for backup in presence of %d uncommitted prepared transactions",
prepared_xact_count)));
}
lockAcquired = DatumGetBool(DirectFunctionCall2(pg_try_advisory_lock_int4,
t_thrd.postmaster_cxt.xc_lockForBackupKey1,
t_thrd.postmaster_cxt.xc_lockForBackupKey2));
if (!lockAcquired)
ereport(ERROR,
(errcode(ERRCODE_LOCK_NOT_AVAILABLE), errmsg("cannot lock cluster for backup, lock is already held")));
* sessin level advisory locks stay for only as long as the session
* that issues them does
*/
elog(INFO, "please do not close this session until you are done adding the new node");
PG_RETURN_BOOL(lockAcquired);
}
* pgxc_unlock_for_sp_database (Name)
* Unlock the specific database in Exclusive mode for specific database.
*/
Datum pgxc_unlock_for_sp_database(PG_FUNCTION_ARGS)
{
Name databaseName = PG_GETARG_NAME(0);
bool result = false;
result = DatumGetBool(DirectFunctionCall3(pg_advisory_unlock_sp_db_int4,
t_thrd.postmaster_cxt.xc_lockForBackupKey1,
t_thrd.postmaster_cxt.xc_lockForBackupKey2,
NameGetDatum(databaseName)));
PG_RETURN_BOOL(result);
}
* pgxc_lock_for_sp_database (Name)
*
* Lock the specific database in Exclusive mode.
* To lock the cluster, try to acquire a session level advisory lock exclusivly
* By lock we mean to disallow any DDL operation.
* Returns true or fails with an error message.
* If all objects in DDL is temp, then we only need to get lock of current coordinator, because
* temp objects' defination will not be sent to remote coordinators.
* if could not get the lock after the first try, pgxc_lock_for_sp_database will go into WaitOnLock until
* try to get the lock again or lock wait timeout.
*/
Datum pgxc_lock_for_sp_database(PG_FUNCTION_ARGS)
{
int prepared_xact_count;
Name databaseName = PG_GETARG_NAME(0);
if (!superuser())
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("only system admin can lock the cluster for backup")));
* The system cannot be locked for backup if there is an uncommitted
* prepared transaction, the reason is as follows:
* Utility statements are divided into two groups, one is allowed group
* and the other is disallowed group. A statement is put in allowed group
* if it does not make changes to the catalog or makes such changes which
* are not backed up by pg_dump or pg_dumpall, otherwise it is put in
* disallowed group. Every time a disallowed statement is issued we try to
* hold an advisory lock in shared mode and if the lock can be acquired
* only then the statement is allowed.
* In case of prepared transactions suppose the lock is not released at
* prepare transaction 'txn_id'
* Consider the following scenario:
*
* begin;
* create table abc_def(a int, b int);
* insert into abc_def values(1,2),(3,4);
* prepare transaction 'abc';
*
* Now assume that the server is restarted for any reason.
* When prepared transactions are saved on disk, session level locks are
* ignored and hence when the prepared transactions are reterieved and all
* the other locks are reclaimed, but session level advisory locks are
* not reclaimed.
* Hence we made the following decisions
* a) Transaction level advisory locks should be used for DDLs which are
* automatically released at prepare transaction 'txn_id'
* b) If there is any uncomitted prepared transaction, it is assumed
* that it must be issuing a statement that belongs to disallowed
* group and hence the request to hold the advisory lock exclusively
* is denied.
*/
SPI_STACK_LOG("connect", NULL, NULL);
if (SPI_connect() < 0) {
ereport(ERROR,
(errcode(ERRCODE_CONNECTION_EXCEPTION), errmsg("internal error while locking the cluster for backup")));
}
SPI_execute("select gid from pg_catalog.pg_prepared_xacts limit 1", true, 0);
prepared_xact_count = SPI_processed;
SPI_STACK_LOG("finish", NULL, NULL);
SPI_finish();
if (prepared_xact_count > 0) {
ereport(ERROR,
(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
errmsg("cannot lock cluster for backup in presence of %d uncommitted prepared transactions",
prepared_xact_count)));
}
DirectFunctionCall3(pg_advisory_lock_sp_db_int4,
t_thrd.postmaster_cxt.xc_lockForBackupKey1,
t_thrd.postmaster_cxt.xc_lockForBackupKey2,
NameGetDatum(databaseName));
PG_RETURN_BOOL(true);
}
bool pg_try_advisory_lock_for_redis(Relation rel)
{
LOCKMODE lockmode = u_sess->attr.attr_sql.enable_cluster_resize ? ExclusiveLock : ShareLock;
LockLevel locklevel = u_sess->attr.attr_sql.enable_cluster_resize ? SESSION_LOCK : TRANSACTION_LOCK;
TryType locktry = u_sess->attr.attr_sql.enable_cluster_resize ? WAIT : DONT_WAIT;
bool result = pgxc_advisory_lock(0, 65534, RelationGetRelCnOid(rel), false, lockmode, locklevel, locktry, NULL);
if (u_sess->attr.attr_sql.enable_cluster_resize && result) {
return true;
} else if (result) {
LOCKTAG tag;
SET_LOCKTAG_INT32_DB(tag, u_sess->proc_cxt.MyDatabaseId, 65534, RelationGetRelCnOid(rel));
(void)LockRelease(&tag, ShareLock, false);
return true;
}
return false;
}
* pgxc_lock_for_backup
*
* Lock the cluster for taking backup
* To lock the cluster, try to acquire a session level advisory lock exclusivly
* By lock we mean to disallow any statements that change
* the portions of the catalog which are backed up by pg_dump/pg_dumpall
* Returns true or fails with an error message.
* If all objects in DDL is temp, then we only need to get lock of current coordinator, because
* temp objects' defination will not be sent to remote coordinators.
*/
void pgxc_lock_for_utility_stmt(Node* parsetree, bool is_temp)
{
bool lockAcquired = false;
LOCKTAG tag;
LockAcquireResult res;
* Reload configuration if we got SIGHUP from the postmaster, since we want to fetch
* latest enable_online_ddl_waitlock values.
*/
reload_configfile();
* Change donwait lock to wait lock for online expansion condition to prevent
* disruption of business DDL.
*/
if (!u_sess->attr.attr_sql.enable_online_ddl_waitlock) {
* Temp table. For temp table, no need to lock other coordinator, because
* it's defination is only on this coordinator.
*/
if (!is_temp) {
lockAcquired = DatumGetBool(DirectFunctionCall2(pg_try_advisory_xact_lock_shared_int4,
t_thrd.postmaster_cxt.xc_lockForBackupKey1,
t_thrd.postmaster_cxt.xc_lockForBackupKey2));
} else {
SET_LOCKTAG_INT32(
tag, t_thrd.postmaster_cxt.xc_lockForBackupKey1, t_thrd.postmaster_cxt.xc_lockForBackupKey2);
res = LockAcquire(&tag, ShareLock, false, true);
lockAcquired = (res != LOCKACQUIRE_NOT_AVAIL);
}
if (!lockAcquired)
ereport(ERROR,
(errcode(ERRCODE_READ_ONLY_SQL_TRANSACTION),
errmsg(
"cannot execute %s in a locked cluster", parsetree ? CreateCommandTag(parsetree) : "VACUUM")));
} else {
* We use pg_advisory_lock to ensure metadata synchronization in online expansion, so
* we need used wait lock for DDL so that DDL can wait for pg_advisory_lock but not
* exit directly when cluster was locked. and there is no need to check the return
* value in wait lock condition.
*/
if (!is_temp) {
DirectFunctionCall2(pg_advisory_xact_lock_shared_int4,
t_thrd.postmaster_cxt.xc_lockForBackupKey1,
t_thrd.postmaster_cxt.xc_lockForBackupKey2);
} else {
* When in online node replace we don't need be blocked by cluster lock,
* and we don't need call LockAcquire here.
*/
if (OM_ONLINE_EXPANSION == get_om_online_state()) {
* Don't support temp table in online scenarios for reasons:
* 1.temp table only created in one coordinator, and can't be blocked if cluster
* lock is on other coordinators.
* 2.temp table's defination is on only one coordinator which cause metadata of
* temp table on difference coordinators will loss when use random one cn build
* dn to restore metadata.
* 3.we won't do data redistribute of temp table through gs_redis which will cause
* data loss of temp table after online expansion.
*/
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("temp table is not supported in online expansion")));
}
}
* Set current_installation_nodegroup to null here for
* PgxcGroupGetInstallationGroup to get the newest one.
*/
CleanNodeGroupStatus();
* We should manually reload pooler message here for online expansion.
* Notice : Reload pooler in transaction block is not supported, but we
* we need do it for single-query scene.
*/
reload_online_pooler();
}
}
Datum pgxc_lock_for_transfer(PG_FUNCTION_ARGS)
{
Name schemaName = PG_GETARG_NAME(0);
bool isLock = false;
if (!superuser() || IS_PGXC_DATANODE) {
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("Only system admin can use the function on coordinator")));
}
if (IsInitdb || u_sess->attr.attr_common.IsInplaceUpgrade) {
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("Can not run the function during initdb or upgrade")));
}
char *databaseName = get_and_check_db_name(u_sess->proc_cxt.MyDatabaseId, true);
if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
CheckIfAnySchemaInRedistribution();
DirectFunctionCall1(pgxc_lock_for_sp_database, CStringGetDatum(databaseName));
}
UpdateSchemaInRedistribution(schemaName, isLock);
if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
PGXCSendTransfer(schemaName, isLock);
DirectFunctionCall1(pgxc_unlock_for_sp_database, CStringGetDatum(databaseName));
}
PG_RETURN_BOOL(true);
}
Datum pgxc_unlock_for_transfer(PG_FUNCTION_ARGS)
{
Name schemaName = PG_GETARG_NAME(0);
bool isLock = true;
if (!superuser() || IS_PGXC_DATANODE) {
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("Only system admin can use the function on coordinator")));
}
if (IsInitdb || u_sess->attr.attr_common.IsInplaceUpgrade) {
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("Can not run the function during initdb or upgrade")));
}
UpdateSchemaInRedistribution(schemaName, isLock);
if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
PGXCSendTransfer(schemaName, isLock);
}
PG_RETURN_BOOL(true);
}
#endif
