OpenGrok

/*	$NetBSD: pgfs_subs.c,v 1.5 2012/04/11 14:28:18 yamt Exp $	*/

/*-
 * Copyright (c)2010,2011 YAMAMOTO Takashi,
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */

/*
 * a file system server which stores the data in a PostgreSQL database.
 */

/*
 * we use large objects to store file contents.  there are a few XXXs wrt it.
 *
 * - large objects don't obey the normal transaction semantics.
 *
 * - we use large object server-side functions directly (instead of via the
 *   libpq large object api) because:
 *	- we want to use asynchronous (in the sense of PQsendFoo) operations
 *	  which is not available with the libpq large object api.
 *	- with the libpq large object api, there's no way to know details of
 *	  an error because PGresult is freed in the library without saving
 *	  PG_DIAG_SQLSTATE etc.
 */

#include <sys/cdefs.h>
#ifndef lint
__RCSID("$NetBSD: pgfs_subs.c,v 1.5 2012/04/11 14:28:18 yamt Exp $");
#endif /* not lint */

#include <assert.h>
#include <err.h>
#include <errno.h>
#include <puffs.h>
#include <inttypes.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <util.h>

#include <libpq-fe.h>
#include <libpq/libpq-fs.h>	/* INV_* */

#include "pgfs.h"
#include "pgfs_db.h"
#include "pgfs_debug.h"
#include "pgfs_waitq.h"
#include "pgfs_subs.h"

const char * const vtype_table[] = {
	[VREG] = "regular",
	[VDIR] = "directory",
	[VLNK] = "link",
};

static unsigned int
tovtype(const char *type)
{
	unsigned int i;

	for (i = 0; i < __arraycount(vtype_table); i++) {
		if (vtype_table[i] == NULL) {
			continue;
		}
		if (!strcmp(type, vtype_table[i])) {
			return i;
		}
	}
	assert(0);
	return 0;
}

static const char *
fromvtype(enum vtype vtype)
{

	if (vtype < __arraycount(vtype_table)) {
		assert(vtype_table[vtype] != NULL);
		return vtype_table[vtype];
	}
	return NULL;
}

/*
 * fileid_lock stuff below is to keep ordering of operations for a file.
 * it is a workaround for the lack of operation barriers in the puffs
 * protocol.
 *
 * currently we do this locking only for SETATTR, GETATTR, and WRITE as
 * they are known to be reorder-unsafe.  they are sensitive to the file
 * attributes, mainly the file size.  note that as the kernel issues async
 * SETATTR/WRITE requests, vnode lock doesn't prevent GETATTR from seeing
 * the stale attributes.
 *
 * we are relying on waiton/wakeup being a FIFO.
 */

struct fileid_lock_handle {
	TAILQ_ENTRY(fileid_lock_handle) list;
	fileid_t fileid;
	struct puffs_cc *owner;	/* diagnostic only */
	struct waitq waitq;
};

TAILQ_HEAD(, fileid_lock_handle) fileid_lock_list =
    TAILQ_HEAD_INITIALIZER(fileid_lock_list);
struct waitq fileid_lock_waitq = TAILQ_HEAD_INITIALIZER(fileid_lock_waitq);

/*
 * fileid_lock: serialize requests for the fileid.
 *
 * this function should be the first yieldable point in a puffs callback.
 */

struct fileid_lock_handle *
fileid_lock(fileid_t fileid, struct puffs_cc *cc)
{
	struct fileid_lock_handle *lock;

	TAILQ_FOREACH(lock, &fileid_lock_list, list) {
		if (lock->fileid == fileid) {
			DPRINTF("fileid wait %" PRIu64 " cc %p\n", fileid, cc);
			assert(lock->owner != cc);
			waiton(&lock->waitq, cc);	/* enter FIFO */
			assert(lock->owner == cc);
			return lock;
		}
	}
	lock = emalloc(sizeof(*lock));
	lock->fileid = fileid;
	lock->owner = cc;
	DPRINTF("fileid lock %" PRIu64 " cc %p\n", lock->fileid, cc);
	waitq_init(&lock->waitq);
	TAILQ_INSERT_HEAD(&fileid_lock_list, lock, list);
	return lock;
}

