Subject: concurrency management

From: [email-masked] (Jean Anderson)
Date: Wed, 9 Sep 92 13:43:35 -0700

Greetings,

I'm having problems with a sequence generator I'm using as a concurrency
management exercise.  A class stores the last id assigned for a key.  A 
subroutine dispenses unique values to multiple, competing processes all
requesting the same exact key. Processes should always get unique keys;
those that get lock timeouts should retry until they succeed.

[OK, I realize that normally a sequence generator should be coded as a 
stored procedure.  I do this exercise as a part of learning a new database. 
Once it all works reliably, I'm more confident I understand how to manage 
concurrency for that database.]

The results so far have been unexpected (code is below):

 o  Sometimes it works.  I watch two or three processes spin, retrying, and
    each comes back with a unique value.
 o  Sometimes processes come back with the same value. As a side effect, a 
    duplicate record has been appended. Putting a unique index in place does
    not keep a duplicate record from being inserted.
 o  Sometimes shared memory gets corrupted.

At any rate, I definitely don't have the hang of postgres yet. I grep'd around 
and didn't see the information I was looking for, so my apologies if these 
questions are already covered somewhere in the docs:

 -  If the retrieve/replace fails, I make a really dreadful assumption
    that deadlock has occurred and automatically retry.  What is the best
    way to specifically trap lock timeouts?

 -  Is there a command to lock a resource so processes cannot do dirty reads
    on a tuple/page/table until the blocker ends a transaction?

 -  Is there a way to reset the lock timeout? Retries take a looong time.

Maybe somebody has already done this exercise and has reliable code they'd be
willing to share?  My first stab at it is down below.

Oh, versions are....

   Postgres 4.0.1
   SUN 4/110
   SUN OS 4.1.1

Many thanks,

    -	jean
	[email-masked]
	Jean T. Anderson
	SAIC Open Systems Division

---------------------------< tst_deadlock.c >--------------------------------
/*    
 * NAME
 *	tst_deadlock -- exercise concurrency management.
 *
 * CONFIGURATION
 *	This assumes the existence of a keyid class with at least one keyname:
 *
 *	create keyid (keyname=char16, keyvalue=int4, lddate=abstime)
 *	append keyid (keyname="arid", keyvalue=1, lddate="now")
 *	define index keyid_keyname on keyid using btree (keyname char16_ops)
 *
 *	The code below hardcodes the following parameters (either add the
 *	parameters or change them to what you prefer):
 *
 *		db_name="geodemo"
 *		tbl_name="keyid"
 *		ctr_name="arid"
 *
 * USAGE
 *	tst_deadlock [-d] [-p port] [-h host]
 *
 *	Copied from
 *		$POSTGRESHOME/sample/pgstats.c
 *
 *	Add these lines to the Makefile in that directory:
 *
 *	tst_deadlock: tst_deadlock.o
 *		$(CC) $(CFLAGS) -o tst_deadlock tst_deadlock.o $(OD)/$(SAMPLIBS) $(LIBS)
 *
 */
#include <stdio.h>
#include <sys/types.h>
#include <sys/file.h>

#include "tmp/c.h"
#include "tmp/oid.h"
#include "tmp/libpq-fe.h"

/* just in case someone left memory debugging turned on... */
#undef palloc
#undef pfree

extern char     *PQhost;     /* machine on which the backend is running */
extern char     *PQport;     /* comm. port with the postgres backend. */

extern char *optarg;
extern int optind,opterr;
extern int getpid();
extern int db_getkey();

#define FAIL		-1
#define SUCCEED		0
#define DONE		1
#define NOT_DONE	0
#define RETRIES		5
#define MAX_SLEEP_TIME  11	/* Actually is one more */

main(argc,argv)
     int argc;
     char **argv;
{
	char	db_name[30];		/* database name */
	char	tbl_name[30];		/* name of key counter table */
	char	ctr_name[30];		/* key name */
	int	increment;		/* number of unique keys to get */
	int	key_value;		/* highest key id for program */
	int	d_flag=0, errflag=0;	/* debug and error flags */
	char	c;

	/* -------------------------------
	 *    process command line options
	 * -------------------------------
	 */
	while ((c = getopt(argc, argv, "h:p:d")) != EOF) {
		switch(c) {
			case 'h':
				PQhost = optarg;
				break;
			case 'p':
				PQport = optarg;
				break;
			case 'd':
				d_flag++;
				break;
			case '?' :
				errflag++;
		}
	}

	if (errflag) {
		fprintf(stderr, "usage: %s [-d] [-p port] [-h host]\n", *argv);
		exit (1);
	}

	/* ---------------------
	 * Hardcoded parameters!
	 * ---------------------
	 */
	bzero (db_name, sizeof(db_name));
	bzero (tbl_name, sizeof(tbl_name));
	bzero (ctr_name, sizeof(ctr_name));
	strncpy (db_name, "geodemo", sizeof(db_name));
	strncpy (tbl_name, "keyid", sizeof(tbl_name));
	strncpy (ctr_name, "arid", sizeof(ctr_name));
	increment=1;

	/*---------------------
	 * Open database
	 *---------------------
	 */
	if(d_flag)
		PQtrace();

	PQsetdb(db_name);
	fprintf(stderr, "Connected to database <%s>.\n", PQdb());

	if(db_getkey(tbl_name, ctr_name, increment, &key_value)==FAIL) {
		fprintf(stderr, "db_getkey failed\n");
		exit(-1);
	}
	else {
		fprintf(stderr, "Back from db_getkey, ");
		fprintf(stderr, "key for %s is %d.\n", 
			ctr_name, key_value);
	}

	/*-----------------
	 * Close Connection
	 *-----------------
	 */
	PQfinish();
	exit(0);
}

