PostgreSQL Database Management System

pPostgreSQL Database Management System

=====================================

This directory contains the source code distribution of the PostgreSQL

intid;

}BgwPoolExecutorCtx;

size_tn_snapshots;

size_tn_active;

staticvoidBgwPoolMainLoop(Datumarg)

{

BgwPoolExecutorCtx*ctx= (BgwPoolExecutorCtx*)arg;

size=*(int*)&pool->queue[pool->head];

Assert(size<pool->size);

work=malloc(size);

pool->active-=1;

pool->pending-=1;

pool->active+=1;

if (pool->head+size+4>pool->size) {

memcpy(work,pool->queue,size);

pool->head=INTALIGN(size);

SpinLockRelease(&pool->lock);

pool->executor(id,work,size);

free(work);

SpinLockAcquire(&pool->lock);

pool->active-=1;

SpinLockRelease(&pool->lock);

}

}

pool->tail=0;

pool->size=queueSize;

pool->active=0;

pool->pending=0;

strcpy(pool->dbname,dbname);

}

PGSemaphoreLock(&pool->overflow);

SpinLockAcquire(&pool->lock);

}else {

pool->active+=1;

n_snapshots+=1;

n_active+=pool->active;

pool->pending+=1;

*(int*)&pool->queue[pool->tail]=size;

if (pool->size-pool->tail >=size+4) {

memcpy(&pool->queue[pool->tail+4],work,size);

size_ttail;

size_tsize;

size_tactive;

size_tpending;

boolproducerBlocked;

chardbname[MAX_DBNAME_LEN];

char*queue;

LANGUAGE C;

CREATETYPEmtm.node_stateAS (idinteger, disabled bool, disconnected bool, catchUp bool, slotLagbigint, avgTransDelaybigint, lastStatusChangetimestamp, connStrtext);

CREATETYPEmtm.node_stateAS ("id"integer,"disabled" bool,"disconnected" bool,"catchUp" bool,"slotLag"bigint,"avgTransDelay"bigint,"lastStatusChange"timestamp,"connStr"text);

CREATEFUNCTIONmtm.get_nodes_state() RETURNS SETOFmtm.node_state

AS'MODULE_PATHNAME','mtm_get_nodes_state'

LANGUAGE C;

CREATETYPEmtm.cluster_stateAS (statustext, disabledNodeMaskbigint, disconnectedNodeMaskbigint, catchUpNodeMaskbigint, nNodesinteger, nActiveQueriesinteger, queueSizebigint, transCountbigint, timeShiftbigint, recoverySlotinteger);

CREATETYPEmtm.cluster_stateAS ("status"text,"disabledNodeMask"bigint,"disconnectedNodeMask"bigint,"catchUpNodeMask"bigint,"nNodes"integer,"nActiveQueries"integer,"nPendingQueries"integer,"queueSize"bigint,"transCount"bigint,"timeShift"bigint,"recoverySlot"integer);

CREATEFUNCTIONmtm.get_cluster_state() RETURNSmtm.cluster_state

AS'MODULE_PATHNAME','mtm_get_cluster_state'

LANGUAGE C;

CREATEFUNCTIONmtm.get_cluster_info() RETURNS SETOFmtm.cluster_state

AS'MODULE_PATHNAME','mtm_get_cluster_info'

LANGUAGE C;

CREATEFUNCTIONmtm.make_table_local(relation regclass) RETURNS void

AS'MODULE_PATHNAME','mtm_make_table_local'

LANGUAGE C;

PG_FUNCTION_INFO_V1(mtm_get_snapshot);

PG_FUNCTION_INFO_V1(mtm_get_nodes_state);

PG_FUNCTION_INFO_V1(mtm_get_cluster_state);

PG_FUNCTION_INFO_V1(mtm_get_cluster_info);

PG_FUNCTION_INFO_V1(mtm_make_table_local);

PG_FUNCTION_INFO_V1(mtm_dump_lock_graph);

"Connected",

"Online",

"Recovery",

"InMinor",

};

boolMtmDoReplication;

voidMtmHandleApplyError(void)

{

ErrorData*edata=CopyErrorData();

switch (edata->sqlerrcode) {

caseERRCODE_DISK_FULL:

caseERRCODE_INSUFFICIENT_RESOURCES:

caseERRCODE_IO_ERROR:

caseERRCODE_DATA_CORRUPTED:

caseERRCODE_INDEX_CORRUPTED:

caseERRCODE_SYSTEM_ERROR:

caseERRCODE_INTERNAL_ERROR:

caseERRCODE_OUT_OF_MEMORY:

elog(WARNING,"Node is excluded from cluster because of non-recoverable error %d",edata->sqlerrcode);

MtmSwitchClusterMode(MTM_OUT_OF_SERVICE);

kill(PostmasterPid,SIGQUIT);

break;

}

}

voidMtmRecoveryCompleted(void)

{

MTM_LOG1("Recovery of node %d is completed",MtmNodeId);

"Minamal amount of time (milliseconds) to wait 2PC confirmation from all nodes",

"Timeout for 2PC is calculated as MAX(prepare_time*2pc_prepare_ratio/100,2pc_min_timeout)",

&Mtm2PCMinTimeout,

0,

INT_MAX,

PGC_BACKEND,

"Percent of prepare time for maximal time of second phase of two-pahse commit",

"Timeout for 2PC is calculated as MAX(prepare_time*2pc_prepare_ratio/100,2pc_min_timeout)",

&Mtm2PCPrepareRatio,

0,

INT_MAX,

PGC_BACKEND,

{

intnodeId;

char*connStrPtr;

TupleDescdesc;

Datumvalues[8];

boolnulls[8];

Datumvalues[Natts_mtm_nodes_state];

boolnulls[Natts_mtm_nodes_state];

}MtmGetNodeStateCtx;

FuncCallContext*funcctx;

MtmGetNodeStateCtx*usrfctx;

MemoryContextoldcontext;

char*p;

int64lag;

boolis_first_call=SRF_IS_FIRSTCALL();

usrfctx= (MtmGetNodeStateCtx*)palloc(sizeof(MtmGetNodeStateCtx));

get_call_result_type(fcinfo,NULL,&usrfctx->desc);

usrfctx->nodeId=1;

usrfctx->connStrPtr=pstrdup(MtmConnStrs);

memset(usrfctx->nulls, false,sizeof(usrfctx->nulls));

funcctx->user_fctx=usrfctx;

MemoryContextSwitchTo(oldcontext);

usrfctx->nulls[4]=lag<0;

usrfctx->values[5]=Int64GetDatum(Mtm->transCount ?Mtm->nodes[usrfctx->nodeId-1].transDelay/Mtm->transCount :0);

usrfctx->values[6]=TimestampTzGetDatum(time_t_to_timestamptz(Mtm->nodes[usrfctx->nodeId-1].lastStatusChangeTime));

p=strchr(usrfctx->connStrPtr,',');

if (p!=NULL) {

*p++='\0';

}

usrfctx->values[7]=CStringGetTextDatum(usrfctx->connStrPtr);

usrfctx->connStrPtr=p;

usrfctx->values[7]=CStringGetTextDatum(Mtm->nodes[usrfctx->nodeId-1].con.connStr);

usrfctx->nodeId+=1;

SRF_RETURN_NEXT(funcctx,HeapTupleGetDatum(heap_form_tuple(usrfctx->desc,usrfctx->values,usrfctx->nulls)));

}

mtm_get_cluster_state(PG_FUNCTION_ARGS)

{

TupleDescdesc;

Datumvalues[10];

boolnulls[10]= {false};

Datumvalues[Natts_mtm_cluster_state];

boolnulls[Natts_mtm_cluster_state]= {false};

get_call_result_type(fcinfo,NULL,&desc);

values[0]=CStringGetTextDatum(MtmNodeStatusMnem[Mtm->status]);

values[3]=Int64GetDatum(Mtm->nodeLockerMask);

values[4]=Int32GetDatum(Mtm->nNodes);

values[5]=Int32GetDatum((int)Mtm->pool.active);

values[6]=Int64GetDatum(BgwPoolGetQueueSize(&Mtm->pool));

values[7]=Int64GetDatum(Mtm->transCount);

values[8]=Int64GetDatum(Mtm->timeShift);

values[9]=Int32GetDatum(Mtm->recoverySlot);

nulls[9]=Mtm->recoverySlot==0;

values[6]=Int32GetDatum((int)Mtm->pool.pending);

values[7]=Int64GetDatum(BgwPoolGetQueueSize(&Mtm->pool));

values[8]=Int64GetDatum(Mtm->transCount);

values[9]=Int64GetDatum(Mtm->timeShift);

values[10]=Int32GetDatum(Mtm->recoverySlot);

PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(desc,values,nulls)));

}