void
fileid_unlock(struct fileid_lock_handle *lock)
{

	DPRINTF("fileid unlock %" PRIu64 "\n", lock->fileid);
	assert(lock != NULL);
	assert(lock->owner != NULL);
	/*
	 * perform direct-handoff to the first waiter.
	 *
	 * a handoff is essential to keep the order of requests.
	 */
	lock->owner = wakeup_one(&lock->waitq);
	if (lock->owner != NULL) {
		return;
	}
	/*
	 * no one is waiting this fileid.
	 */
	TAILQ_REMOVE(&fileid_lock_list, lock, list);
	free(lock);
}

/*
 * timespec_to_pgtimestamp: create a text representation of timestamp which
 * can be recognized by the database server.
 *
 * it's caller's responsibility to free(3) the result.
 */

int
timespec_to_pgtimestamp(const struct timespec *tv, char **resultp)
{
	/*
	 * XXX is there any smarter way?
	 */
	char buf1[1024];
	char buf2[1024];
	struct tm tm_store;
	struct tm *tm;

	tm = gmtime_r(&tv->tv_sec, &tm_store);
	if (tm == NULL) {
		assert(errno != 0);
		return errno;
	}
	strftime(buf1, sizeof(buf1), "%Y%m%dT%H%M%S", tm);
	snprintf(buf2, sizeof(buf2), "%s.%ju", buf1,
	    (uintmax_t)tv->tv_nsec / 1000);
	*resultp = estrdup(buf2);
	return 0;
}

int
my_lo_truncate(struct Xconn *xc, int32_t fd, int32_t size)
{
	static struct cmd *c;
	int32_t ret;
	int error;

	CREATECMD(c, "SELECT lo_truncate($1, $2)", INT4OID, INT4OID);
	error = sendcmd(xc, c, fd, size);
	if (error != 0) {
		return error;
	}
	error = simplefetch(xc, INT4OID, &ret);
	if (error != 0) {
		if (error == EEXIST) {
			/*
			 * probably the insertion of the new-sized page
			 * caused a duplicated key error.  retry.
			 */
			DPRINTF("map EEXIST to EAGAIN\n");
			error = EAGAIN;
		}
		return error;
	}
	assert(ret == 0);
	return 0;
}

int
my_lo_lseek(struct Xconn *xc, int32_t fd, int32_t offset, int32_t whence,
    int32_t *retp)
{
	static struct cmd *c;
	int32_t ret;
	int error;

	CREATECMD(c, "SELECT lo_lseek($1, $2, $3)", INT4OID, INT4OID, INT4OID);
	error = sendcmd(xc, c, fd, offset, whence);
	if (error != 0) {
		return error;
	}
	error = simplefetch(xc, INT4OID, &ret);
	if (error != 0) {
		return error;
	}
	if (retp != NULL) {
		*retp = ret;
	}
	return 0;
}

int
my_lo_read(struct Xconn *xc, int32_t fd, void *buf, size_t size,
    size_t *resultsizep)
{
	static struct cmd *c;
	size_t resultsize;
	int error;

	CREATECMD(c, "SELECT loread($1, $2)", INT4OID, INT4OID);
	error = sendcmdx(xc, 1, c, fd, (int32_t)size);
	if (error != 0) {
		return error;
	}
	error = simplefetch(xc, BYTEA, buf, &resultsize);
	if (error != 0) {
		return error;
	}
	*resultsizep = resultsize;
	if (size != resultsize) {
		DPRINTF("shortread? %zu != %zu\n", size, resultsize);
	}
	return 0;
}

