lock_xt.cc

/* Copyright (c) 2005 PrimeBase Technologies GmbH
 *
 * PrimeBase XT
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 *
 * 2008-01-24	Paul McCullagh
 *
 * Row lock functions.
 *
 * H&G2JCtL
 */

#include "xt_config.h"

#ifdef DRIZZLED
#include <bitset>
#endif

#include <stdio.h>

#include "lock_xt.h"
#include "thread_xt.h"
#include "table_xt.h"
#include "xaction_xt.h"
#include "database_xt.h"
#include "trace_xt.h"

#ifdef DEBUG
//#define XT_TRACE_LOCKS
//#define CHECK_ROWLOCK_GROUP_CONSISTENCY
#endif

/*
 * This function should never be called. It indicates a link
 * error!
 */
xtPublic void xt_log_atomic_error_and_abort(c_char *func, c_char *file, u_int line)
{
	xt_logf(NULL, func, file, line, XT_LOG_ERROR, "%s", "Atomic operations not supported\n");
	abort();
}

/*
 * -----------------------------------------------------------------------
 * ROW LOCKS, LIST BASED
 */
#ifdef XT_USE_LIST_BASED_ROW_LOCKS

#ifdef CHECK_ROWLOCK_GROUP_CONSISTENCY
/* 
 * Requires a spin-lock on group->lg_lock!
 */
static void check_rowlock_group(XTLockGroupPtr group)
{
	XTThreadPtr self = xt_get_self();

	char *crash = NULL;

	if (group->lg_lock.spl_locker != self)
		*crash = 1;

	if (group->lg_list_in_use > group->lg_list_size)
		*crash = 1;

	xtRowID prev_row = 0;
	XTLockItemPtr item = group->lg_list;

	for (int i = 0; i < group->lg_list_in_use; i++, item++) {

		if (!item->li_thread_id)
			*crash = 1;

		if(!xt_thr_array[item->li_thread_id]->st_xact_data)
			*crash = 1;

		if(item->li_count > XT_TEMP_LOCK_BYTES)
			*crash = 1;

		// rows per thread must obey the row_id > prev_row_id + prev_count*group_size rule
		if (prev_row >= item->li_row_id)
			*crash = 1;

		// calculate the new prev. row
		if (item->li_count < XT_TEMP_LOCK_BYTES)
			prev_row = item->li_row_id + (item->li_count - 1) * XT_ROW_LOCK_GROUP_COUNT;
		else
			prev_row = item->li_row_id;
	}
}
#endif

static int xlock_cmp_row_ids(XTThreadPtr XT_UNUSED(self), register const void *XT_UNUSED(thunk), register const void *a, register const void *b)
{
	xtRowID			row_id = *((xtTableID *) a);
	XTLockItemPtr	item = (XTLockItemPtr) b;

	if (row_id < item->li_row_id)
		return -1;
	if (row_id > item->li_row_id)
		return 1;
	return 0;
}

void XTRowLockList::xt_remove_all_locks(struct XTDatabase *, XTThreadPtr thread)
{
#ifdef XT_TRACE_LOCKS
	xt_ttracef(xt_get_self(), "remove all locks\n");
#endif
	if (!bl_count)
		return;

	xtThreadID			thd_id;
	XTPermRowLockPtr	plock;
#ifndef XT_USE_TABLE_REF
	XTOpenTablePtr		pot = NULL;
#endif

	thd_id = thread->t_id;
	plock = (XTPermRowLockPtr) bl_data;
	for (u_int i=0; i<bl_count; i++) {
#ifdef XT_USE_TABLE_REF
		XTTableHPtr		tab = plock->pr_table;
#else
		if (!xt_db_open_pool_table_ns(&pot, db, plock->pr_tab_id)) {
			/* Should not happen, but just in case, we just don't
			 * remove the lock. We will probably end up with a deadlock
			 * somewhere.
			 */
			xt_log_and_clear_exception_ns();
		}
		else {
#endif
			for (int j=0; j<XT_ROW_LOCK_GROUP_COUNT; j++) {
				if (plock->pr_group[j]) {
					/* Go through group j and compact. */
#ifndef XT_USE_TABLE_REF
					XTTableHPtr		tab = pot->ot_table;
#endif
					XTLockGroupPtr	group;
					XTLockItemPtr	copy;
					XTLockItemPtr	item;
					int				new_count;

					group = &tab->tab_locks.rl_groups[j];
					xt_spinlock_lock(&group->lg_lock);
					copy = group->lg_list;
					item = group->lg_list;
					new_count = 0;
					for (size_t k=0; k<group->lg_list_in_use; k++) {
						if (item->li_thread_id != thd_id) {
							if (copy != item) {
								copy->li_row_id = item->li_row_id;
								copy->li_count = item->li_count;
								copy->li_thread_id = item->li_thread_id;
							}
							new_count++;
							copy++;
						}
#ifdef XT_TRACE_LOCKS
						else {
							if (item->li_count == XT_TEMP_LOCK_BYTES)
								xt_ttracef(xt_get_self(), "remove group %d lock row_id=%d TEMP\n", j, (int) item->li_row_id);
							else
								xt_ttracef(xt_get_self(), "remove group %d locks row_id=%d (%d)\n", j, (int) item->li_row_id, (int) item->li_count);
						}
#endif
						item++;
					}
					group->lg_list_in_use = new_count;
#ifdef CHECK_ROWLOCK_GROUP_CONSISTENCY
					check_rowlock_group(group);
#endif
					if (group->lg_wait_queue)
						tab->tab_locks.rl_grant_locks(group, thread);

					xt_spinlock_unlock(&group->lg_lock);
					
					xt_xn_wakeup_thread_list(thread);
				}
			}
#ifdef XT_USE_TABLE_REF
			xt_heap_release(NULL, plock->pr_table);
#else
			xt_db_return_table_to_pool_ns(pot);
		}
#endif
		plock++;
	}
	bl_count = 0;
}

#ifdef DEBUG_LOCK_QUEUE
int *dummy_ptr = 0;

void XTRowLocks::rl_check(XTLockWaitPtr no_lw)
{
	XTLockGroupPtr	group;
	XTLockWaitPtr	lw, lw_prev;

	for (int i=0; i<XT_ROW_LOCK_GROUP_COUNT; i++) {
		group = &rl_groups[i];
		xt_spinlock_lock(&group->lg_lock);

		lw = group->lg_wait_queue;
		lw_prev = NULL;
		while (lw) {
			if (lw == no_lw)
				*dummy_ptr = 1;
			if (lw->lw_prev != lw_prev)
				*dummy_ptr = 2;
			lw_prev = lw;
			lw = lw->lw_next;
		}
		xt_spinlock_unlock(&group->lg_lock);
	}
}
#endif

xtBool XTRowLocks::rl_lock_row(XTLockGroupPtr group, XTLockWaitPtr lw, XTRowLockListPtr, int *result)
{
	XTLockItemPtr	item;
	size_t			index;
	xtRowID			row_id = lw->lw_row_id;

#ifdef CHECK_ROWLOCK_GROUP_CONSISTENCY
	check_rowlock_group(group);
#endif
	if (group->lg_list_size == group->lg_list_in_use) {
		if (!xt_realloc_ns((void **) &group->lg_list, (group->lg_list_size + 2) * sizeof(XTLockItemRec)))
			return FAILED;
		group->lg_list_size += 2;
	}
	item = (XTLockItemPtr) xt_bsearch(NULL, &row_id, group->lg_list, group->lg_list_in_use, sizeof(XTLockItemRec), &index, NULL, xlock_cmp_row_ids);
	
	/* There's no item with this ID, but there could be an item with a range that covers this row */
	if (!item && group->lg_list_in_use) {
		if (index > 0) {
			int count;
	
			item = group->lg_list + index - 1;

			count = item->li_count;
			if (item->li_count == XT_TEMP_LOCK_BYTES)
				count = 1;

			if (row_id >= item->li_row_id + count * XT_ROW_LOCK_GROUP_COUNT)
				item = NULL;
		}
	}
	
	if (item) {
		/* Item already exists. */
		if (item->li_thread_id == lw->lw_thread->t_id) {
			/* Already have a permanent lock: */
			*result = XT_NO_LOCK;
			lw->lw_curr_lock = XT_NO_LOCK;
			return OK;
		}
		/* {REMOVE-LOCKS}
		 * This must be valid because a thread must remove
		 * the locks before it frees its st_xact_data structure,
		 * xt_thr_array entry must also be valid, because
		 * transaction must be ended before the thread is
		 * killed.
		 */
		*result = item->li_count == XT_TEMP_LOCK_BYTES ? XT_TEMP_LOCK : XT_PERM_LOCK;
		lw->lw_xn_id = xt_thr_array[item->li_thread_id]->st_xact_data->xd_start_xn_id;
		lw->lw_curr_lock = *result;
		return OK;
	}

	/* Add the lock: */
	XT_MEMMOVE(group->lg_list, &group->lg_list[index+1], 
		&group->lg_list[index], (group->lg_list_in_use - index) * sizeof(XTLockItemRec));
	group->lg_list[index].li_row_id = row_id;
	group->lg_list[index].li_count = XT_TEMP_LOCK_BYTES;
	group->lg_list[index].li_thread_id = lw->lw_thread->t_id;
	group->lg_list_in_use++;

#ifdef XT_TRACE_LOCKS
	xt_ttracef(ot->ot_thread, "set temp lock row=%d setby=%s\n", (int) row_id, xt_get_self()->t_name);
#endif
#ifdef CHECK_ROWLOCK_GROUP_CONSISTENCY
	check_rowlock_group(group);
#endif
	*result = XT_NO_LOCK;
	lw->lw_ot->ot_temp_row_lock = row_id;
	lw->lw_curr_lock = XT_NO_LOCK;
	return OK;
}

void XTRowLocks::rl_grant_locks(XTLockGroupPtr group, XTThreadPtr thread)
{
	XTLockWaitPtr	lw, lw_next, lw_prev;
	int				result;
	xtThreadID		lw_thd_id;

	thread->st_thread_list_count = 0;
	lw = group->lg_wait_queue;
	while (lw) {
		lw_next = lw->lw_next;
		lw_prev = lw->lw_prev;
		lw_thd_id = lw->lw_thread->t_id;
		/* NOTE: after lw_curr_lock is changed, lw may no longer be referenced
		 * by this function!!!
		 */
		if (!rl_lock_row(group, lw, &lw->lw_thread->st_lock_list, &result)) {
			/* We transfer the error to the other thread! */
			XTThreadPtr self = xt_get_self();

			result = XT_LOCK_ERR;
			memcpy(&lw->lw_thread->t_exception, &self->t_exception, sizeof(XTExceptionRec));
			lw->lw_curr_lock = XT_LOCK_ERR;
		}
		if (result == XT_NO_LOCK || result == XT_LOCK_ERR) {
			/* Remove from the wait queue: */
			if (lw_next)
				lw_next->lw_prev = lw_prev;
			if (lw_prev)
				lw_prev->lw_next = lw_next;
			if (group->lg_wait_queue == lw)
				group->lg_wait_queue = lw_next;
			if (group->lg_wait_queue_end == lw)
				group->lg_wait_queue_end = lw_prev;
			if (result == XT_NO_LOCK) {
				/* Add to the thread list: */
				if (thread->st_thread_list_count == thread->st_thread_list_size) {
					if (!xt_realloc_ns((void **) &thread->st_thread_list, (thread->st_thread_list_size+1) * sizeof(xtThreadID))) {
						xt_xn_wakeup_thread(lw_thd_id);
						goto done;
					}
					thread->st_thread_list_size++;
				}
				thread->st_thread_list[thread->st_thread_list_count] = lw_thd_id;
				thread->st_thread_list_count++;
				done:;
			}
		}
		lw = lw_next;
	}
}

void XTRowLocks::xt_cancel_temp_lock(XTLockWaitPtr lw)
{
	XTLockGroupPtr	group;

	group = &rl_groups[lw->lw_row_id % XT_ROW_LOCK_GROUP_COUNT];
	xt_spinlock_lock(&group->lg_lock);
	if (lw->lw_curr_lock == XT_TEMP_LOCK || lw->lw_curr_lock == XT_PERM_LOCK) {
		/* In case of XT_LOCK_ERR or XT_NO_LOCK, the lw structure will
		 * no longer be on the wait queue.
		 */
		XTLockWaitPtr	lw_next, lw_prev;

		lw_next = lw->lw_next;
		lw_prev = lw->lw_prev;

		/* Remove from the wait queue: */
		if (lw_next)
			lw_next->lw_prev = lw_prev;
		if (lw_prev)
			lw_prev->lw_next = lw_next;
		if (group->lg_wait_queue == lw)
			group->lg_wait_queue = lw_next;
		if (group->lg_wait_queue_end == lw)
			group->lg_wait_queue_end = lw_prev;
	}
	xt_spinlock_unlock(&group->lg_lock);
}