{

intnodeId;

}MtmGetClusterInfoCtx;

mtm_get_cluster_info(PG_FUNCTION_ARGS)

{

FuncCallContext*funcctx;

MtmGetClusterInfoCtx*usrfctx;

MemoryContextoldcontext;

TupleDescdesc;

boolis_first_call=SRF_IS_FIRSTCALL();

inti;

PGconn*conn;

PGresult*result;

char*values[Natts_mtm_cluster_state];

HeapTupletuple;

if (is_first_call) {

funcctx=SRF_FIRSTCALL_INIT();

oldcontext=MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);

usrfctx= (MtmGetClusterInfoCtx*)palloc(sizeof(MtmGetNodeStateCtx));

get_call_result_type(fcinfo,NULL,&desc);

funcctx->attinmeta=TupleDescGetAttInMetadata(desc);

usrfctx->nodeId=1;

funcctx->user_fctx=usrfctx;

MemoryContextSwitchTo(oldcontext);

}

funcctx=SRF_PERCALL_SETUP();

usrfctx= (MtmGetClusterInfoCtx*)funcctx->user_fctx;

if (usrfctx->nodeId>MtmNodes) {

SRF_RETURN_DONE(funcctx);

}

conn=PQconnectdb(Mtm->nodes[usrfctx->nodeId-1].con.connStr);