int
my_lo_write(struct Xconn *xc, int32_t fd, const void *buf, size_t size,
    size_t *resultsizep)
{
	static struct cmd *c;
	int32_t resultsize;
	int error;

	CREATECMD(c, "SELECT lowrite($1, $2)", INT4OID, BYTEA);
	error = sendcmd(xc, c, fd, buf, (int32_t)size);
	if (error != 0) {
		return error;
	}
	error = simplefetch(xc, INT4OID, &resultsize);
	if (error != 0) {
		if (error == EEXIST) {
			/*
			 * probably the insertion of the new data page
			 * caused a duplicated key error.  retry.
			 */
			DPRINTF("map EEXIST to EAGAIN\n");
			error = EAGAIN;
		}
		return error;
	}
	*resultsizep = resultsize;
	if (size != (size_t)resultsize) {
		DPRINTF("shortwrite? %zu != %zu\n", size, (size_t)resultsize);
	}
	return 0;
}

int
my_lo_open(struct Xconn *xc, Oid loid, int32_t mode, int32_t *fdp)
{
	static struct cmd *c;
	int error;

	CREATECMD(c, "SELECT lo_open($1, $2)", OIDOID, INT4OID);
	error = sendcmd(xc, c, loid, mode);
	if (error != 0) {
		return error;
	}
	return simplefetch(xc, INT4OID, fdp);
}

int
my_lo_close(struct Xconn *xc, int32_t fd)
{
#if 1
	/*
	 * do nothing.
	 *
	 * LO handles are automatically closed at the end of transactions.
	 * our transactions are small enough.
	 */
#else
	static struct cmd *c;
	int32_t ret;
	int error;

	CREATECMD(c, "SELECT lo_close($1)", INT4OID);
	error = sendcmd(xc, c, fd);
	if (error != 0) {
		return error;
	}
	error = simplefetch(xc, INT4OID, &ret);
	if (error != 0) {
		return error;
	}
	assert(ret == 0);
#endif
	return 0;
}

static int
lo_lookup_by_fileid(struct Xconn *xc, fileid_t fileid, Oid *idp)
{
	static struct cmd *c;
	static const Oid types[] = { OIDOID, };
	struct fetchstatus s;
	int error;

	CREATECMD(c, "SELECT loid FROM datafork WHERE fileid = $1", INT8OID);
	error = sendcmd(xc, c, fileid);
	if (error != 0) {
		return error;
	}
	fetchinit(&s, xc);
	error = FETCHNEXT(&s, types, idp);
	fetchdone(&s);
	DPRINTF("error %d\n", error);
	return error;
}

int
lo_open_by_fileid(struct Xconn *xc, fileid_t fileid, int mode, int *fdp)
{
	Oid loid;
	int fd;
	int error;

	error = lo_lookup_by_fileid(xc, fileid, &loid);
	if (error != 0) {
		return error;
	}
	error = my_lo_open(xc, loid, mode, &fd);
	if (error != 0) {
		return error;
	}
	*fdp = fd;
	return 0;
}

static int
getsize(struct Xconn *xc, fileid_t fileid, int *resultp)
{
	int32_t size;
	int fd;
	int error;

	error = lo_open_by_fileid(xc, fileid, INV_READ, &fd);
	if (error != 0) {
		return error;
	}
	error = my_lo_lseek(xc, fd, 0, SEEK_END, &size);
	if (error != 0) {
		return error;
	}
	error = my_lo_close(xc, fd);
	if (error != 0) {
		return error;
	}
	*resultp = size;
	return 0;
}

#define	GETATTR_TYPE	0x00000001
#define	GETATTR_NLINK	0x00000002
#define	GETATTR_SIZE	0x00000004
#define	GETATTR_MODE	0x00000008
#define	GETATTR_UID	0x00000010
#define	GETATTR_GID	0x00000020
#define	GETATTR_TIME	0x00000040
#define	GETATTR_ALL	\
	(GETATTR_TYPE|GETATTR_NLINK|GETATTR_SIZE|GETATTR_MODE| \
	GETATTR_UID|GETATTR_GID|GETATTR_TIME)