//#define QUEUE_ORDER_FIFO

/* Try to lock a row.
 * This function returns:
 * XT_NO_LOCK on success.
 * XT_TEMP_LOCK if there is a temporary lock on the row.
 * XT_PERM_LOCK if there is a permanent lock in the row.
 * XT_FAILED an error occured.
 *
 * If there is a lock on this row, the transaction ID of the
 * locker is also returned.
 *
 * The caller must wait if the row is locked. If the lock is
 * permanent, then the caller must wait for the transaction to
 * terminate. If the lock is temporary, then the caller must
 * wait for the transaction to signal that the lock has been
 * released.
 */
xtBool XTRowLocks::xt_set_temp_lock(XTOpenTablePtr ot, XTLockWaitPtr lw, XTRowLockListPtr lock_list)
{
	XTLockGroupPtr	group;
	int				result;

	if (ot->ot_temp_row_lock) {
		/* Check if we don't already have this temp lock: */
		if (ot->ot_temp_row_lock == lw->lw_row_id) {
			lw->lw_curr_lock = XT_NO_LOCK;
			return OK;
		}

		xt_make_lock_permanent(ot, lock_list);
	}

	/* Add a temporary lock. */
	group = &rl_groups[lw->lw_row_id % XT_ROW_LOCK_GROUP_COUNT];
	xt_spinlock_lock(&group->lg_lock);

	if (!rl_lock_row(group, lw, lock_list, &result)) {
		xt_spinlock_unlock(&group->lg_lock);
		return FAILED;
	}

	if (result != XT_NO_LOCK) {
		/* Add the thread to the end of the thread queue: */
#ifdef QUEUE_ORDER_FIFO
		if (group->lg_wait_queue_end) {
			group->lg_wait_queue_end->lw_next = lw;
			lw->lw_prev = group->lg_wait_queue_end;
		}
		else {
			group->lg_wait_queue = lw;
			lw->lw_prev = NULL;
		}
		lw->lw_next = NULL;
		group->lg_wait_queue_end = lw;
#else
		XTLockWaitPtr	pos = group->lg_wait_queue_end;
		xtXactID		xn_id = ot->ot_thread->st_xact_data->xd_start_xn_id;
		
		while (pos) {
			if (pos->lw_thread->st_xact_data->xd_start_xn_id < xn_id)
				break;
			pos = pos->lw_prev;
		}
		if (pos) {
			lw->lw_prev = pos;
			lw->lw_next = pos->lw_next;
			if (pos->lw_next)
				pos->lw_next->lw_prev = lw;
			else
				group->lg_wait_queue_end = lw;
			pos->lw_next = lw;
		}
		else {
			/* Front of the queue: */
			lw->lw_prev = NULL;
			lw->lw_next = group->lg_wait_queue;
			if (group->lg_wait_queue)
				group->lg_wait_queue->lw_prev = lw;
			else
				group->lg_wait_queue_end = lw;
			group->lg_wait_queue = lw;
		}
#endif
	}

	xt_spinlock_unlock(&group->lg_lock);
	return OK;
}

/*
 * Remove a temporary lock.
 * 
 * If updated is set to TRUE this means that the row was update.
 * This means that any thread waiting on the temporary lock will
 * also have to wait for the transaction to quit before
 * continuing.
 *
 * If the thread were to continue it would just hang again because
 * it will discover that the transaction has updated the row.
 *
 * So the 'updated' flag is an optimisation which prevents the
 * thread from making an unncessary retry.
 */
void XTRowLocks::xt_remove_temp_lock(XTOpenTablePtr ot, xtBool updated)
{
	xtRowID			row_id;
	XTLockGroupPtr	group;
	XTLockItemPtr	item;
	size_t			index;
	xtBool			lock_granted = FALSE;
	xtThreadID		locking_thread_id = 0;

	if (!(row_id = ot->ot_temp_row_lock))
		return;

	group = &rl_groups[row_id % XT_ROW_LOCK_GROUP_COUNT];
	xt_spinlock_lock(&group->lg_lock);
#ifdef CHECK_ROWLOCK_GROUP_CONSISTENCY
	check_rowlock_group(group);
#endif

#ifdef XT_TRACE_LOCKS
	xt_ttracef(xt_get_self(), "remove temp lock %d\n", (int) row_id);
#endif
	item = (XTLockItemPtr) xt_bsearch(NULL, &row_id, group->lg_list, group->lg_list_in_use, sizeof(XTLockItemRec), &index, NULL, xlock_cmp_row_ids);
	if (item) {
		/* Item exists. */
		if (item->li_thread_id == ot->ot_thread->t_id &&
			item->li_count == XT_TEMP_LOCK_BYTES) {
			XTLockWaitPtr	lw;

			/* First check if there is some thread waiting to take over this lock: */
			lw = group->lg_wait_queue;
			while (lw) {
				if (lw->lw_row_id == row_id) {
					lock_granted = TRUE;
					break;
				}
				lw = lw->lw_next;
			}

			if (lock_granted) {
				/* Grant the lock just released... */
				XTLockWaitPtr	lw_next, lw_prev;
				xtXactID		locking_xact_id;

				/* Store this info, lw will soon be untouchable! */
				lw_next = lw->lw_next;
				lw_prev = lw->lw_prev;
				locking_xact_id = lw->lw_thread->st_xact_data->xd_start_xn_id;
				locking_thread_id = lw->lw_thread->t_id;

				/* Lock has moved from one thread to the next.
				 * change the thread holding this lock:
				 */
				item->li_thread_id = locking_thread_id;

				/* Remove from the wait queue: */
				if (lw_next)
					lw_next->lw_prev = lw_prev;
				if (lw_prev)
					lw_prev->lw_next = lw_next;
				if (group->lg_wait_queue == lw)
					group->lg_wait_queue = lw_next;
				if (group->lg_wait_queue_end == lw)
					group->lg_wait_queue_end = lw_prev;

				/* If the thread that release the lock updated the
				 * row then we will have to wait for the transaction
				 * to terminate:
				 */
				if (updated) {
					lw->lw_row_updated = TRUE;
					lw->lw_updating_xn_id = ot->ot_thread->st_xact_data->xd_start_xn_id;
				}

				/* The thread has the lock now: */
				lw->lw_ot->ot_temp_row_lock = row_id;
				lw->lw_curr_lock = XT_NO_LOCK;

				/* Everyone after this that is waiting for the same lock is
				 * now waiting for a different transaction:
				 */
				lw = lw_next;
				while (lw) {
					if (lw->lw_row_id == row_id) {
						lw->lw_xn_id = locking_xact_id;
						lw->lw_curr_lock = XT_TEMP_LOCK;
					}
					lw = lw->lw_next;
				}
			}
			else {
				/* Remove the lock: */
				XT_MEMMOVE(group->lg_list, &group->lg_list[index], 
					&group->lg_list[index+1], (group->lg_list_in_use - index - 1) * sizeof(XTLockItemRec));
				group->lg_list_in_use--;
			}
		}
	}
#ifdef CHECK_ROWLOCK_GROUP_CONSISTENCY
	check_rowlock_group(group);
#endif
	xt_spinlock_unlock(&group->lg_lock);

	ot->ot_temp_row_lock = 0;
	if (lock_granted)
		xt_xn_wakeup_thread(locking_thread_id);
}