if (PQstatus(conn)!=CONNECTION_OK) {

elog(ERROR,"Failed to establish connection '%s' to node %d",Mtm->nodes[usrfctx->nodeId-1].con.connStr,usrfctx->nodeId);

}

result=PQexec(conn,"select * from mtm.get_cluster_state()");

if (PQresultStatus(result)!=PGRES_TUPLES_OK||PQntuples(result)!=1) {

elog(ERROR,"Failed to receive data from %d",usrfctx->nodeId);

}

for (i=0;i<Natts_mtm_cluster_state;i++) {

values[i]=PQgetvalue(result,0,i);

}

tuple=BuildTupleFromCStrings(funcctx->attinmeta,values);

PQclear(result);

PQfinish(conn);

usrfctx->nodeId+=1;

SRF_RETURN_NEXT(funcctx,HeapTupleGetDatum(tuple));

}

Datummtm_make_table_local(PG_FUNCTION_ARGS)

{

Oidreloid=PG_GETARG_OID(1);

#defineAnum_mtm_local_tables_rel_schema 1

#defineAnum_mtm_local_tables_rel_name 2

#defineNatts_mtm_cluster_state 11

#defineNatts_mtm_nodes_state 8

typedefuint64csn_t;/* commit serial number */

#defineINVALID_CSN ((csn_t)-1)

{

MTM_INITIALIZATION,/* Initial status */

MTM_OFFLINE,/* Node isout of quorum */

MTM_OFFLINE,/* Node isexcluded from cluster */

MTM_CONNECTED,/* Arbiter is established connections with other nodes */

MTM_ONLINE,/* Ready to receive client's queries */

MTM_RECOVERY,/* Node is in recovery process */

MTM_IN_MINORITY,/* Node is out of quorum */

}MtmNodeStatus;

externboolMtmRecoveryCaughtUp(intnodeId,XLogRecPtrslotLSN);

externvoidMtmRecoveryCompleted(void);

externvoidMtmMakeTableLocal(char*schema,char*name);

externvoidMtmHandleApplyError(void);

}

PG_CATCH();

{

MtmHandleApplyError();

EmitErrorReport();

FlushErrorState();

MTM_LOG2("%d: REMOTE begin abort transaction %d",MyProcPid,MtmGetCurrentTransactionId());