int
getattr(struct Xconn *xc, fileid_t fileid, struct vattr *va, unsigned int mask)
{
	char *type;
	long long atime_s;
	long long atime_us;
	long long ctime_s;
	long long ctime_us;
	long long mtime_s;
	long long mtime_us;
	long long btime_s;
	long long btime_us;
	uint64_t mode;
	long long uid;
	long long gid;
	long long nlink;
	long long rev;
	struct fetchstatus s;
	int error;

	if (mask == 0) {
		return 0;
	}
	/*
	 * unless explicitly requested, avoid fetching timestamps as they
	 * are a little more expensive than other simple attributes.
	 */
	if ((mask & GETATTR_TIME) != 0) {
		static struct cmd *c;
		static const Oid types[] = {
			TEXTOID,
			INT8OID,
			INT8OID,
			INT8OID,
			INT8OID,
			INT8OID,
			INT8OID,
			INT8OID,
			INT8OID,
			INT8OID,
			INT8OID,
			INT8OID,
			INT8OID,
			INT8OID,
		};

		CREATECMD(c, "SELECT type::text, mode, uid, gid, nlink, rev, "
		    "extract(epoch from date_trunc('second', atime))::int8, "
		    "extract(microseconds from atime)::int8, "
		    "extract(epoch from date_trunc('second', ctime))::int8, "
		    "extract(microseconds from ctime)::int8, "
		    "extract(epoch from date_trunc('second', mtime))::int8, "
		    "extract(microseconds from mtime)::int8, "
		    "extract(epoch from date_trunc('second', btime))::int8, "
		    "extract(microseconds from btime)::int8 "
		    "FROM file "
		    "WHERE fileid = $1", INT8OID);
		error = sendcmd(xc, c, fileid);
		if (error != 0) {
			return error;
		}
		fetchinit(&s, xc);
		error = FETCHNEXT(&s, types, &type, &mode, &uid, &gid, &nlink,
		    &rev,
		    &atime_s, &atime_us,
		    &ctime_s, &ctime_us,
		    &mtime_s, &mtime_us,
		    &btime_s, &btime_us);
	} else {
		static struct cmd *c;
		static const Oid types[] = {
			TEXTOID,
			INT8OID,
			INT8OID,
			INT8OID,
			INT8OID,
			INT8OID,
		};

		CREATECMD(c, "SELECT type::text, mode, uid, gid, nlink, rev "
		    "FROM file "
		    "WHERE fileid = $1", INT8OID);
		error = sendcmd(xc, c, fileid);
		if (error != 0) {
			return error;
		}
		fetchinit(&s, xc);
		error = FETCHNEXT(&s, types, &type, &mode, &uid, &gid, &nlink,
		    &rev);
	}
	fetchdone(&s);
	if (error != 0) {
		return error;
	}
	memset(va, 0xaa, sizeof(*va)); /* fill with garbage for debug */
	va->va_type = tovtype(type);
	free(type);
	va->va_mode = mode;
	va->va_uid = uid;
	va->va_gid = gid;
	if (nlink > 0 && va->va_type == VDIR) {
		nlink++; /* "." */
	}
	va->va_nlink = nlink;
	va->va_fileid = fileid;
	va->va_atime.tv_sec = atime_s;
	va->va_atime.tv_nsec = atime_us * 1000;
	va->va_ctime.tv_sec = ctime_s;
	va->va_ctime.tv_nsec = ctime_us * 1000;
	va->va_mtime.tv_sec = mtime_s;
	va->va_mtime.tv_nsec = mtime_us * 1000;
	va->va_birthtime.tv_sec = btime_s;
	va->va_birthtime.tv_nsec = btime_us * 1000;
	va->va_blocksize = LOBLKSIZE;
	va->va_gen = 1;
	va->va_filerev = rev;
	if ((mask & GETATTR_SIZE) != 0) {
		int size;

		size = 0;
		if (va->va_type == VREG || va->va_type == VLNK) {
			error = getsize(xc, fileid, &size);
			if (error != 0) {
				return error;
			}
		} else if (va->va_type == VDIR) {
			size = 100; /* XXX */
		}
		va->va_size = size;
	}
	/*
	 * XXX va_bytes: likely wrong due to toast compression.
	 * there's no cheap way to get the compressed size of LO.
	 */
	va->va_bytes = va->va_size;
	va->va_flags = 0;
	return 0;
}

