#include "gtm/gtm_c.h"
#include "gtm/gtm_lock.h"
#include "gtm/elog.h"
+#include "gtm/gtm.h"
/*
* Acquire the request lock. Block if the lock is not available
break;
}
- return status ? false : true;
+ if (status != 0)
+ return false;
+
+ RWLocksHeld[NumRWLocksHeld++] = lock;
+ return true;
}
/*
bool
GTM_RWLockRelease(GTM_RWLock *lock)
{
- int status;
+ int i;
+ int status;
+
status = pthread_rwlock_unlock(&lock->lk_lock);
#ifdef GTM_LOCK_DEBUG
if (status)
pthread_mutex_unlock(&lock->lk_debug_mutex);
}
#endif
- return status ? false : true;
+ if (status != 0)
+ return false;
+
+ /*
+ * Remove lock from list of locks held. Usually, but not always, it will
+ * be the latest-acquired lock; so search array backwards.
+ */
+ for (i = NumRWLocksHeld; --i >= 0;)
+ if (lock == RWLocksHeld[i])
+ break;
+
+ if (i < 0)
+ elog(ERROR, "lock is not held");
+
+ NumRWLocksHeld--;
+ for (; i < NumRWLocksHeld; i++)
+ RWLocksHeld[i] = RWLocksHeld[i + 1];
+
+ return true;
}
/*
return pthread_rwlock_destroy(&lock->lk_lock);
}
+void
+GTM_RWLockReleaseAll(void)
+{
+ while (NumRWLocksHeld > 0)
+ GTM_RWLockRelease(RWLocksHeld[NumRWLocksHeld - 1]);
+}
+
/*
* Initialize a mutex lock
*/
GTM_MutexLockAcquire(GTM_MutexLock *lock)
{
int status = pthread_mutex_lock(&lock->lk_lock);
- return status ? false : true;
+
+ if (status != 0)
+ return false;
+
+ MutexLocksHeld[NumMutexLocksHeld++] = lock;
+ return true;
}
/*
bool
GTM_MutexLockRelease(GTM_MutexLock *lock)
{
- return pthread_mutex_unlock(&lock->lk_lock);
+ int i;
+ int status = pthread_mutex_unlock(&lock->lk_lock);
+
+ if (status != 0)
+ return false;
+
+ /*
+ * Remove lock from list of locks held. Usually, but not always, it will
+ * be the latest-acquired lock; so search array backwards.
+ */
+ for (i = NumMutexLocksHeld; --i >= 0;)
+ if (lock == MutexLocksHeld[i])
+ break;
+
+ if (i < 0)
+ elog(ERROR, "mutex is not held");
+
+ NumMutexLocksHeld--;
+ for (; i < NumMutexLocksHeld; i++)
+ MutexLocksHeld[i] = MutexLocksHeld[i + 1];
+
+ return true;
+}
+
+void
+GTM_MutexLockReleaseAll(void)
+{
+ while (NumMutexLocksHeld > 0)
+ GTM_MutexLockRelease(MutexLocksHeld[NumMutexLocksHeld - 1]);
}
/*
{
elog(LOG, "Sequence not active");
GTM_RWLockRelease(&curr_seqinfo->gs_lock);
+ GTM_RWLockRelease(&bucket->shb_lock);
return NULL;
}
Assert(curr_seqinfo->gs_ref_count != SEQ_MAX_REFCOUNT);
if (newcount <= seq_maxcount)
{
/* give up */
+ GTM_RWLockRelease(&b->shb_lock);
ereport(ERROR,
(ERANGE,
errmsg("Can not list all the sequences")));
(seqinfo->gs_called ? 't' : 'f'),
seqinfo->gs_state);
- GTM_RWLockRelease(&seqinfo->gs_lock);
+ GTM_RWLockRelease(&seqinfo->gs_lock);
}
GTM_RWLockRelease(&bucket->shb_lock);
}
if (GTM_IsTransSerializable(mygtm_txninfo))
{
if ((mygtm_txninfo->gti_snapshot_set) && (txn_count > 1))
+ {
+ GTM_RWLockRelease(>MTransactions.gt_TransArrayLock);
elog(ERROR, "Grouped snapshot can only include first snapshot in Serializable transaction");
+ }
if (!mygtm_txninfo->gti_snapshot_set)
{
mysnap->sn_xip = (GlobalTransactionId *)
palloc(GTM_MAX_GLOBAL_TRANSACTIONS * sizeof(GlobalTransactionId));
if (mysnap->sn_xip == NULL)
+ {
+ GTM_RWLockRelease(>MTransactions.gt_TransArrayLock);
ereport(ERROR, (ENOMEM, errmsg("out of memory")));
+ }
}
mysnap->sn_xmin = snapshot->sn_xmin;
mysnap->sn_xmax = snapshot->sn_xmax;
GTM_Threads GTMThreadsData;
GTM_Threads *GTMThreads = >MThreadsData;
-#define GTM_MIN_THREADS 32 /* Provision for minimum threads */
-#define GTM_MAX_THREADS 1024 /* Max threads allowed in the GTM */
+/*
+ * Max threads allowed in the GTM. If you change this, consider changing
+ * GTM_MAX_SIMUL_RWLOCKS too.
+ */
+#define GTM_MAX_THREADS 1024
+/* Provision for minimum threads */
+#define GTM_MIN_THREADS 32
#define GTMThreadsFull (GTMThreads->gt_thread_count == GTMThreads->gt_array_size)
/*
GTMThreads->gt_thread_count--;
GTM_RWLockRelease(>MThreads->gt_lock);
- pfree(thrinfo);
+ free(thrinfo);
return 0;
}
int err;
/*
- * We are still running in the context of the main thread. So the
- * allocation below would last as long as the main thread exists or the
- * memory is explicitely freed.
+ * Allocate outside the memory context.
+ *
+ * We also track locks held by various threads in their thread-specific
+ * info. One such lock is also used by the memory manager and it gets
+ * acquired/released when the thrinfo is freed. We don't want the lock
+ * information to be lost during that process.
+ *
+ * The thread-cleanup routine should ensure that this allocation is freed,
+ * avoiding any memory leak.
*/
- thrinfo = (GTM_ThreadInfo *)palloc0(sizeof (GTM_ThreadInfo));
+ thrinfo = (GTM_ThreadInfo *)malloc(sizeof (GTM_ThreadInfo));
+ memset(thrinfo, 0, sizeof (GTM_ThreadInfo));
thrinfo->thr_conn = conninfo;
GTM_RWLockInit(&thrinfo->thr_lock);
if (GTM_ThreadAdd(thrinfo) == -1)
{
GTM_RWLockDestroy(&thrinfo->thr_lock);
- pfree(thrinfo);
+ free(thrinfo);
return NULL;
}
}
}
+ /* Release any currently held mutex and rwlocks */
+ GTM_MutexLockReleaseAll();
+ GTM_RWLockReleaseAll();
+
/*
* Close a connection to GTM standby.
*/
thrinfo->is_main_thread = true;
GTM_RWLockInit(&thrinfo->thr_lock);
- GTM_RWLockAcquire(&thrinfo->thr_lock, GTM_LOCKMODE_WRITE);
if (SetMyThreadInfo(thrinfo))
{
exit(1);
}
+ /* Must be done after thread-info is set */
+ GTM_RWLockAcquire(&thrinfo->thr_lock, GTM_LOCKMODE_WRITE);
+
TopMostThreadID = pthread_self();
return thrinfo;
void *
GTM_ThreadMain(void *argp)
{
- GTM_ThreadInfo *thrinfo = (GTM_ThreadInfo *)argp;
- int qtype;
- StringInfoData input_message;
- sigjmp_buf local_sigjmp_buf;
+ GTM_ThreadInfo *thrinfo = (GTM_ThreadInfo *)argp;
+ int qtype;
+ StringInfoData input_message;
+ sigjmp_buf local_sigjmp_buf;
+ char startup_type;
+ GTM_StartupPacket sp;
+ StringInfoData inBuf;
+ StringInfoData buf;
elog(DEBUG3, "Starting the connection helper thread");
-
/*
* Create the memory context we will use in the main loop.
*
*/
GTM_RWLockAcquire(&thrinfo->thr_lock, GTM_LOCKMODE_WRITE);
- {
- /*
- * We expect a startup message at the very start. The message type is
- * REGISTER_COORD, followed by the 4 byte Coordinator ID
- */
- char startup_type;
- GTM_StartupPacket sp;
- StringInfoData inBuf;
+ /*
+ * We expect a startup message at the very start. The message type is
+ * REGISTER_COORD, followed by the 4 byte Coordinator ID
+ */
- startup_type = pq_getbyte(thrinfo->thr_conn->con_port);
+ startup_type = pq_getbyte(thrinfo->thr_conn->con_port);
- if (startup_type != 'A')
- ereport(ERROR,
- (EPROTO,
- errmsg("Expecting a startup message, but received %c",
- startup_type)));
+ if (startup_type != 'A')
+ {
+ ereport(ERROR,
+ (EPROTO,
+ errmsg("Expecting a startup message, but received %c",
+ startup_type)));
+ }
- initStringInfo(&inBuf);
+ initStringInfo(&inBuf);
- /*
- * All frontend messages have a length word next
- * after the type code; we can read the message contents independently of
- * the type.
- */
- if (pq_getmessage(thrinfo->thr_conn->con_port, &inBuf, 0))
- ereport(ERROR,
- (EPROTO,
- errmsg("Expecting coordinator ID, but received EOF")));
+ /*
+ * All frontend messages have a length word next
+ * after the type code; we can read the message contents independently of
+ * the type.
+ */
+ if (pq_getmessage(thrinfo->thr_conn->con_port, &inBuf, 0))
+ ereport(ERROR,
+ (EPROTO,
+ errmsg("Expecting coordinator ID, but received EOF")));
- memcpy(&sp,
- pq_getmsgbytes(&inBuf, sizeof (GTM_StartupPacket)),
- sizeof (GTM_StartupPacket));
- pq_getmsgend(&inBuf);
+ memcpy(&sp,
+ pq_getmsgbytes(&inBuf, sizeof (GTM_StartupPacket)),
+ sizeof (GTM_StartupPacket));
+ pq_getmsgend(&inBuf);
- GTM_RegisterPGXCNode(thrinfo->thr_conn->con_port, sp.sp_node_name);
+ GTM_RegisterPGXCNode(thrinfo->thr_conn->con_port, sp.sp_node_name);
- thrinfo->thr_conn->con_port->remote_type = sp.sp_remotetype;
- thrinfo->thr_conn->con_port->is_postmaster = sp.sp_ispostmaster;
+ thrinfo->thr_conn->con_port->remote_type = sp.sp_remotetype;
+ thrinfo->thr_conn->con_port->is_postmaster = sp.sp_ispostmaster;
- /*
- * If the client has resent the identifier assigned to it previously
- * (by GTM master), use that identifier.
- *
- * We only accept identifiers which are lesser or equal to the last
- * identifier we had seen when we were promoted. All other identifiers
- * will be overwritten by what we have assigned
- */
- if ((sp.sp_client_id != 0) &&
- (sp.sp_client_id <= GTMThreads->gt_starting_client_id))
- {
- thrinfo->thr_client_id = sp.sp_client_id;
- }
+ /*
+ * If the client has resent the identifier assigned to it previously
+ * (by GTM master), use that identifier.
+ *
+ * We only accept identifiers which are lesser or equal to the last
+ * identifier we had seen when we were promoted. All other identifiers
+ * will be overwritten by what we have assigned.
+ */
+ if ((sp.sp_client_id != 0) &&
+ (sp.sp_client_id <= GTMThreads->gt_starting_client_id))
+ {
+ thrinfo->thr_client_id = sp.sp_client_id;
}
- {
- /*
- * Send a dummy authentication request message 'R' as the client
- * expects that in the current protocol. Also send the client
- * identifier issued by us (or sent by the client in the startup packet
- * if we concluded to use the same)
- */
- StringInfoData buf;
- pq_beginmessage(&buf, 'R');
- pq_sendint(&buf, thrinfo->thr_client_id, 4);
- pq_endmessage(thrinfo->thr_conn->con_port, &buf);
- pq_flush(thrinfo->thr_conn->con_port);
+ /*
+ * Send a dummy authentication request message 'R' as the client
+ * expects that in the current protocol. Also send the client
+ * identifier issued by us (or sent by the client in the startup packet
+ * if we concluded to use the same)
+ */
+ pq_beginmessage(&buf, 'R');
+ pq_sendint(&buf, thrinfo->thr_client_id, 4);
+ pq_endmessage(thrinfo->thr_conn->con_port, &buf);
+ pq_flush(thrinfo->thr_conn->con_port);
- elog(DEBUG3, "Sent connection authentication message to the client");
- }
+ elog(DEBUG3, "Sent connection authentication message to the client");
/*
* Get the input_message in the TopMemoryContext so that we don't need to
if (sigsetjmp(local_sigjmp_buf, 1) != 0)
{
- /*
- * NOTE: if you are tempted to add more code in this if-block,
- * consider the high probability that it should be in
- * AbortTransaction() instead. The only stuff done directly here
- * should be stuff that is guaranteed to apply *only* for outer-level
- * error recovery, such as adjusting the FE/BE protocol status.
- */
+ /* Release all mutex and rwlocks */
+ GTM_MutexLockReleaseAll();
+ GTM_RWLockReleaseAll();
+
+ /* We're expected to hold this lock */
+ GTM_RWLockAcquire(&thrinfo->thr_lock, GTM_LOCKMODE_WRITE);
/* Report the error to the client and/or server log */
if (thrinfo->thr_conn)
oldContext = MemoryContextSwitchTo(TopMemoryContext);
/* Acquire global locks to copy resource data to the standby. */
+
+ /*
+ * XXX Weird locking semantics.. the locks are released in
+ * ProcessEndReplicationInitialSyncRequest()
+ */
GTM_RWLockAcquire(>MTransactions.gt_XidGenLock, GTM_LOCKMODE_WRITE);
GTM_RWLockAcquire(>MTransactions.gt_TransArrayLock, GTM_LOCKMODE_WRITE);
elog(LOG, "Prepared for copying data with holding XidGenLock and TransArrayLock.");
if (sigsetjmp(local_sigjmp_buf, 1) != 0)
{
+ /* Release all mutex and rwlocks */
+ GTM_MutexLockReleaseAll();
+ GTM_RWLockReleaseAll();
+
/*
* NOTE: if you are tempted to add more code in this if-block,
* consider the high probability that it should be in
*/
if (!first_turn)
{
+ GTMProxy_ConnectionInfo *auth_required[GTM_PROXY_MAX_CONNECTIONS];
+ int auth_required_count = 0;
+
/*
* Check if there are any changes to the connection array assigned to
* this thread. If so, we need to rebuild the fd array.
}
else
{
+ /* Release all mutex and rwlocks */
+ GTM_MutexLockReleaseAll();
+ GTM_RWLockReleaseAll();
+
+ /*
+ * and re-acquire the only mutex lock we are supposed
+ * to hold after the long jump.
+ */
+ GTM_MutexLockAcquire(&thrinfo->thr_lock);
+
/* SIGUSR2 here */
workerThreadReconnectToGTM();
}
* If this is a newly added connection, complete the handshake
*/
if (!conninfo->con_authenticated)
- GTMProxy_HandshakeConnection(conninfo);
+ {
+ auth_required[auth_required_count++] = conninfo;
+ }
thrinfo->thr_poll_fds[ii].fd = conninfo->con_port->sock;
thrinfo->thr_poll_fds[ii].events = POLLIN;
}
GTM_MutexLockRelease(&thrinfo->thr_lock);
+ for (ii = 0; ii < auth_required_count; ii++)
+ {
+ /*
+ * Complete handshake with the remote node. In case we
+ * don't get the expected handshake message, the
+ * connection is dropped.
+ */
+ GTMProxy_HandshakeConnection(auth_required[ii]);
+ }
+
while (true)
{
Enable_Longjmp();
startup_type = pq_getbyte(conninfo->con_port);
if (startup_type != 'A')
+ {
+ conninfo->con_disconnected = true;
+ if (conninfo->con_port->sock > 0)
+ StreamClose(conninfo->con_port->sock);
ereport(ERROR,
(EPROTO,
errmsg("Expecting a startup message, but received %c",
startup_type)));
+ }
initStringInfo(&inBuf);
* the type.
*/
if (pq_getmessage(conninfo->con_port, &inBuf, 0))
+ {
+ conninfo->con_disconnected = true;
+ if (conninfo->con_port->sock > 0)
+ StreamClose(conninfo->con_port->sock);
ereport(ERROR,
(EPROTO,
errmsg("Expecting PGXC Node ID, but received EOF")));
+ }
memcpy(&sp,
pq_getmsgbytes(&inBuf, sizeof (GTM_StartupPacket)),
* allocation
*/
if (GTMProxyThreads->gt_array_size == GTM_PROXY_MAX_THREADS)
+ {
+ GTM_RWLockRelease(>MProxyThreads->gt_lock);
elog(ERROR, "Too many threads active");
+ }
if (GTMProxyThreads->gt_array_size == 0)
newsize = GTM_PROXY_MIN_THREADS;
GTMProxy_ThreadInfo *thrinfo;
int err;
- /*
- * We are still running in the context of the main thread. So the
- * allocation below would last as long as the main thread exists or the
- * memory is explicitely freed.
- */
- thrinfo = (GTMProxy_ThreadInfo *)palloc0(sizeof (GTMProxy_ThreadInfo));
+ thrinfo = (GTMProxy_ThreadInfo *)malloc(sizeof (GTMProxy_ThreadInfo));
+ memset(thrinfo, 0, sizeof (GTMProxy_ThreadInfo));
GTM_MutexLockInit(&thrinfo->thr_lock);
GTM_CVInit(&thrinfo->thr_cv);
elog(DEBUG1, "Cleaning up thread state");
+ /* Release any currently held mutex and rwlocks */
+ GTM_MutexLockReleaseAll();
+ GTM_RWLockReleaseAll();
+
/*
* TODO Close the open connection.
*/
#define recursion_depth (GetMyThreadInfo->thr_error_recursion_depth)
#define errordata_stack_depth (GetMyThreadInfo->thr_error_stack_depth)
#define CritSectionCount (GetMyThreadInfo->thr_criticalsec_count)
+#define NumRWLocksHeld (GetMyThreadInfo->thr_num_rwlocks_held)
+#define RWLocksHeld (GetMyThreadInfo->thr_rwlocks_held)
+#define NumMutexLocksHeld (GetMyThreadInfo->thr_num_mutexlocks_held)
+#define MutexLocksHeld (GetMyThreadInfo->thr_mutexlocks_held)
#define PG_exception_stack (GetMyThreadInfo->thr_sigjmp_buf)
#define MyConnection (GetMyThreadInfo->thr_conn)
#ifndef _GTM_COMMON_H
#define _GTM_COMMON_H
+/*
+ * We expect a very small number of concurrent locks, except for some cases
+ * where a thread may try to acquire thr_lock of all other threads. So keep the
+ * value relatively high.
+ *
+ * If you change GTM_MAX_THREADS, consider changing this too.
+ */
+#define GTM_MAX_SIMUL_RWLOCKS (1024 + 32)
+#define GTM_MAX_SIMUL_MUTEX (32)
+
#define GTM_COMMON_THREAD_INFO \
GTM_ThreadID thr_id; \
uint32 thr_localid; \
ErrorData thr_error_data[ERRORDATA_STACK_SIZE]; \
int thr_error_stack_depth; \
int thr_error_recursion_depth; \
- int thr_criticalsec_count;
+ int thr_criticalsec_count; \
+ int thr_num_rwlocks_held; \
+ GTM_RWLock *thr_rwlocks_held[GTM_MAX_SIMUL_RWLOCKS]; \
+ int thr_num_mutexlocks_held; \
+ GTM_MutexLock *thr_mutexlocks_held[GTM_MAX_SIMUL_MUTEX];
#endif
extern bool GTM_RWLockRelease(GTM_RWLock *lock);
extern int GTM_RWLockInit(GTM_RWLock *lock);
extern int GTM_RWLockDestroy(GTM_RWLock *lock);
+extern void GTM_RWLockReleaseAll(void);
extern bool GTM_MutexLockAcquire(GTM_MutexLock *lock);
extern bool GTM_MutexLockRelease(GTM_MutexLock *lock);
extern int GTM_MutexLockInit(GTM_MutexLock *lock);
extern int GTM_MutexLockDestroy(GTM_MutexLock *lock);
+extern void GTM_MutexLockReleaseAll(void);
extern int GTM_CVInit(GTM_CV *cv);
extern int GTM_CVSignal(GTM_CV *cv);
#define recursion_depth (GetMyThreadInfo->thr_error_recursion_depth)
#define errordata_stack_depth (GetMyThreadInfo->thr_error_stack_depth)
#define CritSectionCount (GetMyThreadInfo->thr_criticalsec_count)
+#define NumRWLocksHeld (GetMyThreadInfo->thr_num_rwlocks_held)
+#define RWLocksHeld (GetMyThreadInfo->thr_rwlocks_held)
+#define NumMutexLocksHeld (GetMyThreadInfo->thr_num_mutexlocks_held)
+#define MutexLocksHeld (GetMyThreadInfo->thr_mutexlocks_held)
#define PG_exception_stack (GetMyThreadInfo->thr_sigjmp_buf)
#define MyConnection (GetMyThreadInfo->thr_conn)
projects / postgres-xl.git / commitdiff