int
db_getkey(tname, cname, nkeys, kvalue)
char	*tname;		/* table name */
char	*cname;		/* counter name */
int	nkeys;		/* increment: number of keys to get */
int	*kvalue;	/* highest value reserved for calling app */
{
	char		DbQuery[512];
	PortalBuffer	*portalbuf;
	char		*res;
	int		ngroups, ntups, tupno, nflds, grpno;
	int		done, i;
	int		sleep_time, pid;

	*kvalue = FAIL;		/* initialize to an invalid number */

	pid=getpid();

	for (i=1, done = NOT_DONE ; (i < RETRIES) && (done == NOT_DONE) ; i++)
	{
		/* -----------------
		 * Begin Transaction
		 * -----------------
		 */
		res=PQexec("begin");
		if (*res == 'E') {
			fprintf(stderr, 
				"db_getkey: begin transaction failed with %s",
				++res);
			return(FAIL);
		}

		/* -----------------
		 * Retrieve keyvalue
		 * -----------------
		 */
		sprintf(DbQuery,
			"retrieve (%s.keyvalue) where %s.keyname=\"%s\"",
			tname, tname, cname);

		res=PQexec(DbQuery);     

			/* Should be a P if no error occurred */
		if (*res == 'E') {
			fprintf(stderr, "db_getkey: %s", ++res);
			PQexec("abort");
			return(FAIL);
		}
		else if (*res != 'P') {
			fprintf(stderr, 
				"db_getkey: retrieve failed--deadlock?\n");
			PQexec("abort");
			sleep_time=( (rand()+pid) % MAX_SLEEP_TIME);
			fprintf(stderr,  
			   "db_getkey: sleeping %d seconds before retry.\n", 
			   sleep_time);
			sleep(sleep_time);
			continue;
		}

		portalbuf = PQparray(++res);
		ngroups = PQngroups(portalbuf);

		for (grpno = 0; grpno < ngroups; grpno++) {
			if ((ntups = PQntuplesGroup(portalbuf, grpno)) != 1) {
				fprintf(stderr, 
					"db_getkey: duplicate tuple for %s\n",
					cname);
				PQexec("abort");
				return (FAIL);
			}

			for (tupno = 0; tupno < ntups; tupno++) {
				*kvalue = atoi(PQgetvalue(portalbuf, tupno, 0));
			}

		}

		/* ---------------
		 * Update keyvalue
		 * ---------------
		 */
		sprintf(DbQuery, 
		"replace %s (keyvalue=%s.keyvalue+%d) where %s.keyname=\"%s\"",
			tname, tname, nkeys, tname, cname);

		res=PQexec(DbQuery);     

			/* Should be a C if no error occurred */
		if (*res == 'E') {
			fprintf(stderr, "db_getkey: %s", ++res);
			PQexec("abort");
			return(FAIL);
		}
		else if (*res != 'C') {
			fprintf(stderr, 
				"db_getkey: replace failed--deadlock?\n");
			PQexec("abort");
			sleep_time=( (rand()+pid) % MAX_SLEEP_TIME);
			fprintf(stderr,  
			   "db_getkey: sleeping %d seconds before retry.\n", 
			   sleep_time);
			sleep(sleep_time);
			continue;
		}

		PQexec("end");		/* commit the change */
		if (*res == 'E') {
			fprintf(stderr,
				"db_getkey: end transaction failed with %s",
				++res);
			return(FAIL);
		}
		else {
			done=DONE;
			*kvalue += nkeys;	/* OK to set since replace OK */
		}
	}

	if(*kvalue == FAIL)		/* probably ran out of retries? */
		return(FAIL);
	else
	{
		fprintf(stderr, "db_getkey succeeded in %d tries.\n", i-1);
		return (SUCCEED);
	}
}
---------------------------< end tst_deadlock.c >--------------------------------