int
update_mctime(struct Xconn *xc, fileid_t fileid)
{
	static struct cmd *c;

	CREATECMD(c,
	    "UPDATE file "
	    "SET mtime = current_timestamp, ctime = current_timestamp, "
		"rev = rev + 1 "
	    "WHERE fileid = $1", INT8OID);
	return simplecmd(xc, c, fileid);
}

int
update_atime(struct Xconn *xc, fileid_t fileid)
{
	static struct cmd *c;

	CREATECMD(c,
	    "UPDATE file SET atime = current_timestamp WHERE fileid = $1",
	    INT8OID);
	return simplecmd(xc, c, fileid);
}

int
update_mtime(struct Xconn *xc, fileid_t fileid)
{
	static struct cmd *c;

	CREATECMD(c,
	    "UPDATE file "
	    "SET mtime = current_timestamp, rev = rev + 1 "
	    "WHERE fileid = $1", INT8OID);
	return simplecmd(xc, c, fileid);
}

int
update_ctime(struct Xconn *xc, fileid_t fileid)
{
	static struct cmd *c;

	CREATECMD(c,
	    "UPDATE file SET ctime = current_timestamp WHERE fileid = $1",
	    INT8OID);
	return simplecmd(xc, c, fileid);
}

int
update_nlink(struct Xconn *xc, fileid_t fileid, int delta)
{
	static struct cmd *c;

	CREATECMD(c,
	    "UPDATE file "
	    "SET nlink = nlink + $1 "
	    "WHERE fileid = $2",
	    INT8OID, INT8OID);
	return simplecmd(xc, c, (int64_t)delta, fileid);
}

int
lookupp(struct Xconn *xc, fileid_t fileid, fileid_t *parent)
{
	static struct cmd *c;
	static const Oid types[] = { INT8OID, };
	struct fetchstatus s;
	int error;

	CREATECMD(c, "SELECT parent_fileid FROM dirent "
		"WHERE child_fileid = $1 LIMIT 1", INT8OID);
	error = sendcmd(xc, c, fileid);
	if (error != 0) {
		return error;
	}
	fetchinit(&s, xc);
	error = FETCHNEXT(&s, types, parent);
	fetchdone(&s);
	if (error != 0) {
		return error;
	}
	return 0;
}

int
mkfile(struct Xconn *xc, enum vtype vtype, mode_t mode, uid_t uid, gid_t gid,
    fileid_t *idp)
{
	static struct cmd *c;
	const char *type;
	int error;

	type = fromvtype(vtype);
	if (type == NULL) {
		return EOPNOTSUPP;
	}
	CREATECMD(c,
		"INSERT INTO file "
		"(fileid, type, mode, uid, gid, nlink, rev, "
		"atime, ctime, mtime, btime) "
		"VALUES(nextval('fileid_seq'), $1::filetype, $2, $3, $4, 0, 0, "
		"current_timestamp, "
		"current_timestamp, "
		"current_timestamp, "
		"current_timestamp) "
		"RETURNING fileid", TEXTOID, INT8OID, INT8OID, INT8OID);
	error = sendcmd(xc, c, type, (uint64_t)mode, (uint64_t)uid,
	    (uint64_t)gid);
	if (error != 0) {
		return error;
	}
	return simplefetch(xc, INT8OID, idp);
}