xtBool XTRowLocks::xt_make_lock_permanent(XTOpenTablePtr ot, XTRowLockListPtr lock_list)
{
	xtRowID			row_id;
	XTLockGroupPtr	group;
	XTLockItemPtr	item;
	size_t			index;

	if (!(row_id = ot->ot_temp_row_lock))
		return OK;

#ifdef XT_TRACE_LOCKS
	xt_ttracef(xt_get_self(), "make lock perm %d\n", (int) ot->ot_temp_row_lock);
#endif

	/* Add to the lock list: */
	XTPermRowLockPtr locks = (XTPermRowLockPtr) lock_list->bl_data;
	for (unsigned i=0; i<lock_list->bl_count; i++) {
#ifdef XT_USE_TABLE_REF
		if (locks->pr_table == ot->ot_table) {
#else
		if (locks->pr_tab_id == ot->ot_table->tab_id) {
#endif
			locks->pr_group[row_id % XT_ROW_LOCK_GROUP_COUNT] = 1;
			goto done;
		}
		locks++;
	}

	/* Add new to lock list: */
	{
		XTPermRowLockRec perm_lock;
		
#ifdef XT_USE_TABLE_REF
		perm_lock.pr_table = ot->ot_table;
		xt_heap_reference(NULL, perm_lock.pr_table);
#else
		perm_lock.pr_tab_id = ot->ot_table->tab_id;
#endif
		memset(perm_lock.pr_group, 0, XT_ROW_LOCK_GROUP_COUNT);
		perm_lock.pr_group[row_id % XT_ROW_LOCK_GROUP_COUNT] = 1;
		if (!xt_bl_append(NULL, lock_list, &perm_lock)) {
			xt_remove_temp_lock(ot, FALSE);
			return FAILED;
		}
	}

	done:
	group = &rl_groups[row_id % XT_ROW_LOCK_GROUP_COUNT];
	xt_spinlock_lock(&group->lg_lock);

	item = (XTLockItemPtr) xt_bsearch(NULL, &row_id, group->lg_list, group->lg_list_in_use, sizeof(XTLockItemRec), &index, NULL, xlock_cmp_row_ids);
	ASSERT_NS(item);
#ifdef CHECK_ROWLOCK_GROUP_CONSISTENCY
	check_rowlock_group(group);
#endif
	if (item) {
		/* Lock exists (it should!). */
		if (item->li_thread_id == ot->ot_thread->t_id &&
			item->li_count == XT_TEMP_LOCK_BYTES) {
			if (index > 0 &&
				group->lg_list[index-1].li_thread_id == ot->ot_thread->t_id &&
				group->lg_list[index-1].li_count < XT_TEMP_LOCK_BYTES-2 &&
				group->lg_list[index-1].li_row_id == row_id - (XT_ROW_LOCK_GROUP_COUNT * group->lg_list[index-1].li_count)) {
				group->lg_list[index-1].li_count++;
				/* Combine with the left: */
				if (index + 1 < group->lg_list_in_use &&
					group->lg_list[index+1].li_thread_id == ot->ot_thread->t_id &&
					group->lg_list[index+1].li_count != XT_TEMP_LOCK_BYTES &&
					group->lg_list[index+1].li_row_id == row_id + XT_ROW_LOCK_GROUP_COUNT) {
					/* And combine with the right */
					u_int left = group->lg_list[index-1].li_count + group->lg_list[index+1].li_count;
					u_int right;

					if (left > XT_TEMP_LOCK_BYTES-1) {
						right = left - (XT_TEMP_LOCK_BYTES-1);
						left = XT_TEMP_LOCK_BYTES-1;
					}
					else
						right = 0;

					group->lg_list[index-1].li_count = left;
					if (right) {
						/* There is something left over on the right: */
						group->lg_list[index+1].li_count = right;
						group->lg_list[index+1].li_row_id = group->lg_list[index-1].li_row_id + left * XT_ROW_LOCK_GROUP_COUNT;
						XT_MEMMOVE(group->lg_list, &group->lg_list[index], 
							&group->lg_list[index+1], (group->lg_list_in_use - index - 1) * sizeof(XTLockItemRec));
						group->lg_list_in_use--;
					}
					else {
						XT_MEMMOVE(group->lg_list, &group->lg_list[index], 
							&group->lg_list[index+2], (group->lg_list_in_use - index - 2) * sizeof(XTLockItemRec));
						group->lg_list_in_use -= 2;
					}
				}
				else {
					XT_MEMMOVE(group->lg_list, &group->lg_list[index], 
						&group->lg_list[index+1], (group->lg_list_in_use - index - 1) * sizeof(XTLockItemRec));
					group->lg_list_in_use--;
				}
			}
			else if (index + 1 < group->lg_list_in_use &&
					group->lg_list[index+1].li_thread_id == ot->ot_thread->t_id &&
					group->lg_list[index+1].li_count < XT_TEMP_LOCK_BYTES-2 &&
					group->lg_list[index+1].li_row_id == row_id + XT_ROW_LOCK_GROUP_COUNT) {
				/* Combine with the right: */
				group->lg_list[index+1].li_count++;
				group->lg_list[index+1].li_row_id = row_id;
				XT_MEMMOVE(group->lg_list, &group->lg_list[index], 
					&group->lg_list[index+1], (group->lg_list_in_use - index - 1) * sizeof(XTLockItemRec));
				group->lg_list_in_use--;
			}
			else
				group->lg_list[index].li_count = 1;
		}
	}
#ifdef CHECK_ROWLOCK_GROUP_CONSISTENCY
	check_rowlock_group(group);
#endif
	xt_spinlock_unlock(&group->lg_lock);

	ot->ot_temp_row_lock = 0;
	return OK;
}

xtBool xt_init_row_locks(XTRowLocksPtr rl)
{
	for (int i=0; i<XT_ROW_LOCK_GROUP_COUNT; i++) {
		xt_spinlock_init_with_autoname(NULL, &rl->rl_groups[i].lg_lock);
		rl->rl_groups[i].lg_wait_queue = NULL;
		rl->rl_groups[i].lg_list_size = 0;
		rl->rl_groups[i].lg_list_in_use = 0;
		rl->rl_groups[i].lg_list = NULL;
	}
	rl->valid = 1;
	return OK;
}

void xt_exit_row_locks(XTRowLocksPtr rl)
{
	if (!rl->valid)
		return;

	for (int i=0; i<XT_ROW_LOCK_GROUP_COUNT; i++) {
		xt_spinlock_free(NULL, &rl->rl_groups[i].lg_lock);
		rl->rl_groups[i].lg_wait_queue = NULL;
		rl->rl_groups[i].lg_list_size = 0;
		rl->rl_groups[i].lg_list_in_use = 0;
		if (rl->rl_groups[i].lg_list) {
			xt_free_ns(rl->rl_groups[i].lg_list);
			rl->rl_groups[i].lg_list = NULL;
		}
	}
	rl->valid = 0;
}

/*
 * -----------------------------------------------------------------------
 * ROW LOCKS, HASH BASED
 */
#else // XT_USE_LIST_BASED_ROW_LOCKS

void XTRowLockList::old_xt_remove_all_locks(struct XTDatabase *db, xtThreadID thd_id)
{
#ifdef XT_TRACE_LOCKS
	xt_ttracef(xt_get_self(), "remove all locks\n");
#endif
	if (!bl_count)
		return;

	int					pgroup;
	xtTableID			ptab_id;
	XTPermRowLockPtr	plock;
	XTOpenTablePtr		pot = NULL;

	plock = (XTPermRowLockPtr) &bl_data[bl_count * bl_item_size];
	for (u_int i=0; i<bl_count; i++) {
		plock--;
		pgroup = plock->pr_group;
		ptab_id = plock->pr_tab_id;
		if (pot) {
			if (pot->ot_table->tab_id == ptab_id)
				goto remove_lock;
			xt_db_return_table_to_pool_ns(pot);
			pot = NULL;
		}

		if (!xt_db_open_pool_table_ns(&pot, db, ptab_id)) {
			/* Should not happen, but just in case, we just don't
			 * remove the lock. We will probably end up with a deadlock
			 * somewhere.
			 */
			xt_log_and_clear_exception_ns();
			goto skip_remove_lock;
		}
		if (!pot)
			/* Can happen of the table has been dropped: */
			goto skip_remove_lock;

		remove_lock:
#ifdef XT_TRACE_LOCKS
		xt_ttracef(xt_get_self(), "remove lock group=%d\n", pgroup);
#endif
		pot->ot_table->tab_locks.tab_row_locks[pgroup] = NULL;
		pot->ot_table->tab_locks.tab_lock_perm[pgroup] = 0;
		skip_remove_lock:;
	}
	bl_count = 0;

	if (pot)
		xt_db_return_table_to_pool_ns(pot);
}

/* Try to lock a row.
 * This function returns:
 * XT_NO_LOCK on success.
 * XT_TEMP_LOCK if there is a temporary lock on the row.
 * XT_PERM_LOCK if there is a permanent lock in the row.
 *
 * If there is a lock on this row, the transaction ID of the
 * locker is also returned.
 *
 * The caller must wait if the row is locked. If the lock is
 * permanent, then the caller must wait for the transaction to
 * terminate. If the lock is temporary, then the caller must
 * wait for the transaction to signal that the lock has been
 * released.
 */
int XTRowLocks::old_xt_set_temp_lock(XTOpenTablePtr ot, xtRowID row, xtXactID *xn_id, XTRowLockListPtr lock_list)
{
	int				group;
	XTXactDataPtr	xact, my_xact;

	if (ot->ot_temp_row_lock) {
		/* Check if we don't already have this temp lock: */
		if (ot->ot_temp_row_lock == row) {
			gl->lw_curr_lock = XT_NO_LOCK;
			return XT_NO_LOCK;
		}

		xt_make_lock_permanent(ot, lock_list);
	}

	my_xact = ot->ot_thread->st_xact_data;
	group = row % XT_ROW_LOCK_COUNT;
	if ((xact = tab_row_locks[group])) {
		if (xact == my_xact)
			return XT_NO_LOCK;
		*xn_id = xact->xd_start_xn_id;
		return tab_lock_perm[group] ? XT_PERM_LOCK : XT_TEMP_LOCK;
	}

	tab_row_locks[row % XT_ROW_LOCK_COUNT] = my_xact;

#ifdef XT_TRACE_LOCKS
	xt_ttracef(xt_get_self(), "set temp lock %d group=%d for %s\n", (int) row, (int) row % XT_ROW_LOCK_COUNT, ot->ot_thread->t_name);
#endif
	ot->ot_temp_row_lock = row;
	return XT_NO_LOCK;
}

/* Just check if there is a lock on the row.
 * This function returns:
 * XT_NO_LOCK if there is no lock.
 * XT_TEMP_LOCK if there is a temporary lock on the row.
 * XT_PERM_LOCK if a lock is a permanent lock in the row.
 */
int XTRowLocks::old_xt_is_locked(struct XTOpenTable *ot, xtRowID row, xtXactID *xn_id)
{
	int				group;
	XTXactDataPtr	xact;

	group = row % XT_ROW_LOCK_COUNT;
	if ((xact = tab_row_locks[group])) {
		if (xact == ot->ot_thread->st_xact_data)
			return XT_NO_LOCK;
		*xn_id = xact->xd_start_xn_id;
		if (tab_lock_perm[group])
			return XT_PERM_LOCK;
		return XT_TEMP_LOCK;
	}
	return XT_NO_LOCK;
}

void XTRowLocks::old_xt_remove_temp_lock(XTOpenTablePtr ot)
{
	int				group;
	XTXactDataPtr	xact, my_xact;

	if (!ot->ot_temp_row_lock)
		return;

	my_xact = ot->ot_thread->st_xact_data;
	group = ot->ot_temp_row_lock % XT_ROW_LOCK_COUNT;
#ifdef XT_TRACE_LOCKS
	xt_ttracef(xt_get_self(), "remove temp lock %d group=%d\n", (int) ot->ot_temp_row_lock, (int) ot->ot_temp_row_lock % XT_ROW_LOCK_COUNT);
#endif
	ot->ot_temp_row_lock = 0;
	if ((xact = tab_row_locks[group])) {
		if (xact == my_xact)
			tab_row_locks[group] = NULL;
	}

	if (ot->ot_table->tab_db->db_xn_wait_count)
		xt_xn_wakeup_transactions(ot->ot_table->tab_db, ot->ot_thread);
}

xtBool XTRowLocks::old_xt_make_lock_permanent(XTOpenTablePtr ot, XTRowLockListPtr lock_list)
{
	int group;

	if (!ot->ot_temp_row_lock)
		return OK;

#ifdef XT_TRACE_LOCKS
	xt_ttracef(xt_get_self(), "make lock perm %d group=%d\n", (int) ot->ot_temp_row_lock, (int) ot->ot_temp_row_lock % XT_ROW_LOCK_COUNT);
#endif
	/* Check if the lock is already permanent: */
	group = ot->ot_temp_row_lock % XT_ROW_LOCK_COUNT;
	if (!tab_lock_perm[group]) {
		XTPermRowLockRec plock;

		plock.pr_tab_id = ot->ot_table->tab_id;
		plock.pr_group = group;
		if (!xt_bl_append(NULL, lock_list, &plock)) {
			xt_remove_temp_lock(ot);
			return FAILED;
		}
		tab_lock_perm[group] = 1;
	}

	ot->ot_temp_row_lock = 0;
	return OK;
}

/* Release this lock, and all locks gained after this lock
 * on this table.
 *
 * The locks are only released temporarily. The will be regained
 * below using regain locks.
 *
 * Returns:
 * XT_NO_LOCK if no lock is released.
 * XT_PERM_LOCK if a lock is released.
 *
 * Note that only permanent locks can be released in this way.
 * So if the thread has a temporary lock, it will first be made
 * permanent.
 *
 * {RELEASING-LOCKS}
 * The idea of the releasing locks comes from the fact that each
 * lock, locks a group of records.
 * So if T1 has a lock (e.g. when doing SELECT FOR UPDATE),
 * and then encounters an updated record x
 * from T2, and it must wait for T2, it firsts releases the
 * lock, just in case T2 tries to gain a lock on another
 * record y in the same group, which will cause it to
 * wait on T1.
 *
 * However, there are several problems with releasing
 * locks.
 * - It can cause a "live-lock", where another transation
 * keeps getting in before.
 * - It may not solve the problem in all cases because
 * the SELECT FOR UPDATE has locked other record groups
 * before it encountered record x.
 * - Further problems occur when locks are granted by
 * callback:
 * T1 waits for T2, because it has a lock on record x
 * T2 releases the lock because it must wait for T3
 * T1 is granted the lock (but does not know about this yet)
 * T2 tries to regain lock (after T3 quits) and
 * must wait for T1 - DEADLOCK
 *
 * In general, it does not make sense to release locks
 * when it can be granted again by a callback.
 *
 * TODO: 2 possible solutions:
 * - Do not lock groups, lock rows.
 *   UPDATE INTENSION ROW LOCK
 * - Use multiple lock types:
 *   UPDATE INTENSION LOCK (required first)
 *   SHARED UPDATE LOCK (used by INSERT or DELETE)
 *   EXCLUSIVE UPDATE LOCK (used by SELECT FOR UPDATE)
 *
 * Temporary solution. Do not release any locks.
int XTRowLocks::xt_release_locks(struct XTOpenTable *ot, xtRowID row, XTRowLockListPtr lock_list)
 */ 

/*
 * Regain a lock previously held. This function regains locks
 * released by xt_release_locks().
 *
 * It will return lock_type and xn_id if the row is locked, and therefore
 * regain cannot continue. In this case, the caller must wait.
 * It returns XT_NO_LOCK if there are no more locks to be regained.
 *
 * Locks are always regained in the order in which they were originally
 * taken.
xtBool XTRowLocks::xt_regain_locks(struct XTOpenTable *ot, int *lock_type, xtXactID *xn_id, XTRowLockListPtr lock_list)
 */

xtBool old_xt_init_row_locks(XTRowLocksPtr rl)
{
	memset(rl->tab_lock_perm, 0, XT_ROW_LOCK_COUNT);
	memset(rl->tab_row_locks, 0, XT_ROW_LOCK_COUNT * sizeof(XTXactDataPtr));
	return OK;
}

void old_xt_exit_row_locks(XTRowLocksPtr XT_UNUSED(rl))
{
}

#endif // XT_USE_LIST_BASED_ROW_LOCKS

xtPublic xtBool xt_init_row_lock_list(XTRowLockListPtr lock_list)
{
	lock_list->bl_item_size = sizeof(XTPermRowLockRec);
	lock_list->bl_size = 0;
	lock_list->bl_count = 0;
	lock_list->bl_data = NULL;
	return OK;
}

xtPublic void xt_exit_row_lock_list(XTRowLockListPtr lock_list)
{
	xt_bl_set_size(NULL, lock_list, 0);
}

/*
 * -----------------------------------------------------------------------
 * SPECIAL EXCLUSIVE/SHARED (XS) LOCK
 */

#ifdef XT_THREAD_LOCK_INFO
xtPublic void xt_rwmutex_init(struct XTThread *self, XTRWMutexPtr xsl, const char *n)
#else
xtPublic void xt_rwmutex_init(XTThreadPtr self, XTRWMutexPtr xsl)
#endif
{
#ifdef DEBUG
	xsl->xs_lock_thread = 0;
	xsl->xs_inited = 12345;
#endif
	xt_init_mutex_with_autoname(self, &xsl->xs_lock);
	xt_init_cond(self, &xsl->xs_cond);
	xt_atomic_set4(&xsl->xs_state, 0);
	xsl->xs_xlocker = 0;
	/* Must be aligned! */
	ASSERT(xt_thr_maximum_threads == xt_align_size(xt_thr_maximum_threads, XT_XS_LOCK_ALIGN));
	xsl->x.xs_rlock = (xtWord1 *) xt_calloc(self, xt_thr_maximum_threads);
#ifdef XT_THREAD_LOCK_INFO
	xsl->xs_name = n;
	xt_thread_lock_info_init(&xsl->xs_lock_info, xsl);
#endif
}

xtPublic void xt_rwmutex_free(XTThreadPtr self, XTRWMutexPtr xsl)
{
#ifdef DEBUG
	ASSERT(!xsl->xs_lock_thread);
	ASSERT(xsl->xs_inited == 12345);
	xsl->xs_inited = 0;
#endif
	if (xsl->x.xs_rlock)
		xt_free(self, (void *) xsl->x.xs_rlock);
	xt_free_mutex(&xsl->xs_lock);
	xt_free_cond(&xsl->xs_cond);
#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_free(&xsl->xs_lock_info);
#endif
}

xtPublic xtBool xt_rwmutex_xlock(XTRWMutexPtr xsl, xtThreadID thd_id)
{
#ifdef DEBUG
	ASSERT_NS(xsl->xs_inited == 12345);
#endif
	ASSERT_NS(xt_get_self()->t_id == thd_id);
	xt_lock_mutex_ns(&xsl->xs_lock);
	ASSERT_NS(xsl->x.xs_rlock[thd_id] == XT_NO_LOCK);
	
	/* Wait for exclusive locker: */
	while (xsl->xs_xlocker) {
		if (!xt_timed_wait_cond_ns(&xsl->xs_cond, &xsl->xs_lock, 10000)) {
			xt_unlock_mutex_ns(&xsl->xs_lock);
			return FAILED;
		}
	}

	/* I am the locker (set state before locker!): */
	xt_atomic_set4(&xsl->xs_state, 0);
	xsl->xs_xlocker = thd_id;

	/* Wait for all the read lockers: */
	while (xsl->xs_state < xt_thr_current_max_threads) {
		while (xsl->x.xs_rlock[xsl->xs_state]) {
			/* {RACE-WR_MUTEX}
			 * Just in case of this, we keep the wait time down!
			 */
			if (!xt_timed_wait_cond_ns(&xsl->xs_cond, &xsl->xs_lock, 10)) {
				xt_atomic_set4(&xsl->xs_state, 0);
				xsl->xs_xlocker = 0;
				xt_unlock_mutex_ns(&xsl->xs_lock);
				return FAILED;
			}
		}
		/* State can be incremented in parallel by a reader
		 * thread!
		 */
		xt_atomic_set4(&xsl->xs_state, xsl->xs_state + 1);
	}

	/* I have waited for all: */
	xt_atomic_set4(&xsl->xs_state, xt_thr_maximum_threads);

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_add_owner(&xsl->xs_lock_info);
#endif

	return OK;
}

xtPublic xtBool xt_rwmutex_slock(XTRWMutexPtr xsl, xtThreadID thd_id)
{
#ifdef DEBUG
	ASSERT_NS(xsl->xs_inited == 12345);
#endif
	ASSERT_NS(xt_get_self()->t_id == thd_id);

	xt_atomic_inc1(&xsl->x.xs_rlock[thd_id]);

	if (xsl->x.xs_rlock[thd_id] > 1)
		return OK;

	/* Check if there could be an X locker: */
	if (xsl->xs_xlocker) {
		/* There is an X locker.
		 * If xs_state < thd_id then the X locker will wait for me.
		 * So I should not wait!
		 */
		if (xsl->xs_state >= thd_id) {
			/* If xsl->xs_state >= thd_id, then the locker has already
			 * checked me, and I will have to wait.
			 *
			 * Otherwise, xs_state <= thd_id, which means the
			 * X locker has not checked me, and will still wait for me (or 
			 * is already waiting for me). In this case, I will have to
			 * take the mutex to make sure exactly how far he
			 * is with the checking.
			 */
			xt_lock_mutex_ns(&xsl->xs_lock);
			while (xsl->xs_state > thd_id && xsl->xs_xlocker) {
				if (!xt_timed_wait_cond_ns(&xsl->xs_cond, &xsl->xs_lock, 10000)) {
					xt_unlock_mutex_ns(&xsl->xs_lock);
					xsl->x.xs_rlock[thd_id]--;
					return FAILED;
				}
			}
			xt_unlock_mutex_ns(&xsl->xs_lock);
		}
	}

	/* There is no exclusive locker, so we have the read lock: */
	ASSERT_NS(xsl->xs_state != xt_thr_maximum_threads);
#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_add_owner(&xsl->xs_lock_info);
#endif
	return OK;
}

xtPublic xtBool xt_rwmutex_unlock(XTRWMutexPtr xsl, xtThreadID thd_id)
{
#ifdef DEBUG
	ASSERT_NS(xsl->xs_inited == 12345);
#endif
	ASSERT_NS(xt_get_self()->t_id == thd_id);
	if (xsl->xs_xlocker == thd_id) {
		/* I have an X lock. */
		ASSERT_NS(xsl->x.xs_rlock[thd_id] == XT_NO_LOCK);
		ASSERT_NS(xsl->xs_state == xt_thr_maximum_threads);
		xt_atomic_set4(&xsl->xs_state, 0);
		xsl->xs_xlocker = 0;
		xt_unlock_mutex_ns(&xsl->xs_lock);
		/* Wake up any other X or shared lockers: */
		if (!xt_broadcast_cond_ns(&xsl->xs_cond))
			return FAILED;
	}
	else {
		/* I have a shared lock: */
		ASSERT_NS(xsl->x.xs_rlock[thd_id] > 0);
		ASSERT_NS(xsl->xs_state != xt_thr_maximum_threads); /* TODO: PMC - HOW can this fail?! - but it does? */
		if (xsl->x.xs_rlock[thd_id] > 1)
			xsl->x.xs_rlock[thd_id]--;
		else {
			/* {RACE-WR_MUTEX}.
			 * A BUG FIX:
			 *
			 * Previously I was checking "xsl->xs_xlocker" after,
			 * descrementing the READ lock.
			 *
			 * This resulted in a race condition that could cause the
			 * unlocking reader to hang in xt_lock_mutex_ns().
			 * This was because the X locker, grabbed the mutex (xs_lock)
			 * but did not wait for the reader.
			 *
			 * The result was that the reader had to wait in UNLOCK
			 * until the X locker did an unlock!
			 *
			 * This only became obvious when it caused a deadlock (because
			 * the reader was waiting for the locker, which it should not
			 * have been, of course).
			 */
			if (xsl->xs_xlocker) {
				xt_lock_mutex_ns(&xsl->xs_lock);
				if (xsl->xs_xlocker && xsl->xs_state == thd_id) {
					/* If the X locker is waiting for me,
					 * then allow him to continue. 
					 */
					if (!xt_broadcast_cond_ns(&xsl->xs_cond)) {
						xt_unlock_mutex_ns(&xsl->xs_lock);
						return FAILED;
					}
				}
				xt_atomic_dec1(&xsl->x.xs_rlock[thd_id]);
				xt_unlock_mutex_ns(&xsl->xs_lock);
			}
			else
				/* {RACE-WR_MUTEX}
				 * There is a race condition between the check above, and the
				 * the decrement here.
				 *
				 * However, if I check xsl->xs_xlocker afterwards, and then
				 * try to get the lock xs_lock, I could hand for the duration
				 * of the X lock.
				 */
				xt_atomic_dec1(&xsl->x.xs_rlock[thd_id]);
		}
	}
#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_release_owner(&xsl->xs_lock_info);
#endif
	return OK;
}

/*
 * -----------------------------------------------------------------------
 * SPIN LOCK
 */

#ifdef XT_THREAD_LOCK_INFO
xtPublic void xt_spinlock_init(XTThreadPtr self, XTSpinLockPtr spl, const char *n)
#else
xtPublic void xt_spinlock_init(XTThreadPtr self, XTSpinLockPtr spl)
#endif
{
	(void) self;
	spl->spl_lock = 0;
#ifdef XT_NO_ATOMICS
	xt_init_mutex_with_autoname(self, &spl->spl_mutex);
#endif
#ifdef DEBUG
	spl->spl_locker = 0;
#endif
#ifdef XT_THREAD_LOCK_INFO
	spl->spl_name = n;
	xt_thread_lock_info_init(&spl->spl_lock_info, spl);
#endif
}

xtPublic void xt_spinlock_free(XTThreadPtr XT_UNUSED(self), XTSpinLockPtr spl)
{
	(void) spl;
#ifdef XT_NO_ATOMICS
	xt_free_mutex(&spl->spl_mutex);
#endif
#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_free(&spl->spl_lock_info);
#endif
}

xtPublic xtBool xt_spinlock_spin(XTSpinLockPtr spl)
{
	volatile xtWord4	*lck = &spl->spl_lock;

	for (;;) {
		for (int i=0; i<10; i++) {
			/* Check the lock variable: */
			if (!*lck) {
				/* Try to get the lock: */
				if (!xt_spinlock_set(spl))
					goto done_ok;
			}
		}

		/* Go to "sleep" */
		xt_critical_wait();
	}

	done_ok:
	return OK;
}

#ifdef DEBUG
xtPublic void xt_spinlock_set_thread(XTSpinLockPtr spl)
{
	spl->spl_locker = xt_get_self();
}
#endif

/*
 * -----------------------------------------------------------------------
 * FAST LOCK
 */

#ifdef XT_THREAD_LOCK_INFO
xtPublic void xt_fastlock_init(XTThreadPtr self, XTFastLockPtr fal, const char *n)
#else
xtPublic void xt_fastlock_init(XTThreadPtr self, XTFastLockPtr fal)
#endif
{
	xt_spinlock_init_with_autoname(self, &fal->fal_spinlock);
	xt_spinlock_init_with_autoname(self, &fal->fal_wait_lock);
	for (u_int i=0; i<XT_FAST_LOCK_MAX_WAIT; i++)
		fal->fal_wait_list[i] = NULL;
	fal->fal_wait_count = 0;
	fal->fal_wait_wakeup = 0;
	fal->fal_wait_alloc = 0;
#ifdef XT_THREAD_LOCK_INFO
	fal->fal_name = n;
	xt_thread_lock_info_init(&fal->fal_lock_info, fal);
#endif
}

xtPublic void xt_fastlock_free(XTThreadPtr self, XTFastLockPtr fal)
{
	xt_spinlock_free(self, &fal->fal_spinlock);
	xt_spinlock_free(self, &fal->fal_wait_lock);
#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_free(&fal->fal_lock_info);
#endif
}

xtPublic xtBool xt_fastlock_spin(XTFastLockPtr fal, XTThreadPtr thread)
{
	volatile xtWord4	*lck = &fal->fal_spinlock.spl_lock;

	do {
		for (int i=0; i<10; i++) {
			/* Check the lock variable: */
			if (!*lck) {
				/* Try to get the lock: */
				if (!xt_spinlock_set(&fal->fal_spinlock)) {
					fal->fal_locker = thread;
					return OK;
				}
			}
		}

		for (int i=0; i<10; i++) {
			xt_critical_wait();
			if (!*lck) {
				/* Try to get the lock: */
				if (!xt_spinlock_set(&fal->fal_spinlock)) {
					fal->fal_locker = thread;
					return OK;
				}
			}
		}

		/* Wait for a wakeup */
		xt_spinlock_lock(&fal->fal_wait_lock);
		if (fal->fal_wait_count == XT_FAST_LOCK_MAX_WAIT) {
			xt_register_ulxterr(XT_REG_CONTEXT, XT_ERR_TOO_MANY_WAITERS, (u_long) XT_FAST_LOCK_MAX_WAIT+1);
			xt_spinlock_unlock(&fal->fal_wait_lock);
			return FAILED;
		}
		while (fal->fal_wait_list[fal->fal_wait_alloc])
			fal->fal_wait_alloc = (fal->fal_wait_alloc + 1) % XT_FAST_LOCK_MAX_WAIT;
		fal->fal_wait_list[fal->fal_wait_alloc] = thread;
		fal->fal_wait_alloc = (fal->fal_wait_alloc + 1) % XT_FAST_LOCK_MAX_WAIT;
		fal->fal_wait_count++;
		xt_lock_thread(thread);
		xt_spinlock_unlock(&fal->fal_wait_lock);
		if (!xt_wait_thread(thread)) {
			xt_unlock_thread(thread);
			if (fal->fal_locker == thread)
				xt_fastlock_unlock(fal, thread);
			return FAILED;
		}
		xt_unlock_thread(thread);
	} while (fal->fal_locker != thread);
	return OK;
}

/* Wake up one of the waiters. */
xtPublic void xt_fastlock_wakeup(XTFastLockPtr fal)
{
	xt_spinlock_lock(&fal->fal_wait_lock);
	if (fal->fal_wait_count) {
		XTThreadPtr thread;

		/* Find a waiting thread, and give it the exclusive lock: */
		while (!fal->fal_wait_list[fal->fal_wait_wakeup])
			fal->fal_wait_wakeup = (fal->fal_wait_wakeup + 1) % XT_FAST_LOCK_MAX_WAIT;
		thread = fal->fal_wait_list[fal->fal_wait_wakeup];
		fal->fal_wait_list[fal->fal_wait_wakeup] = NULL;
		fal->fal_wait_wakeup = (fal->fal_wait_wakeup + 1) % XT_FAST_LOCK_MAX_WAIT;
		fal->fal_wait_count--;
		fal->fal_locker = thread;

		xt_lock_thread(thread);
		xt_spinlock_unlock(&fal->fal_wait_lock);
		xt_signal_thread(thread);
		xt_unlock_thread(thread);
	}
	else {
		xt_spinlock_unlock(&fal->fal_wait_lock);
		fal->fal_locker = NULL;
		xt_spinlock_reset(&fal->fal_spinlock);
	}
}

/*
 * -----------------------------------------------------------------------
 * READ/WRITE SPIN LOCK
 *
 * An extremely genius very fast read/write lock based on atomics!
 */

#ifdef XT_THREAD_LOCK_INFO
xtPublic void xt_spinxslock_init(struct XTThread *XT_UNUSED(self), XTSpinXSLockPtr sxs, const char *name)
#else
xtPublic void xt_spinxslock_init(struct XTThread *XT_UNUSED(self), XTSpinXSLockPtr sxs)
#endif
{
	sxs->sxs_xlocked = 0;
	sxs->sxs_xwaiter = 0;
	sxs->sxs_rlock_count = 0;
	sxs->sxs_wait_count = 0;
#ifdef DEBUG
	sxs->sxs_locker = 0;
#endif
#ifdef XT_THREAD_LOCK_INFO
	sxs->sxs_name = name;
	xt_thread_lock_info_init(&sxs->sxs_lock_info, sxs);
#endif
}

xtPublic void xt_spinxslock_free(struct XTThread *XT_UNUSED(self), XTSpinXSLockPtr sxs)
{
#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_free(&sxs->sxs_lock_info);
#else
	(void) sxs;
#endif
}

xtPublic xtBool xt_spinxslock_xlock(XTSpinXSLockPtr sxs, xtBool try_lock, xtThreadID XT_NDEBUG_UNUSED(thd_id))
{
	register xtWord2 set;

	/* Wait for exclusive locker: */
	for (;;) {
		set = xt_atomic_tas2(&sxs->sxs_xlocked, 1);
		if (!set)
			break;
		if (try_lock)
			return FALSE;
		xt_yield();
	}

#ifdef DEBUG
	sxs->sxs_locker = thd_id;
#endif

	/* Wait for all the readers to wait! */
	while (sxs->sxs_wait_count < sxs->sxs_rlock_count) {
		sxs->sxs_xwaiter = 1;
		xt_yield(); //*
		/* This should not be required, because there is only one thread
		 * accessing this value. However, the lock fails if this
		 * is not done with an atomic op.
		 *
		 * This is because threads on other processors have the
		 * value in processor cache. So they do not
		 * notice that the value has been set to zero.
		 * They think it is still 1 and march through
		 * the barrier (sxs->sxs_xwaiter < sxs->sxs_xlocked) below.
		 *
		 * In the meantime, this X locker has gone on thinking
		 * all is OK.
		 */
		xt_atomic_tas2(&sxs->sxs_xwaiter, 0);
	}

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_add_owner(&sxs->sxs_lock_info);
#endif
	return OK;
}

xtPublic xtBool xt_spinxslock_slock(XTSpinXSLockPtr sxs)
{
	xt_atomic_inc2(&sxs->sxs_rlock_count);

	/* Wait as long as the locker is not waiting: */
	while (sxs->sxs_xwaiter < sxs->sxs_xlocked) {
		xt_atomic_inc2(&sxs->sxs_wait_count);
		while (sxs->sxs_xwaiter < sxs->sxs_xlocked) {
			xt_yield();
		}
		xt_atomic_dec2(&sxs->sxs_wait_count);
	}

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_add_owner(&sxs->sxs_lock_info);
#endif
	return OK;
}

xtPublic xtBool xt_spinxslock_unlock(XTSpinXSLockPtr sxs, xtBool xlocked)
{
	if (xlocked) {
#ifdef DEBUG
		ASSERT_NS(sxs->sxs_locker && sxs->sxs_xlocked);
		sxs->sxs_locker = 0;
#endif
		sxs->sxs_xlocked = 0;
	}
	else {
#ifdef DEBUG
		ASSERT_NS(sxs->sxs_rlock_count > 0);
#endif
		xt_atomic_dec2(&sxs->sxs_rlock_count);
	}

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_release_owner(&sxs->sxs_lock_info);
#endif
	return OK;
}

/*
 * -----------------------------------------------------------------------
 * FAST READ/WRITE LOCK (BASED ON FAST MUTEX)
 */

#ifdef XT_THREAD_LOCK_INFO
xtPublic void xt_xsmutex_init(struct XTThread *self, XTXSMutexLockPtr xsm, const char *name)
#else
xtPublic void xt_xsmutex_init(struct XTThread *self, XTXSMutexLockPtr xsm)
#endif
{
	xt_init_mutex_with_autoname(self, &xsm->xsm_lock);
	xt_init_cond(self, &xsm->xsm_cond);
	xt_init_cond(self, &xsm->xsm_cond_2);
	xsm->xsm_xlocker = 0;
	xsm->xsm_rlock_count = 0;
	xsm->xsm_wait_count = 0;
#ifdef DEBUG
	xsm->xsm_locker = 0;
#endif
#ifdef XT_THREAD_LOCK_INFO
	xsm->xsm_name = name;
	xt_thread_lock_info_init(&xsm->xsm_lock_info, xsm);
#endif
}

xtPublic void xt_xsmutex_free(struct XTThread *XT_UNUSED(self), XTXSMutexLockPtr xsm)
{
	xt_free_mutex(&xsm->xsm_lock);
	xt_free_cond(&xsm->xsm_cond);
	xt_free_cond(&xsm->xsm_cond_2);
#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_free(&xsm->xsm_lock_info);
#endif
}

xtPublic xtBool xt_xsmutex_xlock(XTXSMutexLockPtr xsm, xtThreadID thd_id)
{
	xt_lock_mutex_ns(&xsm->xsm_lock);

	/* Wait for exclusive locker: */
	while (xsm->xsm_xlocker) {
		if (!xt_timed_wait_cond_ns(&xsm->xsm_cond, &xsm->xsm_lock, 10000)) {
			xt_unlock_mutex_ns(&xsm->xsm_lock);
			return FAILED;
		}
	}

	/* GOTCHA: You would think this is not necessary...
	 * But is does not always work, if a normal insert is used.
	 * The reason is, I guess, on MMP the assignment is not
	 * always immediately visible to other processors, because they
	 * have old versions of this variable in there cache.
	 *
	 * But this is required, because the locking mechanism is based
	 * on:
	 * Locker: sets xlocker, tests rlock_count
	 * Reader: incs rlock_count, tests xlocker
	 *
	 * The test, in both cases, may not read stale values.
	 * volatile does not help, because this just turns compiler
	 * optimisations off.
	 */
	xt_atomic_set4(&xsm->xsm_xlocker, thd_id);

	/* Wait for all the reader to wait! */
	while (xsm->xsm_wait_count < xsm->xsm_rlock_count) {
		/* {RACE-WR_MUTEX} Here as well: */
		if (!xt_timed_wait_cond_ns(&xsm->xsm_cond, &xsm->xsm_lock, 100)) {
			xsm->xsm_xlocker = 0;
			xt_unlock_mutex_ns(&xsm->xsm_lock);
			return FAILED;
		}
	}

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_add_owner(&xsm->xsm_lock_info);
#endif
	return OK;
}

xtPublic xtBool xt_xsmutex_slock(XTXSMutexLockPtr xsm, xtThreadID XT_UNUSED(thd_id))
{
	xt_atomic_inc2(&xsm->xsm_rlock_count);

	/* Check if there could be an X locker: */
	if (xsm->xsm_xlocker) {
		/* I am waiting... */
		xt_lock_mutex_ns(&xsm->xsm_lock);
		xsm->xsm_wait_count++;
		/* Wake up the xlocker: */
		if (xsm->xsm_xlocker && xsm->xsm_wait_count == xsm->xsm_rlock_count) {
			if (!xt_broadcast_cond_ns(&xsm->xsm_cond)) {
				xsm->xsm_wait_count--;
				xt_unlock_mutex_ns(&xsm->xsm_lock);
				return FAILED;
			}
		}
		while (xsm->xsm_xlocker) {
			if (!xt_timed_wait_cond_ns(&xsm->xsm_cond_2, &xsm->xsm_lock, 10000)) {
				xsm->xsm_wait_count--;
				xt_unlock_mutex_ns(&xsm->xsm_lock);
				return FAILED;
			}
		}
		xsm->xsm_wait_count--;
		xt_unlock_mutex_ns(&xsm->xsm_lock);
	}

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_add_owner(&xsm->xsm_lock_info);
#endif
	return OK;
}

xtPublic xtBool xt_xsmutex_unlock(XTXSMutexLockPtr xsm, xtThreadID thd_id)
{
	if (xsm->xsm_xlocker == thd_id) {
		xsm->xsm_xlocker = 0;
		if (xsm->xsm_wait_count) {
			if (!xt_broadcast_cond_ns(&xsm->xsm_cond_2)) {
				xt_unlock_mutex_ns(&xsm->xsm_lock);
				return FAILED;
			}
		}
		else {
			/* Wake up any other X or shared lockers: */
			if (!xt_broadcast_cond_ns(&xsm->xsm_cond)) {
				xt_unlock_mutex_ns(&xsm->xsm_lock);
				return FAILED;
			}
		}
		xt_unlock_mutex_ns(&xsm->xsm_lock);
	}
	else {
		/* Taking the advice from {RACE-WR_MUTEX} I do the decrement
		 * after I have a lock!
		 */
		if (xsm->xsm_xlocker) {
			xt_lock_mutex_ns(&xsm->xsm_lock);
			xt_atomic_dec2(&xsm->xsm_rlock_count);
			if (xsm->xsm_xlocker && xsm->xsm_wait_count == xsm->xsm_rlock_count) {
				/* If the X locker is waiting for me,
				 * then allow him to continue. 
				 */
				if (!xt_broadcast_cond_ns(&xsm->xsm_cond)) {
					xt_unlock_mutex_ns(&xsm->xsm_lock);
					return FAILED;
				}
			}
			xt_unlock_mutex_ns(&xsm->xsm_lock);
		}
		else
			xt_atomic_dec2(&xsm->xsm_rlock_count);
	}

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_release_owner(&xsm->xsm_lock_info);
#endif
	return OK;
}

/*
 * -----------------------------------------------------------------------
 * ATOMIC READ/WRITE LOCK (BASED ON ATOMIC OPERATIONS)
 */

#ifdef XT_THREAD_LOCK_INFO
xtPublic void xt_atomicrwlock_init(struct XTThread *XT_UNUSED(self), XTAtomicRWLockPtr arw, const char *n)
#else
xtPublic void xt_atomicrwlock_init(struct XTThread *XT_UNUSED(self), XTAtomicRWLockPtr arw)
#endif
{
	arw->arw_reader_count = 0;
	arw->arw_xlock_set = 0;
#ifdef XT_THREAD_LOCK_INFO
	arw->arw_name = n;
	xt_thread_lock_info_init(&arw->arw_lock_info, arw);
#endif
}

#ifdef XT_THREAD_LOCK_INFO
xtPublic void xt_atomicrwlock_free(struct XTThread *, XTAtomicRWLockPtr arw)
#else
xtPublic void xt_atomicrwlock_free(struct XTThread *, XTAtomicRWLockPtr XT_UNUSED(arw))
#endif
{
#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_free(&arw->arw_lock_info);
#endif
}

xtPublic xtBool xt_atomicrwlock_xlock(XTAtomicRWLockPtr arw, xtBool try_lock, xtThreadID XT_NDEBUG_UNUSED(thr_id))
{
	register xtWord2 set;

	/* First get an exclusive lock: */
	for (;;) {
		set = xt_atomic_tas2(&arw->arw_xlock_set, 1);
		if (!set)
			break;
		if (try_lock)
			return FALSE;
		xt_yield();
	}

	/* Wait for the remaining readers: */
	while (arw->arw_reader_count)
		xt_yield();

#ifdef DEBUG
	arw->arw_locker = thr_id;
#endif

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_add_owner(&arw->arw_lock_info);
#endif
	return TRUE;
}

xtPublic xtBool xt_atomicrwlock_slock(XTAtomicRWLockPtr arw)
{
	register xtWord2 set;

	/* First get an exclusive lock: */
	for (;;) {
		set = xt_atomic_tas2(&arw->arw_xlock_set, 1);
		if (!set)
			break;
		xt_yield();
	}

	/* Add a reader: */
	xt_atomic_inc2(&arw->arw_reader_count);

	/* Release the xlock: */
	arw->arw_xlock_set = 0;

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_add_owner(&arw->arw_lock_info);
#endif
	return OK;
}

xtPublic xtBool xt_atomicrwlock_unlock(XTAtomicRWLockPtr arw, xtBool xlocked)
{
	if (xlocked) {
#ifdef DEBUG
		arw->arw_locker = 0;
#endif
		arw->arw_xlock_set = 0;
	}
	else
		xt_atomic_dec2(&arw->arw_reader_count);

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_release_owner(&arw->arw_lock_info);
#endif

	return OK;
}

/*
 * -----------------------------------------------------------------------
 * "SKEW" ATOMITC READ/WRITE LOCK (BASED ON ATOMIC OPERATIONS)
 *
 * This lock type favors writers. It only works if the proportion of readers
 * to writer is high.
 */

#ifdef XT_THREAD_LOCK_INFO
xtPublic void xt_skewrwlock_init(struct XTThread *XT_UNUSED(self), XTSkewRWLockPtr srw, const char *n)
#else
xtPublic void xt_skewrwlock_init(struct XTThread *XT_UNUSED(self), XTSkewRWLockPtr srw)
#endif
{
	srw->srw_reader_count = 0;
	srw->srw_xlock_set = 0;
#ifdef XT_THREAD_LOCK_INFO
	srw->srw_name = n;
	xt_thread_lock_info_init(&srw->srw_lock_info, srw);
#endif
}

#ifdef XT_THREAD_LOCK_INFO
xtPublic void xt_skewrwlock_free(struct XTThread *, XTSkewRWLockPtr srw)
#else
xtPublic void xt_skewrwlock_free(struct XTThread *, XTSkewRWLockPtr XT_UNUSED(srw))
#endif
{
#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_free(&srw->srw_lock_info);
#endif
}

xtPublic xtBool xt_skewrwlock_xlock(XTSkewRWLockPtr srw, xtBool try_lock, xtThreadID XT_NDEBUG_UNUSED(thr_id))
{
	register xtWord2 set;

	/* First get an exclusive lock: */
	for (;;) {
		set = xt_atomic_tas2(&srw->srw_xlock_set, 1);
		if (!set)
			break;
		if (try_lock)
			return FALSE;
		xt_yield();
	}

	/* Wait for the remaining readers: */
	while (srw->srw_reader_count)
		xt_yield();

#ifdef DEBUG
	srw->srw_locker = thr_id;
#endif

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_add_owner(&srw->srw_lock_info);
#endif
	return TRUE;
}

xtPublic xtBool xt_skewrwlock_slock(XTSkewRWLockPtr srw)
{
	/* Wait for an exclusive lock: */
	retry:
	for (;;) {
		if (!srw->srw_xlock_set)
			break;
		xt_yield();
	}

	/* Add a reader: */
	xt_atomic_inc2(&srw->srw_reader_count);

	/* Check for xlock again: */
	if (srw->srw_xlock_set) {
		xt_atomic_dec2(&srw->srw_reader_count);
		goto retry;
	}

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_add_owner(&srw->srw_lock_info);
#endif
	return OK;
}

xtPublic xtBool xt_skewrwlock_unlock(XTSkewRWLockPtr srw, xtBool xlocked)
{
	if (xlocked)
		srw->srw_xlock_set = 0;
	else
		xt_atomic_dec2(&srw->srw_reader_count);

#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_release_owner(&srw->srw_lock_info);
#endif
#ifdef DEBUG
	srw->srw_locker = 0;
#endif

	return OK;
}

/*
 * -----------------------------------------------------------------------
 * RECURSIVE R/W LOCK (allows X lockers to lock again)
 */

#ifdef XT_THREAD_LOCK_INFO
void xt_recursivemutex_init(XTThreadPtr self, XTRecursiveMutexPtr rm, const char *name)
{
	rm->rm_locker = NULL;
	rm->rm_lock_count = 0;
	xt_init_mutex(self, &rm->rm_mutex, name);
}
#else
xtPublic void xt_recursivemutex_init(XTThreadPtr self, XTRecursiveMutexPtr rm)
{
	rm->rm_locker = NULL;
	rm->rm_lock_count = 0;
	xt_init_mutex(self, &rm->rm_mutex);
}
#endif

xtPublic void xt_recursivemutex_free(XTRecursiveMutexPtr rm)
{
	xt_free_mutex(&rm->rm_mutex);
#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_free(&rm->rm_lock_info);
#endif
}

xtPublic void xt_recursivemutex_lock(XTThreadPtr self, XTRecursiveMutexPtr rm)
{
	if (self != rm->rm_locker) {
		xt_lock_mutex(self, &rm->rm_mutex);
		rm->rm_locker = self;
	}
	rm->rm_lock_count++;
}

xtPublic void xt_recursivemutex_unlock(XTThreadPtr self, XTRecursiveMutexPtr rm)
{
	ASSERT(self == rm->rm_locker);
	ASSERT(rm->rm_lock_count > 0);
	rm->rm_lock_count--;
	if (!rm->rm_lock_count) {
		rm->rm_locker = NULL;
		xt_unlock_mutex(self, &rm->rm_mutex);
	}
}

/*
 * -----------------------------------------------------------------------
 * RECURSIVE MUTEX (allows lockers to lock again)
 */

#ifdef XT_THREAD_LOCK_INFO
void xt_recurrwlock_init(struct XTThread *self, XTRecurRWLockPtr rrw, const char *name)
{
	rrw->rrw_locker = NULL;
	rrw->rrw_lock_count = 0;
	xt_init_rwlock(self, &rrw->rrw_lock, name);
}
#else
void xt_recurrwlock_init(struct XTThread *self, XTRecurRWLockPtr rrw)
{
	rrw->rrw_locker = NULL;
	rrw->rrw_lock_count = 0;
	xt_init_rwlock(self, &rrw->rrw_lock);
}
#endif

void xt_recurrwlock_free(XTRecurRWLockPtr rrw)
{
	xt_free_rwlock(&rrw->rrw_lock);
#ifdef XT_THREAD_LOCK_INFO
	xt_thread_lock_info_free(&rrw->rrw_lock_info);
#endif
}

void xt_recurrwlock_xlock(struct XTThread *self, XTRecurRWLockPtr rrw)
{
	if (self != rrw->rrw_locker) {
		xt_xlock_rwlock(self, &rrw->rrw_lock);
		rrw->rrw_locker = self;
	}
	rrw->rrw_lock_count++;
}

void xt_recurrwlock_slock(struct XTThread *self, XTRecurRWLockPtr rrw)
{
	xt_slock_rwlock(self, &rrw->rrw_lock);
}

void xt_recurrwlock_slock_ns(XTRecurRWLockPtr rrw)
{
	xt_slock_rwlock_ns(&rrw->rrw_lock);
}

void xt_recurrwlock_unxlock(struct XTThread *self, XTRecurRWLockPtr rrw)
{
	ASSERT(self == rrw->rrw_locker);
	ASSERT(rrw->rrw_lock_count > 0);
	rrw->rrw_lock_count--;
	if (!rrw->rrw_lock_count) {
		rrw->rrw_locker = NULL;
		xt_unlock_rwlock(self, &rrw->rrw_lock);
	}
}

void xt_recurrwlock_unslock(struct XTThread *self, XTRecurRWLockPtr rrw)
{
	xt_unlock_rwlock(self, &rrw->rrw_lock);
}

void xt_recurrwlock_unslock_ns(XTRecurRWLockPtr rrw)
{
	xt_unlock_rwlock_ns(&rrw->rrw_lock);
}

/*
 * -----------------------------------------------------------------------
 * UNIT TESTS
 */

#define JOB_MEMCPY			1
#define JOB_SLEEP			2
#define JOB_PRINT			3
#define JOB_INCREMENT		4
#define JOB_SNOOZE			5
#define JOB_DOUBLE_INC		6

#define LOCK_PTHREAD_RW		1
#define LOCK_PTHREAD_MUTEX	2
#define LOCK_RWMUTEX		3
#define LOCK_SPINLOCK		4
#define LOCK_FASTLOCK		5
#define LOCK_SPINXSLOCK		6
#define LOCK_XSMUTEX		7
#define LOCK_ATOMICRWLOCK	8
#define LOCK_SKEWRWLOCK		9

typedef struct XSLockTest {
	u_int			xs_interations;
	xtBool			xs_which_lock;
	xtBool			xs_which_job;
	xtBool			xs_debug_print;
	XTRWMutexRec	xs_lock;
	xt_rwlock_type	xs_plock;
	XTSpinLockRec	xs_spinlock;
	xt_mutex_type	xs_mutex;
	XTFastLockRec	xs_fastlock;
	XTSpinXSLockRec	xs_spinrwlock;
	XTXSMutexRec	xs_fastrwlock;
	XTAtomicRWLockRec xs_atomicrwlock;
	XTSkewRWLockRec xs_skewrwlock;
	int				xs_progress;
	xtWord4			xs_inc;
} XSLockTestRec, *XSLockTestPtr;

static void lck_free_thread_data(XTThreadPtr XT_UNUSED(self), void *XT_UNUSED(data))
{
}

static void lck_do_job(XTThreadPtr self, int job, XSLockTestPtr data, xtBool reader)
{
	char b1[1024], b2[1024];

	switch (job) {
		case JOB_MEMCPY:
                        memset(b1, 0, sizeof(b1));
                        memset(b2, 1, sizeof(b2));
			data->xs_inc++;
			break;
		case JOB_SLEEP:
			xt_sleep_milli_second(1);
			data->xs_inc++;
			break;
		case JOB_PRINT:
			printf("- %s got lock\n", self->t_name);
			xt_sleep_milli_second(10);
			data->xs_inc++;
			break;
		case JOB_INCREMENT:
			data->xs_inc++;
			break;
		case JOB_SNOOZE:
			xt_sleep_milli_second(10);
			data->xs_inc++;
			break;
		case JOB_DOUBLE_INC:
			if (reader) {
				if ((data->xs_inc & 1) != 0)
					printf("Noooo!\n");
			}
			else {
				data->xs_inc++;
				data->xs_inc++;
			}
			break;
	}
}

#if 0
static void *lck_run_dumper(XTThreadPtr self)
{
	int state = 0;

	while (state != 1) {
		sleep(1);
		if (state == 2) {
			xt_dump_trace();
			state = 0;
		}
	}
}
#endif

static void *lck_run_reader(XTThreadPtr self)
{
	XSLockTestRec	*data = (XSLockTestRec *) self->t_data;

	if (data->xs_debug_print)
		printf("- %s start\n", self->t_name);
	for (u_int i=0; i<data->xs_interations; i++) {
		if (data->xs_progress && ((i+1) % data->xs_progress) == 0)
			printf("- %s %d\n", self->t_name, i+1);
		if (data->xs_which_lock == LOCK_PTHREAD_RW) {
			xt_slock_rwlock_ns(&data->xs_plock);
			lck_do_job(self, data->xs_which_job, data, TRUE);
			xt_unlock_rwlock_ns(&data->xs_plock);
		}
		else if (data->xs_which_lock == LOCK_RWMUTEX) {
			xt_rwmutex_slock(&data->xs_lock, self->t_id);
			lck_do_job(self, data->xs_which_job, data, TRUE);
			xt_rwmutex_unlock(&data->xs_lock, self->t_id);
		}
		else if (data->xs_which_lock == LOCK_SPINXSLOCK) {
			xt_spinxslock_slock(&data->xs_spinrwlock);
			lck_do_job(self, data->xs_which_job, data, TRUE);
			xt_spinxslock_unlock(&data->xs_spinrwlock, FALSE);
		}
		else if (data->xs_which_lock == LOCK_XSMUTEX) {
			xt_xsmutex_slock(&data->xs_fastrwlock, self->t_id);
			lck_do_job(self, data->xs_which_job, data, TRUE);
			xt_xsmutex_unlock(&data->xs_fastrwlock, self->t_id);
		}
		else if (data->xs_which_lock == LOCK_ATOMICRWLOCK) {
			xt_atomicrwlock_slock(&data->xs_atomicrwlock);
			lck_do_job(self, data->xs_which_job, data, TRUE);
			xt_atomicrwlock_unlock(&data->xs_atomicrwlock, FALSE);
		}
		else if (data->xs_which_lock == LOCK_SKEWRWLOCK) {
			xt_skewrwlock_slock(&data->xs_skewrwlock);
			lck_do_job(self, data->xs_which_job, data, TRUE);
			xt_skewrwlock_unlock(&data->xs_skewrwlock, FALSE);
		}
		else
			ASSERT(FALSE);
	}
	if (data->xs_debug_print)
		printf("- %s stop\n", self->t_name);
	return NULL;
}

static void *lck_run_writer(XTThreadPtr self)
{
	XSLockTestRec	*data = (XSLockTestRec *) self->t_data;

	if (data->xs_debug_print)
		printf("- %s start\n", self->t_name);
	for (u_int i=0; i<data->xs_interations; i++) {
		if (data->xs_progress && ((i+1) % data->xs_progress) == 0)
			printf("- %s %d\n", self->t_name, i+1);
		if (data->xs_which_lock == LOCK_PTHREAD_RW) {
			xt_xlock_rwlock_ns(&data->xs_plock);
			lck_do_job(self, data->xs_which_job, data, FALSE);
			xt_unlock_rwlock_ns(&data->xs_plock);
		}
		else if (data->xs_which_lock == LOCK_RWMUTEX) {
			xt_rwmutex_xlock(&data->xs_lock, self->t_id);
			lck_do_job(self, data->xs_which_job, data, FALSE);
			xt_rwmutex_unlock(&data->xs_lock, self->t_id);
		}
		else if (data->xs_which_lock == LOCK_SPINXSLOCK) {
			xt_spinxslock_xlock(&data->xs_spinrwlock, FALSE, self->t_id);
			lck_do_job(self, data->xs_which_job, data, FALSE);
			xt_spinxslock_unlock(&data->xs_spinrwlock, TRUE);
		}
		else if (data->xs_which_lock == LOCK_XSMUTEX) {
			xt_xsmutex_xlock(&data->xs_fastrwlock, self->t_id);
			lck_do_job(self, data->xs_which_job, data, FALSE);
			xt_xsmutex_unlock(&data->xs_fastrwlock, self->t_id);
		}
		else if (data->xs_which_lock == LOCK_ATOMICRWLOCK) {
			xt_atomicrwlock_xlock(&data->xs_atomicrwlock, FALSE, self->t_id);
			lck_do_job(self, data->xs_which_job, data, FALSE);
			xt_atomicrwlock_unlock(&data->xs_atomicrwlock, TRUE);
		}
		else if (data->xs_which_lock == LOCK_SKEWRWLOCK) {
			xt_skewrwlock_xlock(&data->xs_skewrwlock, FALSE, self->t_id);
			lck_do_job(self, data->xs_which_job, data, FALSE);
			xt_skewrwlock_unlock(&data->xs_skewrwlock, TRUE);
		}
		else
			ASSERT(FALSE);
	}
	if (data->xs_debug_print)
		printf("- %s stop\n", self->t_name);
	return NULL;
}

static void lck_print_test(XSLockTestRec *data)
{
	switch (data->xs_which_lock) {
		case LOCK_PTHREAD_RW:
			printf("pthread read/write");
			break;
		case LOCK_PTHREAD_MUTEX:
			printf("pthread mutex");
			break;
		case LOCK_RWMUTEX:
			printf("fast read/write mutex");
			break;
		case LOCK_SPINLOCK:
			printf("spin mutex");
			break;
		case LOCK_FASTLOCK:
			printf("fast mutex");
			break;
		case LOCK_SPINXSLOCK:
			printf("spin read/write lock");
			break;
		case LOCK_XSMUTEX:
			printf("fast x/s mutex");
			break;
		case LOCK_ATOMICRWLOCK:
			printf("atomic read/write lock");
			break;
		case LOCK_SKEWRWLOCK:
			printf("skew read/write lock");
			break;
	}

	switch (data->xs_which_job) {
		case JOB_MEMCPY:
			printf(" MEMCPY 2K");
			break;
		case JOB_SLEEP:
			printf(" SLEEP 1/1000s");
			break;
		case JOB_PRINT:
			printf(" PRINT DEBUG");
			break;
		case JOB_INCREMENT:
			printf(" INCREMENT");
			break;
		case JOB_SNOOZE:
			printf(" SLEEP 1/100s");
			break;
	}
	
	printf(" %d interations", data->xs_interations);
}

static void *lck_run_mutex_locker(XTThreadPtr self)
{
	XSLockTestRec *data = (XSLockTestRec *) self->t_data;

	if (data->xs_debug_print)
		printf("- %s start\n", self->t_name);
	for (u_int i=0; i<data->xs_interations; i++) {
		if (data->xs_progress && ((i+1) % data->xs_progress) == 0)
			printf("- %s %d\n", self->t_name, i+1);
		if (data->xs_which_lock == LOCK_PTHREAD_MUTEX) {
			xt_lock_mutex_ns(&data->xs_mutex);
			lck_do_job(self, data->xs_which_job, data, FALSE);
			xt_unlock_mutex_ns(&data->xs_mutex);
		}
		else if (data->xs_which_lock == LOCK_SPINLOCK) {
			xt_spinlock_lock(&data->xs_spinlock);
			lck_do_job(self, data->xs_which_job, data, FALSE);
			xt_spinlock_unlock(&data->xs_spinlock);
		}
		else if (data->xs_which_lock == LOCK_FASTLOCK) {
			xt_fastlock_lock(&data->xs_fastlock, self);
			lck_do_job(self, data->xs_which_job, data, FALSE);
			xt_fastlock_unlock(&data->xs_fastlock, self);
		}
		else
			ASSERT(FALSE);
	}
	if (data->xs_debug_print)
		printf("- %s stop\n", self->t_name);
	return NULL;
}

typedef struct LockThread {
	xtThreadID		id;
	XTThreadPtr		ptr;
} LockThreadRec, *LockThreadPtr;

static void lck_reader_writer_test(XTThreadPtr self, XSLockTestRec *data, int reader_cnt, int writer_cnt)
{
	xtWord8			start;
	LockThreadPtr	threads;
	int				thread_cnt = reader_cnt + writer_cnt;
	char			buffer[40];

	//XTThreadPtr dumper = xt_create_daemon(self, "DUMPER");
	//xt_run_thread(self, dumper, lck_run_dumper);

	printf("READ/WRITE TEST: ");
	lck_print_test(data);
	printf(", %d readers, %d writers\n", reader_cnt, writer_cnt);
	threads = (LockThreadPtr) xt_malloc(self, thread_cnt * sizeof(LockThreadRec));

	for (int i=0; i<thread_cnt; i++) {
		sprintf(buffer, "%s%d", i < reader_cnt ? "READER-" : "WRITER-", i+1);
		threads[i].ptr = xt_create_daemon(self, buffer);
		threads[i].id = threads[i].ptr->t_id;
		xt_set_thread_data(threads[i].ptr, data, lck_free_thread_data);
	}

	start = xt_trace_clock();
	for (int i=0; i<reader_cnt; i++)
		xt_run_thread(self, threads[i].ptr, lck_run_reader);
	for (int i=reader_cnt; i<thread_cnt; i++)
		xt_run_thread(self, threads[i].ptr, lck_run_writer);

	for (int i=0; i<thread_cnt; i++)
		xt_wait_for_thread(threads[i].id, TRUE);
	printf("----- %d reader, %d writer time=%s\n", reader_cnt, writer_cnt, xt_trace_clock_diff(buffer, start));

	xt_free(self, threads);
	printf("TEST RESULT = %d\n", data->xs_inc);

	//xt_wait_for_thread(dumper, TRUE);
}

static void lck_mutex_lock_test(XTThreadPtr self, XSLockTestRec *data, int thread_cnt)
{
	xtWord8			start;
	LockThreadPtr	threads;
	char			buffer[40];

	printf("LOCK MUTEX TEST: ");
	lck_print_test(data);
	printf(", %d threads\n", thread_cnt);
	threads = (LockThreadPtr) xt_malloc(self, thread_cnt * sizeof(LockThreadRec));

	for (int i=0; i<thread_cnt; i++) {
		sprintf(buffer, "THREAD%d", i+1);
		threads[i].ptr = xt_create_daemon(self, buffer);
		threads[i].id = threads[i].ptr->t_id;
		xt_set_thread_data(threads[i].ptr, data, lck_free_thread_data);
	}

	start = xt_trace_clock();
	for (int i=0; i<thread_cnt; i++)
		xt_run_thread(self, threads[i].ptr, lck_run_mutex_locker);

	for (int i=0; i<thread_cnt; i++)
		xt_wait_for_thread(threads[i].id, TRUE);
	printf("----- %d threads time=%s\n", thread_cnt, xt_trace_clock_diff(buffer, start));

	xt_free(self, threads);
	printf("TEST RESULT = %d\n", data->xs_inc);
}

xtPublic void xt_unit_test_read_write_locks(XTThreadPtr self)
{
	XSLockTestRec	data;

	memset(&data, 0, sizeof(data));

	printf("TEST: xt_unit_test_read_write_locks\n");
	printf("size of XTXSMutexRec = %d\n", (int) sizeof(XTXSMutexRec));
	printf("size of pthread_cond_t = %d\n", (int) sizeof(pthread_cond_t));
	printf("size of pthread_mutex_t = %d\n", (int) sizeof(pthread_mutex_t));
	xt_rwmutex_init_with_autoname(self, &data.xs_lock);
	xt_init_rwlock_with_autoname(self, &data.xs_plock);
	xt_spinxslock_init_with_autoname(self, &data.xs_spinrwlock);
	xt_xsmutex_init_with_autoname(self, &data.xs_fastrwlock);
	xt_atomicrwlock_init_with_autoname(self, &data.xs_atomicrwlock);
	xt_skewrwlock_init_with_autoname(self, &data.xs_skewrwlock);

	/**
	data.xs_interations = 10;
	data.xs_which_lock = LOCK_RWMUTEX; // LOCK_PTHREAD_RW, LOCK_RWMUTEX, LOCK_SPINXSLOCK, LOCK_XSMUTEX
	data.xs_which_job = JOB_PRINT;
	data.xs_debug_print = TRUE;
	data.xs_progress = 0;
	lck_reader_writer_test(self, &data, 4, 0);
	lck_reader_writer_test(self, &data, 0, 2);
	lck_reader_writer_test(self, &data, 1, 1);
	lck_reader_writer_test(self, &data, 4, 2);
	**/

	/**
	data.xs_interations = 4000;
	data.xs_which_lock = LOCK_RWMUTEX; // LOCK_PTHREAD_RW, LOCK_RWMUTEX, LOCK_SPINXSLOCK, LOCK_XSMUTEX
	data.xs_which_job = JOB_SLEEP;
	data.xs_debug_print = TRUE;
	data.xs_progress = 200;
	lck_reader_writer_test(self, &data, 4, 0);
	lck_reader_writer_test(self, &data, 0, 2);
	lck_reader_writer_test(self, &data, 1, 1);
	lck_reader_writer_test(self, &data, 4, 2);
	**/

	// LOCK_PTHREAD_RW, LOCK_RWMUTEX, LOCK_SPINXSLOCK, LOCK_XSMUTEX, LOCK_ATOMICRWLOCK, LOCK_SKEWRWLOCK
	/**/
	data.xs_interations = 100000;
	data.xs_which_lock = LOCK_XSMUTEX;
	data.xs_which_job = JOB_DOUBLE_INC; // JOB_INCREMENT, JOB_DOUBLE_INC
	data.xs_debug_print = FALSE;
	data.xs_progress = 0;
	lck_reader_writer_test(self, &data, 10, 0);
	data.xs_which_lock = LOCK_XSMUTEX;
	lck_reader_writer_test(self, &data, 10, 0);
	//lck_reader_writer_test(self, &data, 0, 5);
	//lck_reader_writer_test(self, &data, 10, 0);
	//lck_reader_writer_test(self, &data, 10, 5);
	/**/

	/**/
	data.xs_interations = 10000;
	data.xs_which_lock = LOCK_XSMUTEX;
	data.xs_which_job = JOB_MEMCPY;
	data.xs_debug_print = FALSE;
	data.xs_progress = 0;
	lck_reader_writer_test(self, &data, 10, 0);
	data.xs_which_lock = LOCK_XSMUTEX;
	lck_reader_writer_test(self, &data, 10, 0);
	//lck_reader_writer_test(self, &data, 0, 5);
	//lck_reader_writer_test(self, &data, 10, 0);
	//lck_reader_writer_test(self, &data, 10, 5);
	/**/

	/**/
	data.xs_interations = 1000;
	data.xs_which_lock = LOCK_XSMUTEX;
	data.xs_which_job = JOB_SLEEP; // JOB_SLEEP, JOB_SNOOZE
	data.xs_debug_print = FALSE;
	data.xs_progress = 0;
	lck_reader_writer_test(self, &data, 10, 0);
	data.xs_which_lock = LOCK_XSMUTEX;
	lck_reader_writer_test(self, &data, 10, 0);
	/**/

	xt_rwmutex_free(self, &data.xs_lock);
	xt_free_rwlock(&data.xs_plock);
	xt_spinxslock_free(self, &data.xs_spinrwlock);
	xt_xsmutex_free(self, &data.xs_fastrwlock);
	xt_atomicrwlock_free(self, &data.xs_atomicrwlock);
	xt_skewrwlock_free(self, &data.xs_skewrwlock);
}

xtPublic void xt_unit_test_mutex_locks(XTThreadPtr self)
{
	XSLockTestRec	data;

	memset(&data, 0, sizeof(data));

	printf("TEST: xt_unit_test_mutex_locks\n");
	xt_spinlock_init_with_autoname(self, &data.xs_spinlock);
	xt_fastlock_init_with_autoname(self, &data.xs_fastlock);
	xt_init_mutex_with_autoname(self, &data.xs_mutex);

	/**/
	data.xs_interations = 10;
	data.xs_which_lock = LOCK_SPINLOCK; // LOCK_SPINLOCK, LOCK_PTHREAD_MUTEX, LOCK_FASTLOCK
	data.xs_which_job = JOB_PRINT;
	data.xs_debug_print = TRUE;
	data.xs_progress = 0;
	data.xs_inc = 0;
	lck_mutex_lock_test(self, &data, 2);
	/**/

	/**/
	data.xs_interations = 100000;
	data.xs_which_lock = LOCK_SPINLOCK; // LOCK_SPINLOCK, LOCK_PTHREAD_MUTEX, LOCK_FASTLOCK
	data.xs_which_job = JOB_INCREMENT;
	data.xs_debug_print = FALSE;
	data.xs_progress = 0;
	data.xs_inc = 0;
	lck_mutex_lock_test(self, &data, 10);
	/**/

	/**/
	data.xs_interations = 10000;
	data.xs_which_lock = LOCK_SPINLOCK; // LOCK_SPINLOCK, LOCK_PTHREAD_MUTEX, LOCK_FASTLOCK
	data.xs_which_job = JOB_MEMCPY;
	data.xs_debug_print = FALSE;
	data.xs_progress = 0;
	data.xs_inc = 0;
	lck_mutex_lock_test(self, &data, 10);
	/**/

	/**/
	data.xs_interations = 1000;
	data.xs_which_lock = LOCK_FASTLOCK; // LOCK_SPINLOCK, LOCK_PTHREAD_MUTEX, LOCK_FASTLOCK
	data.xs_which_job = JOB_SLEEP;
	data.xs_debug_print = FALSE;
	data.xs_progress = 0;
	data.xs_inc = 0;
	lck_mutex_lock_test(self, &data, 10);
	/**/

	/**/
	data.xs_interations = 100;
	data.xs_which_lock = LOCK_FASTLOCK; // LOCK_SPINLOCK, LOCK_PTHREAD_MUTEX, LOCK_FASTLOCK
	data.xs_which_job = JOB_SNOOZE;
	data.xs_debug_print = FALSE;
	data.xs_progress = 0;
	data.xs_inc = 0;
	lck_mutex_lock_test(self, &data, 10);
	/**/

	xt_spinlock_free(self, &data.xs_spinlock);
	xt_fastlock_free(self, &data.xs_fastlock);
	xt_free_mutex(&data.xs_mutex);
}

xtPublic void xt_unit_test_create_threads(XTThreadPtr self)
{
	XTThreadPtr		threads[10];

	printf("TEST: xt_unit_test_create_threads\n");
	printf("current max threads = %d, in use = %d\n", xt_thr_current_max_threads, xt_thr_current_thread_count);

	/* Create some threads: */
	threads[0] = xt_create_daemon(self, "test0");
	printf("thread = %d\n", threads[0]->t_id);
	threads[1] = xt_create_daemon(self, "test1");
	printf("thread = %d\n", threads[1]->t_id);
	threads[2] = xt_create_daemon(self, "test2");
	printf("thread = %d\n", threads[2]->t_id);
	threads[3] = xt_create_daemon(self, "test3");
	printf("thread = %d\n", threads[3]->t_id);
	threads[4] = xt_create_daemon(self, "test4");
	printf("thread = %d\n", threads[4]->t_id);
	printf("current max threads = %d, in use = %d\n", xt_thr_current_max_threads, xt_thr_current_thread_count);

	/* Max stays the same: */
	xt_free_thread(threads[3]);
	xt_free_thread(threads[2]);
	xt_free_thread(threads[1]);
	printf("current max threads = %d, in use = %d\n", xt_thr_current_max_threads, xt_thr_current_thread_count);

	/* Fill in the gaps: */
	threads[1] = xt_create_daemon(self, "test1");
	printf("thread = %d\n", threads[1]->t_id);
	threads[2] = xt_create_daemon(self, "test2");
	printf("thread = %d\n", threads[2]->t_id);
	threads[3] = xt_create_daemon(self, "test3");
	printf("thread = %d\n", threads[3]->t_id);
	printf("current max threads = %d, in use = %d\n", xt_thr_current_max_threads, xt_thr_current_thread_count);

	/* And add one: */
	threads[5] = xt_create_daemon(self, "test5");
	printf("thread = %d\n", threads[5]->t_id);
	printf("current max threads = %d, in use = %d\n", xt_thr_current_max_threads, xt_thr_current_thread_count);

	/* Max stays the same: */
	xt_free_thread(threads[3]);
	xt_free_thread(threads[2]);
	xt_free_thread(threads[1]);
	xt_free_thread(threads[4]);
	printf("current max threads = %d, in use = %d\n", xt_thr_current_max_threads, xt_thr_current_thread_count);

	/* Recalculate the max: */
	xt_free_thread(threads[5]);
	printf("current max threads = %d, in use = %d\n", xt_thr_current_max_threads, xt_thr_current_thread_count);

	/* Fill in the gaps: */
	threads[1] = xt_create_daemon(self, "test1");
	printf("thread = %d\n", threads[1]->t_id);
	threads[2] = xt_create_daemon(self, "test2");
	printf("thread = %d\n", threads[2]->t_id);
	threads[3] = xt_create_daemon(self, "test3");
	printf("thread = %d\n", threads[3]->t_id);
	printf("current max threads = %d, in use = %d\n", xt_thr_current_max_threads, xt_thr_current_thread_count);

	xt_free_thread(threads[3]);
	xt_free_thread(threads[2]);
	xt_free_thread(threads[1]);
	xt_free_thread(threads[0]);
	printf("current max threads = %d, in use = %d\n", xt_thr_current_max_threads, xt_thr_current_thread_count);
}

#ifdef UNUSED_CODE
int XTRowLocks::xt_release_locks(struct XTOpenTable *ot, xtRowID row, XTRowLockListPtr lock_list)
{
	if (ot->ot_temp_row_lock)
		xt_make_lock_permanent(ot, lock_list);

	if (!lock_list->bl_count)
		return XT_NO_LOCK;

	int					group, pgroup;
	XTXactDataPtr		xact;
	xtTableID			tab_id, ptab_id;
	XTPermRowLockPtr	plock;
	XTOpenTablePtr		pot = NULL;
	XTRowLocksPtr		row_locks;

	/* Do I have the lock? */
	group = row % XT_ROW_LOCK_COUNT;
	if (!(xact = tab_row_locks[group]))
		/* There is no lock: */
		return XT_NO_LOCK;

	if (xact != ot->ot_thread->st_xact_data)
		/* There is a lock but it does not belong to me! */
		return XT_NO_LOCK;

	tab_id = ot->ot_table->tab_id;
	plock = (XTPermRowLockPtr) &lock_list->bl_data[lock_list->bl_count * lock_list->bl_item_size];
	lock_list->rll_release_point = lock_list->bl_count;
	for (u_int i=0; i<lock_list->bl_count; i++) {
		plock--;

		pgroup = plock->pr_group;
		ptab_id = plock->pr_tab_id;

		if (ptab_id == tab_id)
			row_locks = this;
		else {
			if (pot) {
				if (pot->ot_table->tab_id == ptab_id)
					goto remove_lock;
				xt_db_return_table_to_pool_ns(pot);
				pot = NULL;
			}

			if (!xt_db_open_pool_table_ns(&pot, ot->ot_table->tab_db, tab_id)) {
				/* Should not happen, but just in case, we just don't
				 * remove the lock. We will probably end up with a deadlock
				 * somewhere.
				 */
				xt_log_and_clear_exception_ns();
				goto skip_remove_lock;
			}
			if (!pot)
				/* Can happen of the table has been dropped: */
				goto skip_remove_lock;

			remove_lock:
			row_locks = &pot->ot_table->tab_locks;
		}

#ifdef XT_TRACE_LOCKS
		xt_ttracef(xt_get_self(), "release lock group=%d\n", pgroup);
#endif
		row_locks->tab_row_locks[pgroup] = NULL;
		row_locks->tab_lock_perm[pgroup] = 0;
		skip_remove_lock:;

		lock_list->rll_release_point--;
		if (tab_id == ptab_id && group == pgroup)
			break;
	}

	if (pot) 
		xt_db_return_table_to_pool_ns(pot);
	return XT_PERM_LOCK;
}

xtBool XTRowLocks::xt_regain_locks(struct XTOpenTable *ot, int *lock_type, xtXactID *xn_id, XTRowLockListPtr lock_list)
{
	int					group;
	XTXactDataPtr		xact, my_xact;
	XTPermRowLockPtr	plock;
	xtTableID			tab_id;
	XTOpenTablePtr		pot = NULL;
	XTRowLocksPtr		row_locks = NULL;
	XTTableHPtr			tab = NULL;

	for (u_int i=lock_list->rll_release_point; i<lock_list->bl_count; i++) {
		plock = (XTPermRowLockPtr) &lock_list->bl_data[i * lock_list->bl_item_size];

		my_xact = ot->ot_thread->st_xact_data;
		group = plock->pr_group;
		tab_id = plock->pr_tab_id;

		if (tab_id == ot->ot_table->tab_id) {
			row_locks = this;
			tab = ot->ot_table;
		}
		else {
			if (pot) {
				if (tab_id == pot->ot_table->tab_id)
					goto gain_lock;
				xt_db_return_table_to_pool_ns(pot);
				pot = NULL;
			}

			if (!xt_db_open_pool_table_ns(&pot, ot->ot_table->tab_db, tab_id))
				return FAILED;
			if (!pot)
				goto no_gain_lock;
			
			gain_lock:
			tab = pot->ot_table;
			row_locks = &tab->tab_locks;
			no_gain_lock:;
		}

#ifdef XT_TRACE_LOCKS
		xt_ttracef(xt_get_self(), "regain lock group=%d\n", group);
#endif
		XT_TAB_ROW_WRITE_LOCK(&tab->tab_row_rwlock[group % XT_ROW_RWLOCKS], ot->ot_thread);
		if ((xact = row_locks->tab_row_locks[group])) {
			if (xact != my_xact) {
				*xn_id = xact->xd_start_xn_id;
				*lock_type = row_locks->tab_lock_perm[group] ? XT_PERM_LOCK : XT_TEMP_LOCK;
				goto done;
			}
		}
		else
			row_locks->tab_row_locks[group] = my_xact;
		row_locks->tab_lock_perm[group] = 1;
		XT_TAB_ROW_UNLOCK(&tab->tab_row_rwlock[group % XT_ROW_RWLOCKS], ot->ot_thread);
		lock_list->rll_release_point++;
	}
	*lock_type = XT_NO_LOCK;
	return OK;

	done:
	XT_TAB_ROW_UNLOCK(&tab->tab_row_rwlock[group % XT_ROW_RWLOCKS], ot->ot_thread);
	return OK;
}

#endif