int
linkfile(struct Xconn *xc, fileid_t parent, const char *name, fileid_t child)
{
	static struct cmd *c;
	int error;

	CREATECMD(c,
		"INSERT INTO dirent "
		"(parent_fileid, name, child_fileid) "
		"VALUES($1, $2, $3)", INT8OID, TEXTOID, INT8OID);
	error = simplecmd(xc, c, parent, name, child);
	if (error != 0) {
		return error;
	}
	error = update_nlink(xc, child, 1);
	if (error != 0) {
		return error;
	}
	return update_mtime(xc, parent);
}

int
unlinkfile(struct Xconn *xc, fileid_t parent, const char *name, fileid_t child)
{
	static struct cmd *c;
	int error;

	/*
	 * in addition to the primary key, we check child_fileid as well here
	 * to avoid removing an entry which was appeared after our VOP_LOOKUP.
	 */
	CREATECMD(c,
		"DELETE FROM dirent "
		"WHERE parent_fileid = $1 AND name = $2 AND child_fileid = $3",
		INT8OID, TEXTOID, INT8OID);
	error = simplecmd(xc, c, parent, name, child);
	if (error != 0) {
		return error;
	}
	error = update_nlink(xc, child, -1);
	if (error != 0) {
		return error;
	}
	error = update_mtime(xc, parent);
	if (error != 0) {
		return error;
	}
	return update_ctime(xc, child);
}

int
mklinkfile(struct Xconn *xc, fileid_t parent, const char *name,
    enum vtype vtype, mode_t mode, uid_t uid, gid_t gid, fileid_t *idp)
{
	fileid_t fileid;
	int error;

	error = mkfile(xc, vtype, mode, uid, gid, &fileid);
	if (error != 0) {
		return error;
	}
	error = linkfile(xc, parent, name, fileid);
	if (error != 0) {
		return error;
	}
	if (idp != NULL) {
		*idp = fileid;
	}
	return 0;
}

int
mklinkfile_lo(struct Xconn *xc, fileid_t parent_fileid, const char *name,
    enum vtype vtype, mode_t mode, uid_t uid, gid_t gid, fileid_t *fileidp,
    int *loidp)
{
	static struct cmd *c;
	fileid_t new_fileid;
	int loid;
	int error;

	error = mklinkfile(xc, parent_fileid, name, vtype, mode, uid, gid,
	    &new_fileid);
	if (error != 0) {
		return error;
	}
	CREATECMD(c,
		"INSERT INTO datafork (fileid, loid) "
		"VALUES($1, lo_creat(-1)) "
		"RETURNING loid", INT8OID);
	error = sendcmd(xc, c, new_fileid);
	if (error != 0) {
		return error;
	}
	error = simplefetch(xc, OIDOID, &loid);
	if (error != 0) {
		return error;
	}
	if (fileidp != NULL) {
		*fileidp = new_fileid;
	}
	if (loidp != NULL) {
		*loidp = loid;
	}
	return 0;
}

int
cleanupfile(struct Xconn *xc, fileid_t fileid)
{
	static struct cmd *c;
	char *type;
	unsigned int vtype;
	int error;

	CREATECMD(c, "DELETE FROM file WHERE fileid = $1 AND nlink = 0 "
		"RETURNING type::text", INT8OID);
	error = sendcmd(xc, c, fileid);
	if (error != 0) {
		return error;
	}
	error = simplefetch(xc, TEXTOID, &type);
	if (error == ENOENT) {
		return 0; /* probably nlink > 0 */
	}
	if (error != 0) {
		return error;
	}
	vtype = tovtype(type);
	free(type);
	if (vtype == VREG || vtype == VLNK) {
		static struct cmd *c_datafork;
		int32_t ret;

		CREATECMD(c_datafork,
			"WITH loids AS (DELETE FROM datafork WHERE fileid = $1 "
			"RETURNING loid) SELECT lo_unlink(loid) FROM loids",
			INT8OID);
		error = sendcmd(xc, c_datafork, fileid);
		if (error != 0) {
			return error;
		}
		error = simplefetch(xc, INT4OID, &ret);
		if (error != 0) {
			return error;
		}
		if (ret != 1) {
			return EIO; /* lo_unlink failed */
		}
	}
	return 0;
}

/*
 * check_path: do locking and check to prevent a rename from creating loop.
 *
 * lock the dirents between child_fileid and the root directory.
 * if gate_fileid is appeared in the path, return EINVAL.
 * caller should ensure that child_fileid is of VDIR beforehand.
 *
 * we uses FOR SHARE row level locks as poor man's predicate locks.
 *
 * the following is an example to show why we need to lock the path.
 *
 * consider:
 * "mkdir -p /a/b/c/d/e/f && mkdir -p /1/2/3/4/5/6"
 * and then
 * thread 1 is doing "mv /a/b /1/2/3/4/5/6"
 * thread 2 is doing "mv /1/2 /a/b/c/d/e/f"
 *
 * a possible consequence:
 *	thread 1: check_path -> success
 *	thread 2: check_path -> success
 *	thread 1: modify directories -> block on row-level lock
 *	thread 2: modify directories -> block on row-level lock
 *			-> deadlock detected
 *			-> rollback and retry
 *
 * another possible consequence:
 *	thread 1: check_path -> success
 *	thread 1: modify directory entries -> success
 *	thread 2: check_path -> block on row-level lock
 *	thread 1: commit
 *	thread 2: acquire the lock and notices the row is updated
 *			-> serialization error
 *			-> rollback and retry
 *
 * XXX it might be better to use real serializable transactions,
 * which will be available for PostgreSQL 9.1
 */

int
check_path(struct Xconn *xc, fileid_t gate_fileid, fileid_t child_fileid)
{
	static struct cmd *c;
	fileid_t parent_fileid;
	struct fetchstatus s;
	int error;

	CREATECMD(c,
		"WITH RECURSIVE r AS "
		"( "
				"SELECT parent_fileid, cookie, child_fileid "
				"FROM dirent "
				"WHERE child_fileid = $1 "
			"UNION ALL "
				"SELECT d.parent_fileid, d.cookie, "
				"d.child_fileid "
				"FROM dirent AS d INNER JOIN r "
				"ON d.child_fileid = r.parent_fileid "
		") "
		"SELECT d.parent_fileid "
		"FROM dirent d "
		"JOIN r "
		"ON d.cookie = r.cookie "
		"FOR SHARE", INT8OID);
	error = sendcmd(xc, c, child_fileid);
	if (error != 0) {
		return error;
	}
	fetchinit(&s, xc);
	do {
		static const Oid types[] = { INT8OID, };

		error = FETCHNEXT(&s, types, &parent_fileid);
		if (error == ENOENT) {
			fetchdone(&s);
			return 0;
		}
		if (error != 0) {
			fetchdone(&s);
			return error;
		}
	} while (gate_fileid != parent_fileid);
	fetchdone(&s);
	return EINVAL;
}

int
isempty(struct Xconn *xc, fileid_t fileid, bool *emptyp)
{
	int32_t dummy;
	static struct cmd *c;
	int error;

	CREATECMD(c,
		"SELECT 1 FROM dirent "
		"WHERE parent_fileid = $1 LIMIT 1", INT8OID);
	error = sendcmd(xc, c, fileid);
	if (error != 0) {
		return error;
	}
	error = simplefetch(xc, INT4OID, &dummy);
	assert(error != 0 || dummy == 1);
	if (error == ENOENT) {
		*emptyp = true;
		error = 0;
	} else {
		*emptyp = false;
	}
	return error;
}