/*
 * Copyright (c) 2002-2006 Sendmail, Inc. and its suppliers.
 *      All rights reserved.
 *
 * By using this file, you agree to the terms and conditions set
 * forth in the LICENSE file which can be found at the top level of
 * the sendmail distribution.
 */

#include "sm/generic.h"
SM_RCSID("@(#)$Id: ibdb.c,v 1.127 2007/06/18 04:42:31 ca Exp $")
#include "sm/error.h"
#include "sm/fcntl.h"
#include "sm/memops.h"
#include "sm/stat.h"
#include "sm/time.h"
#include "sm/heap.h"
#include "sm/assert.h"
#include "sm/str.h"
#include "sm/cstr.h"
#include "sm/io.h"
#include "sm/fdset.h"
#include "sm/reccom.h"
#include "sm/ibdb.h"
#include "sm/rcb.h"
#include "sm/pthread.h"
#include "sm/fs.h"
#include "ibdb.h"
#include "sm/bhtable.h"
#include "ibdbc.h"

/* CONF: Make this configurable */
#define IBC_HT_SIZE	5013
#define IBC_HT_LIMIT	65536
#define SEQ_MIN_PURGE	10	/* minimum difference between seq and first */
#define SEQ_MIN_CLEAN	2	/* min diff. between cleanup runs */

#ifndef SM_IBDB_STATS
# define SM_IBDB_STATS	1
#endif

/*
**  ToDo:
**	Deal with entries that are too long for a single record.
**
**	What happens if the various directories are symbolic links?
**	Does renaming work? mkdir/rmdir will break things?
**	A startup script could create another IBDB directory
**	(ibdbn) which will be renamed to the default directory
**	on startup if necessary (i.e., W -> R, N -> W, use W).
**
**	Check disk space (at least) when creating a new file.
*/

/* specify an INCEDB backup on disk */
struct ibdb_ctx_S
{
	/* sm_magic? */
	uint		 ibdb_version;	/* version of this ibdb */
	sm_file_T	*ibdb_fp;	/* current file */
	sm_str_P	 ibdb_path;	/* pathname of file */
	sm_str_P	 ibdb_name_rm;	/* pathname of file to remove */
	char		*ibdb_dir;	/* directory of file */
	char		*ibdb_name;	/* basename of file */
	char		*ibdb_base_dir;	/* name of base directory */
	int			 ibdb_mode;	/* file mode */
	uint32_t	 ibdb_first;	/* first sequence number */
	uint32_t	 ibdb_sequence;	/* current sequence number */
	size_t		 ibdb_maxsize;	/* maximum size of one file */
	size_t		 ibdb_cursize;	/* current size of file */
	time_t		 ibdb_created;	/* creation time */
#if 0
	time_t		 ibdb_touched;	/* last commit; not (yet) used */
#endif
	uint		 ibdb_changes;	/* number of changes since last commit */
#if SM_IBDB_STATS

	/*
	**  This could be used to remove previous files if both counters
	**  are zero when a new file is created in ibdb_next().
	**  It could also be used to decide when to "rollover", i.e., when
	**  these counters are zero and cursize is "close" to maxsize.
	*/

	uint		 ibdb_opentas;	/* outstanding transactions */
	uint		 ibdb_openrcpts;	/* outstanding recipients */
	ulong		 ibdb_commits;
#endif /* SM_IBDB_STATS */

	/* locks entire ibdb_ctx... */
	pthread_mutex_t	 ibdb_mutex;

	/* Pool for reuse. See include/sm/edb.h */
	ibdb_req_hd_T	 ibdb_reql_pool;
	uint32_t	 ibdb_cleanrun;	/* seq. # of last cleanup run */
	bht_P		 ibdb_ct;

	fs_ctx_P	 ibdb_fs_ctx;
	int		 ibdb_fs_idx;

	uchar		*ibdb_fp_buf;	/* buffer for fp if necessary */
};

/*
**  IBDB_REQ_NEW -- return new ibdb request (allocated or from pool)
**
**	Parameters:
**		ibdb_ctx -- IBDB context
**		pedb_req -- ibdb req (output)
**
**	Returns:
**		usual sm_error code; ENOMEM
**
**	Locking:
**		ibdb_ctx (reql_pool) must be locked by caller
**
**	Last code review: 2005-03-31 18:29:33
**	Last code change:
*/

static sm_ret_T
ibdb_req_new(ibdb_ctx_P ibdb_ctx, ibdb_req_P *pedb_req)
{
	ibdb_req_P ibdb_req;

	/* Add a parameter for locking?? It can be done locally... */
	SM_REQUIRE(pedb_req != NULL);
	SM_REQUIRE(ibdb_ctx != NULL);
	if (!IBDBREQL_EMPTY(&ibdb_ctx->ibdb_reql_pool)) {
		ibdb_req = IBDBREQL_FIRST(&ibdb_ctx->ibdb_reql_pool);
		IBDBREQL_REMOVE(&ibdb_ctx->ibdb_reql_pool);
	}
	else {
		ibdb_req = (ibdb_req_P) sm_zalloc(sizeof(*ibdb_req));
		if (NULL == ibdb_req)
			return sm_error_temp(SM_EM_IBDB, ENOMEM);
	}
	*pedb_req = ibdb_req;
	return SM_SUCCESS;
}

/*
**  IBDB_REQ_REL -- release ibdb request: put it into pool, free allocated
**		substructures/fields
**
**	Parameters:
**		ibdb_ctx -- IBDB context
**		ibdb_req -- ibdb req
**
**	Returns:
**		SM_SUCCESS
**
**	Locking:
**		ibdb_ctx (reql_pool) must be locked by caller
**
**	Last code review: 2005-03-31 18:36:28
**	Last code change:
*/

static sm_ret_T
ibdb_req_rel(ibdb_ctx_P ibdb_ctx, ibdb_req_P ibdb_req)
{
	if (NULL == ibdb_req)
		return SM_SUCCESS;
	SM_REQUIRE(ibdb_ctx != NULL);

	/* Add a parameter for locking?? It can be done locally... */
	IBDBREQL_PRE(&ibdb_ctx->ibdb_reql_pool, ibdb_req);
	SM_STR_FREE(ibdb_req->ibdb_req_addr_pa);
	SM_CSTR_FREE(ibdb_req->ibdb_req_cdb_id);

	/*
	**  Clean out?
	**  sm_memzero(ibdb_req, sizeof(*ibdb_req));
	**  can't be used since it would destroy the linked list.
	*/

	return SM_SUCCESS;
}

/*
**  IBDB_REQ_FREE -- free ibdb request
**
**	Parameters:
**		ibdb_req -- ibdb req
**
**	Returns:
**		SM_SUCCESS
**
**	Last code review: 2005-03-31 18:30:29
**	Last code change:
*/

static sm_ret_T
ibdb_req_free(ibdb_req_P ibdb_req)
{
	if (NULL == ibdb_req)
		return SM_SUCCESS;
	SM_STR_FREE(ibdb_req->ibdb_req_addr_pa);
	SM_CSTR_FREE(ibdb_req->ibdb_req_cdb_id);
	sm_free_size(ibdb_req, sizeof(*ibdb_req));
	return SM_SUCCESS;
}

/*
**  IBDB_REQL_FREE -- free ibdb request list
**
**	Parameters:
**		ibdb_ctx -- IBDB context
**
**	Returns:
**		SM_SUCCESS
**
**	Locking:
**		ibdb_ctx (reql_pool) must be locked by caller
**
**	Last code review: 2005-03-31 22:35:30
**	Last code change:
*/

static sm_ret_T
ibdb_reql_free(ibdb_ctx_P ibdb_ctx)
{
	ibdb_req_P ibdb_req;

	SM_REQUIRE(ibdb_ctx != NULL);
	while (!IBDBREQL_EMPTY(&ibdb_ctx->ibdb_reql_pool)) {
		ibdb_req = IBDBREQL_FIRST(&ibdb_ctx->ibdb_reql_pool);
		IBDBREQL_REMOVE(&ibdb_ctx->ibdb_reql_pool);
		(void) ibdb_req_free(ibdb_req);
	}
	return SM_SUCCESS;
}

/*
**  IBDB_REQ_CANCEL -- cancel ibdb requests
**
**	Parameters:
**		ibdb_ctx -- IBDB context
**		ibdb_req_hd -- head of request list for IBDB to cancel
**
**	Returns:
**		SM_SUCCESS except for (un)lock errors
**
**	Locking:
**		ibdb_ctx (reql_pool) is locked during operation.
**
**	Last code review: 2005-03-31 22:37:35
**	Last code change:
*/

sm_ret_T
ibdb_req_cancel(ibdb_ctx_P ibdb_ctx, ibdb_req_hd_P ibdb_req_hd)
{
	sm_ret_T ret;
	int r;
	ibdb_req_P ibdb_req;

	if (IBDBREQL_EMPTY(ibdb_req_hd))
		return SM_SUCCESS;
	SM_REQUIRE(ibdb_ctx != NULL);
	r = pthread_mutex_lock(&ibdb_ctx->ibdb_mutex);
	SM_LOCK_OK(r);
	if (r != 0) {
		/* LOG? */
		return sm_error_perm(SM_EM_IBDB, r);
	}
	ret = SM_SUCCESS;

	/* Remove request list */
	while (!IBDBREQL_EMPTY(ibdb_req_hd)) {
		ibdb_req = IBDBREQL_FIRST(ibdb_req_hd);
		IBDBREQL_REMOVE(ibdb_req_hd);
		(void) ibdb_req_rel(ibdb_ctx, ibdb_req); /* always ok... */
	}
	r = pthread_mutex_unlock(&ibdb_ctx->ibdb_mutex);
	SM_ASSERT(0 == r);
	if (r != 0 && sm_is_success(ret))
		ret = sm_error_perm(SM_EM_IBDB, r);
	return ret;
}

/*
**  IBDB_CTX_FREE -- free IBDB context
**
**	Parameters:
**		ibdb_ctx -- ibdb context
**
**	Returns:
**		none.
**
**	Last code review: 2005-03-31 22:51:10
**	Last code change:
*/

static void
ibdb_ctx_free(ibdb_ctx_P ibdb_ctx)
{
	if (NULL == ibdb_ctx)
		return;
	SM_FREE(ibdb_ctx->ibdb_name);
	SM_FREE(ibdb_ctx->ibdb_fp_buf);
	SM_FREE(ibdb_ctx->ibdb_base_dir);
	SM_STR_FREE(ibdb_ctx->ibdb_path);
	SM_STR_FREE(ibdb_ctx->ibdb_name_rm);
	bht_destroy(ibdb_ctx->ibdb_ct, NULL, NULL);
	sm_free_size(ibdb_ctx, sizeof(*ibdb_ctx));
	return;
}

/*
**  IBDB_DIR_CRT -- create path for IBDB directory
**
**	Parameters:
**		base_dir -- name of base directory (can be NULL)
**		flags -- flags
**		pdir -- (pointer to) directory
**
**	Returns:
**
**	Last code review:
**	Last code change:
*/

sm_ret_T
ibdb_dir_crt(const char *base_dir, uint flags, char **pdir)
{
	size_t len;
	char *dir;

	SM_REQUIRE(pdir != NULL);
	len = IBDB_DIR_LEN;
	if (base_dir != NULL && *base_dir != '\0')
		len += strlen(base_dir) + 1;
	dir = (char *) sm_zalloc(len);
	if (dir != NULL && base_dir != NULL && *base_dir != '\0')
		strlcpy(dir, base_dir, len);
	if (dir != NULL) {
		if (IBDB_IS_OPEN_FLAG(flags, IBDB_OFL_WRITE))
			strlcat(dir, IBDB_WR_DIR, len);
		else if (IBDB_IS_OPEN_FLAG(flags, IBDB_OFL_RCVR))
			strlcat(dir, IBDB_REC_DIR, len);
		else if (IBDB_IS_OPEN_FLAG(flags, IBDB_OFL_CLEAN))
			strlcat(dir, IBDB_CL_DIR, len);
		else
			dir = ".";	/* Really?? abort instead? */
	}
	*pdir = dir;
	return SM_SUCCESS;
}

/*
**  IBDB_CTX_NEW -- allocate new IBDB context
**
**	Parameters:
**		base_dir -- name of base directory (can be NULL)
**		name -- base name, will be extended by seq
**		seq -- sequence number
**		flags -- flags
**
**	Returns:
**		NULL on error (out of memory)
**		otherwise pointer to new IBDB context
**
**	Last code review: 2005-03-31 22:50:37
**	Last code change: 2005-03-31 22:40:47
*/

static ibdb_ctx_P
ibdb_ctx_new(const char *base_dir, const char *name, uint32_t seq, uint flags)
{
	size_t l;
	ibdb_ctx_P ibdb_ctx;

	SM_ASSERT(name != NULL);

	ibdb_ctx = (ibdb_ctx_P) sm_zalloc(sizeof(*ibdb_ctx));
	if (NULL == ibdb_ctx)
		return NULL;
	ibdb_ctx->ibdb_ct = bht_new(IBC_HT_SIZE, IBC_HT_LIMIT);
	if (NULL == ibdb_ctx->ibdb_ct)
		goto error;
	l = 0;

	if (base_dir != NULL && *base_dir != '\0') {
		size_t lb;

		lb = strlen(base_dir) + 1;
		ibdb_ctx->ibdb_base_dir = (char *) sm_malloc(lb);
		if (NULL == ibdb_ctx->ibdb_base_dir)
			goto error;
		strlcpy(ibdb_ctx->ibdb_base_dir, base_dir, lb);
	}

	IBDB_DIR(ibdb_ctx->ibdb_dir, ibdb_ctx->ibdb_base_dir, flags);
	if (NULL == ibdb_ctx->ibdb_dir)
		goto error;

	l += strlen(name) + 1;
	ibdb_ctx->ibdb_name = (char *) sm_malloc(l);
	if (NULL == ibdb_ctx->ibdb_name)
		goto error;
	strlcpy(ibdb_ctx->ibdb_name, name, l);

	l += strlen(ibdb_ctx->ibdb_dir) + 1 + UINT32_LEN;
	ibdb_ctx->ibdb_path = sm_str_new(NULL, l, l);
	if (NULL == ibdb_ctx->ibdb_path)
		goto error;
	crt_ibdb_path(ibdb_ctx->ibdb_path, ibdb_ctx->ibdb_dir, name, seq);

	++l;
	ibdb_ctx->ibdb_name_rm = sm_str_new(NULL, l, l);
	if (NULL == ibdb_ctx->ibdb_name_rm)
		goto error;

	ibdb_ctx->ibdb_sequence = seq;
	ibdb_ctx->ibdb_first = seq;
	return ibdb_ctx;

  error:
	ibdb_ctx_free(ibdb_ctx);
	return NULL;
}

/*
**  IBDB_NEXT -- switch to next IBDB file
**
**	Parameters:
**		ibdb_ctx -- ibdb context
**
**	Returns:
**		usual sm_error code; I/O errors (flush, fsync, ...) et.al.
**
**	Side Effects: close fp, open next file, change free space, write
**		version number to file
**		does not undo any of these on error (that could occur
**		(almost) anywhere in the sequence)
**
**	Last code review: 2005-03-31 23:13:10
**	Last code change: 2005-11-20 17:27:27
*/

static sm_ret_T
ibdb_next(ibdb_ctx_P ibdb_ctx)
{
	sm_ret_T ret;
	sm_file_T *fp;

	SM_ASSERT(ibdb_ctx != NULL);
	ret = sm_io_flush(ibdb_ctx->ibdb_fp);
	if (sm_is_err(ret))
		return ret;
	ret = fsync(sm_io_fileno(ibdb_ctx->ibdb_fp));
	if (sm_is_err(ret))
		return ret;
	ret = sm_io_close(ibdb_ctx->ibdb_fp, SM_IO_CF_NONE);
	ibdb_ctx->ibdb_fp = NULL;
	if (sm_is_err(ret))
		return ret;

	if (ibdb_ctx->ibdb_fs_ctx != NULL) {
		ulong kbfree;

		ret = fs_chgfree(ibdb_ctx->ibdb_fs_ctx, ibdb_ctx->ibdb_fs_idx,
			(long) ((0L - (long) ibdb_ctx->ibdb_cursize) / ONEKB),
			&kbfree);
	}

	ibdb_ctx->ibdb_sequence++;
	sm_str_clr(ibdb_ctx->ibdb_path);
	crt_ibdb_path(ibdb_ctx->ibdb_path, ibdb_ctx->ibdb_dir,
		ibdb_ctx->ibdb_name, ibdb_ctx->ibdb_sequence);
	ret = sm_io_open(SmStStdio, sm_str_getdata(ibdb_ctx->ibdb_path),
			ibdb_ctx->ibdb_mode, &fp, SM_IO_WHAT_END);
	if (sm_is_err(ret))
		return ret;
	if (NULL != ibdb_ctx->ibdb_fp_buf) {
		ret = sm_io_setvbuf(fp, ibdb_ctx->ibdb_fp_buf, SM_IO_FBF,
				IBDB_BUF_SIZE);
		if (sm_is_err(ret))
			goto error;
	}

	/* only if writing? */
	ibdb_ctx->ibdb_version = IBDB_VERSION;
	ibdb_ctx->ibdb_fp = fp;
	ibdb_ctx->ibdb_cursize = 0;
#if 0
	ibdb_ctx->ibdb_maxsize = ?;
	ibdb_ctx->ibdb_created = now;
	ibdb_ctx->ibdb_touched = now;
#endif /* 0 */

	/* first: record type */
	ret = sm_io_fputuint32(fp, RT_IBDB_VERS_R);
	if (sm_is_err(ret))
		goto error;
	ret = sm_io_fputv(fp,
		SM_RCBV_INT, RT_IBDB_VERS_N, ibdb_ctx->ibdb_version,
		SM_RCBV_END);
	if (sm_is_err(ret))
		goto error;
	ret = sm_io_falign(fp, IBDB_REC_SIZE);
	if (sm_is_err(ret))
		goto error;

  error:
	/* no cleanup... many side effects possible! */
	return ret;
}

/*
**  IBDB_COMMIT -- commit ibdb backup to disk
**
**	Parameters:
**		ibdb_ctx -- ibdb context
**
**	Returns:
**		usual sm_error code; I/O errors (flush, fsync), (un)lock
**
**	Side Effects: flush/fsync file
**
**	Last code review: 2005-04-12 15:56:25
**	Last code change: 2005-04-12 15:56:14
*/

sm_ret_T
ibdb_commit(ibdb_ctx_P ibdb_ctx)
{
	sm_ret_T ret;
	int r;

	SM_ASSERT(ibdb_ctx != NULL);
	r = pthread_mutex_lock(&ibdb_ctx->ibdb_mutex);
	SM_LOCK_OK(r);
	if (r != 0) {
		/* LOG? */
		return sm_error_perm(SM_EM_IBDB, r);
	}

	if (ibdb_ctx->ibdb_changes > 0) {
		ret = sm_io_flush(ibdb_ctx->ibdb_fp);
		if (sm_is_err(ret))
			goto error;
#if HAVE_FDATASYNC
		r = fdatasync(sm_io_fileno(ibdb_ctx->ibdb_fp));
#else
		r = fsync(sm_io_fileno(ibdb_ctx->ibdb_fp));
#endif
		if (r != 0) {
			ret = sm_error_perm(SM_EM_IBDB, r);
			goto error;
		}
#if SM_IBDB_STATS
		++ibdb_ctx->ibdb_commits;
#endif
#if 0
		ibdb_ctx->ibdb_touched = now;
#endif
		ibdb_ctx->ibdb_changes = 0;
	}
	else
		ret = SM_SUCCESS;

  error:
	r = pthread_mutex_unlock(&ibdb_ctx->ibdb_mutex);
	SM_ASSERT(0 == r);
	if (r != 0 && sm_is_success(ret))
		ret = sm_error_perm(SM_EM_IBDB, r);
	return ret;
}

/*
**  IBDB_OPEN -- open ibdb backup on disk
**
**	Parameters:
**		base_dir -- name of base directory (can be NULL)
**		name -- base name, will be extended by seq
**		mode -- open mode
**		seq -- initial sequence number (> 0)
**		size -- maximum size (> 0)
**		flags -- flags
**		fs_ctx -- file system context (can be NULL)
**		pibdb_ctx -- (pointer to) ibdb context (output)
**
**	Returns:
**		usual sm_error code; ENOMEM, I/O errors
**
**	Last code review: 2005-03-31 23:58:30
**	Last code change: 2005-11-20 17:27:37
*/

sm_ret_T
ibdb_open(const char *base_dir, const char *name, int mode, uint32_t seq, size_t size, uint flags, fs_ctx_P fs_ctx, ibdb_ctx_P *pibdb_ctx)
{
	int r;
	sm_ret_T ret;
	sm_file_T *fp;
	size_t bufsize;
	uint fct_state;
	ibdb_ctx_P ibdb_ctx;
	struct stat sb;

#define FST_MUTEX_INIT	0x01	/* mutex is initialized */

	SM_REQUIRE(pibdb_ctx != NULL);
	SM_REQUIRE(seq > 0);
	SM_REQUIRE(size > 0);	/* larger minimum? */
	fp = NULL;
	fct_state = 0;
	ibdb_ctx = ibdb_ctx_new(base_dir, name, seq, flags);
	if (NULL == ibdb_ctx)
		return sm_error_temp(SM_EM_IBDB, ENOMEM);

	/* Check whether ibdb_ctx->ibdb_dir exists: if no: try to create it */
	r = stat(ibdb_ctx->ibdb_dir, &sb);
	if (r < 0) {
		r = sm_mkdir(ibdb_ctx->ibdb_dir, 0700);
		if (r < 0) {
			ret = sm_error_temp(SM_EM_IBDB, errno);
			goto error;
		}
	}

	ret = sm_io_open(SmStStdio, sm_str_getdata(ibdb_ctx->ibdb_path), mode,
			&fp, SM_IO_WHAT_END);
	if (sm_is_err(ret))
		goto error;
	(void) sm_whatbuf(fp, &bufsize);
	if (0 == bufsize || (bufsize % IBDB_REC_SIZE) != 0) {
		bufsize = (0 == bufsize)
			? IBDB_BUF_SIZE
			: (((bufsize / IBDB_REC_SIZE) + 1) * IBDB_REC_SIZE);

		ibdb_ctx->ibdb_fp_buf = sm_malloc(bufsize);
		if (NULL == ibdb_ctx->ibdb_fp_buf)
			goto error;
		ret = sm_io_setvbuf(fp, ibdb_ctx->ibdb_fp_buf, SM_IO_FBF,
				bufsize);
		if (sm_is_err(ret))
			goto error;
	}
	r = pthread_mutex_init(&ibdb_ctx->ibdb_mutex, SM_PTHREAD_MUTEXATTR);
	if (r != 0) {
		ret = sm_error_perm(SM_EM_IBDB, r);
		goto error;
	}
	SM_SET_FLAG(fct_state, FST_MUTEX_INIT);

	IBDBREQL_INIT(&ibdb_ctx->ibdb_reql_pool);
	/* preallocate some entries?? */

	if (fs_ctx != NULL) {
		ibdb_ctx->ibdb_fs_ctx = fs_ctx;
		ret = fs_new(fs_ctx, ibdb_ctx->ibdb_dir, &ibdb_ctx->ibdb_fs_idx);
		if (sm_is_err(ret))
			goto error;
	}

	/* only if writing? */
	ibdb_ctx->ibdb_version = IBDB_VERSION;
	ibdb_ctx->ibdb_fp = fp;
	ibdb_ctx->ibdb_maxsize = size;
	ibdb_ctx->ibdb_mode = mode;
/*
//	ibdb_ctx->ibdb_created;
//	ibdb_ctx->ibdb_touched;
*/

	/* first: record type */
	ret = sm_io_fputuint32(fp, RT_IBDB_VERS_R);
	if (sm_is_err(ret))
		goto error;
	ret = sm_io_fputv(fp,
		SM_RCBV_INT, RT_IBDB_VERS_N, ibdb_ctx->ibdb_version,
		SM_RCBV_END);
	if (sm_is_err(ret))
		goto error;
	ret = sm_io_falign(fp, IBDB_REC_SIZE);
	if (sm_is_err(ret))
		goto error;

	*pibdb_ctx = ibdb_ctx;
	return ret;

  error:
	/* note: fs_new() not undone on error */
	if (fp != NULL) {
		(void) sm_io_close(fp, SM_IO_CF_NONE);
		fp = NULL;
	}
	if (SM_IS_FLAG(fct_state, FST_MUTEX_INIT)) {
		(void) pthread_mutex_destroy(&ibdb_ctx->ibdb_mutex);
		SM_CLR_FLAG(fct_state, FST_MUTEX_INIT);
	}
	ibdb_ctx_free(ibdb_ctx);
	return ret;
#undef FST_MUTEX_INIT
}

/*
**  We could put the stuff ourselves in uio and call fvwrite() just once.
**  That would require that we store the constants in an int array.
**  We could also optimize and write putrecordTYPE() functions that put
**  and int, str, char * into the buffer.
*/

/*
**  IBDB_HDRMOD_WR -- write header modification record
**
**	Parameters:
**		ibdb_ctx -- INCEDB context
**		ibdb_hdrmod -- header modification data
**		flags -- flags
**		locktype -- kind of locking
**
**	Returns:
**		usual sm_error code; ENOMEM (for new ta), I/O errors
**
**	Last code review:
**	Last code change:
*/

sm_ret_T
ibdb_hdrmod_wr(ibdb_ctx_P ibdb_ctx, ibdb_hdrmod_P ibdb_hdrmod, uint flags, thr_lock_T locktype)
{
	sm_file_T *fp;
	sm_ret_T ret;
	int r;

	SM_ASSERT(ibdb_ctx != NULL);
	SM_ASSERT(ibdb_hdrmod != NULL);

	if (thr_lock_it(locktype)) {
		r = pthread_mutex_lock(&ibdb_ctx->ibdb_mutex);
		SM_LOCK_OK(r);
		if (r != 0) {
			/* LOG? */
			return sm_error_perm(SM_EM_IBDB, r);
		}
	}
	if (!ibdb_is_noroll(flags) &&
	    ibdb_ctx->ibdb_cursize >= ibdb_ctx->ibdb_maxsize)
	{
		/* roll-over */
		ret = ibdb_next(ibdb_ctx);
		if (sm_is_err(ret))
			goto error;
	}
	fp = ibdb_ctx->ibdb_fp;

	/* first: record type */
	ret = sm_io_fputuint32(fp, RT_IBDB_HDRMOD);
	if (sm_is_err(ret))
		goto error;

	/* header modification */
	ret = sm_io_fputv(fp,
		SM_RCBV_BUF, RT_IBDB_HM_TAID,
			(uchar *) ibdb_hdrmod->ibh_ta_id,
			strlen(ibdb_hdrmod->ibh_ta_id), /* use constant?? */
		SM_RCBV_INT, RT_IBDB_HM_TYPE,
			(uint32_t) ibdb_hdrmod->ibh_type,
		SM_RCBV_INT, RT_IBDB_HM_POS, ibdb_hdrmod->ibh_pos,
		SM_RCBV_CSTR, RT_IBDB_HM_HDR, ibdb_hdrmod->ibh_hdr,
		SM_RCBV_END);
	if (sm_is_err(ret))
		goto error;
	ret = sm_io_falign(fp, IBDB_REC_SIZE);
	if (sm_is_err(ret))
		goto error;
#if 0
	ibdb_ctx->ibdb_touched = now;
#endif
	ibdb_ctx->ibdb_cursize += IBDB_REC_SIZE;
	ibdb_ctx->ibdb_changes++;

	ret = SM_SUCCESS;
	/* fall through for unlocking */

  error:
	if ((!sm_is_err(ret) && thr_unl_no_err(locktype))
	    || (sm_is_err(ret) && thr_unl_if_err(locktype)))
	{
		r = pthread_mutex_unlock(&ibdb_ctx->ibdb_mutex);
		SM_ASSERT(0 == r);
		if (r != 0 && sm_is_success(ret))
			ret = sm_error_perm(SM_EM_IBDB, r);
	}

	/* cleanup? */
	return ret;
}

/*
**  IBDB_HDRMODL_WR -- write header modification list
**
**	Parameters:
**		ibdb_ctx -- INCEDB context
**		hdrmodhd -- header of header modification list
**		ta_id -- ta_id
**		flags -- flags
**		locktype -- kind of locking
**
**	Returns:
**		usual sm_error code; ENOMEM (for new ta), I/O errors
**
**	Last code review:
**	Last code change:
*/

sm_ret_T
ibdb_hdrmodl_wr(ibdb_ctx_P ibdb_ctx, sm_hdrmodhd_P sm_hdrmodhd, sessta_id_T ta_id, uint flags, thr_lock_T locktype)
{
	sm_file_T *fp;
	sm_ret_T ret;
	int r;
	sm_hdrmod_P sm_hdrmod;

	if (NULL == sm_hdrmodhd)
		return SM_SUCCESS;
	SM_ASSERT(ibdb_ctx != NULL);

	if (thr_lock_it(locktype)) {
		r = pthread_mutex_lock(&ibdb_ctx->ibdb_mutex);
		SM_LOCK_OK(r);
		if (r != 0) {
			/* LOG? */
			return sm_error_perm(SM_EM_IBDB, r);
		}
	}
	if (!ibdb_is_noroll(flags) &&
	    ibdb_ctx->ibdb_cursize >= ibdb_ctx->ibdb_maxsize)
	{
		/* roll-over */
		ret = ibdb_next(ibdb_ctx);
		if (sm_is_err(ret))
			goto error;
	}

	fp = ibdb_ctx->ibdb_fp;
	for (sm_hdrmod = HDRMODL_FIRST(sm_hdrmodhd);
	     sm_hdrmod != HDRMODL_END(sm_hdrmodhd);
	     sm_hdrmod = HDRMODL_NEXT(sm_hdrmod))
	{
		/* first: record type */
		ret = sm_io_fputuint32(fp, RT_IBDB_HDRMOD);
		if (sm_is_err(ret))
			goto error;

		/* header modification */
		ret = sm_io_fputv(fp,
			SM_RCBV_BUF, RT_IBDB_HM_TAID, (uchar *) ta_id,
				strlen(ta_id), /* use constant?? */
			SM_RCBV_INT, RT_IBDB_HM_TYPE,
				(uint32_t) sm_hdrmod->sm_hm_type,
			SM_RCBV_INT, RT_IBDB_HM_POS, sm_hdrmod->sm_hm_pos,
			/* this may exceed the record size??? */
				SM_RCBV_CSTR, RT_IBDB_HM_HDR,
					sm_hdrmod->sm_hm_hdr,
			SM_RCBV_END);
		if (sm_is_err(ret))
			goto error;
		ret = sm_io_falign(fp, IBDB_REC_SIZE);
		if (sm_is_err(ret))
			goto error;
		ibdb_ctx->ibdb_cursize += IBDB_REC_SIZE;
		ibdb_ctx->ibdb_changes++;
	}

	ret = SM_SUCCESS;
	/* fall through for unlocking */

  error:
	if ((!sm_is_err(ret) && thr_unl_no_err(locktype))
	    || (sm_is_err(ret) && thr_unl_if_err(locktype)))
	{
		r = pthread_mutex_unlock(&ibdb_ctx->ibdb_mutex);
		SM_ASSERT(0 == r);
		if (r != 0 && sm_is_success(ret))
			ret = sm_error_perm(SM_EM_IBDB, r);
	}

	/* cleanup? */
	return ret;
}

/*
**  IBDB_TA_STATUS -- write transaction status
**
**	Parameters:
**		ibdb_ctx -- INCEDB context
**		ibdb_ta -- transaction (sender) data
**		status -- transaction status
**		flags -- flags
**		rcpt_idx_high -- highest rcpt idx (only for cancel)
**		locktype -- kind of locking
**
**	Returns:
**		usual sm_error code; ENOMEM (for new ta), I/O errors
**
**	Last code review: 2005-04-01 00:01:25
**	Last code change: 2006-08-05 04:30:39
*/

sm_ret_T
ibdb_ta_status(ibdb_ctx_P ibdb_ctx, ibdb_ta_P ibdb_ta, int status, uint flags, rcpt_idx_T rcpt_idx_high, thr_lock_T locktype)
{
	sm_file_T *fp;
	sm_ret_T ret;
	int r;

	SM_ASSERT(ibdb_ctx != NULL);
	SM_ASSERT(ibdb_ta != NULL);

	if (thr_lock_it(locktype)) {
		r = pthread_mutex_lock(&ibdb_ctx->ibdb_mutex);
		SM_LOCK_OK(r);
		if (r != 0) {
			/* LOG? */
			return sm_error_perm(SM_EM_IBDB, r);
		}
	}
	if (!ibdb_is_noroll(flags) &&
	    ibdb_ctx->ibdb_cursize >= ibdb_ctx->ibdb_maxsize)
	{
		/* roll-over */
		ret = ibdb_next(ibdb_ctx);
		if (sm_is_err(ret))
			goto error;
	}
	fp = ibdb_ctx->ibdb_fp;

	/* first: record type */
	ret = sm_io_fputuint32(fp, RT_IBDB_TA);
	if (sm_is_err(ret))
		goto error;

	/* transaction status */
	ret = sm_io_fputv(fp,
		SM_RCBV_INT, RT_IBDB_TA_ST, status,
		SM_RCBV_BUF, RT_IBDB_TAID, (uchar *) ibdb_ta->ibt_ta_id,
			strlen(ibdb_ta->ibt_ta_id), /* use constant?? */
		SM_RCBV_CSTR, RT_IBDB_CDBID, ibdb_ta->ibt_cdb_id,
		SM_RCBV_INT, RT_IBDB_NRCPTS, ibdb_ta->ibt_nrcpts,
	/* this may exceed the record size??? */
		SM_RCBV_STR, RT_IBDB_MAIL, ibdb_ta->ibt_mail_pa,
		SM_RCBV_END);
	if (sm_is_err(ret))
		goto error;
	ret = sm_io_falign(fp, IBDB_REC_SIZE);
	if (sm_is_err(ret))
		goto error;
#if 0
	ibdb_ctx->ibdb_touched = now;
#endif
	ibdb_ctx->ibdb_cursize += IBDB_REC_SIZE;
	ibdb_ctx->ibdb_changes++;

	if (IBDB_TA_NEW == status)
		ret = sm_ibc_ta_add(ibdb_ctx->ibdb_ct, ibdb_ta->ibt_ta_id,
			ibdb_ctx->ibdb_sequence);
	else if (IBDB_TA_CANCEL == status) {
		rcpt_idx_T u, lim;

		/* remove all rcpts from cache */
		lim = SM_MAX(rcpt_idx_high + 1, ibdb_ta->ibt_nrcpts);
		for (u = 0; u < lim; u++)
			(void) sm_ibc_rcpt_rm(ibdb_ctx->ibdb_ct, ibdb_ta->ibt_ta_id, u);
	}
	else if (status != IBDB_TA_CANCEL)
		ret = sm_ibc_ta_rm(ibdb_ctx->ibdb_ct, ibdb_ta->ibt_ta_id);
	if (sm_is_err(ret))
		goto error;

#if SM_IBDB_STATS
	if (IBDB_TA_NEW == status)
		ibdb_ctx->ibdb_opentas++;
	else if (status != IBDB_TA_CANCEL) {
		SM_ASSERT(ibdb_ctx->ibdb_opentas > 0);
		ibdb_ctx->ibdb_opentas--;
	}
#endif /* SM_IBDB_STATS */

	ret = SM_SUCCESS;
	if (IBDB_IS_FLAG(flags, IBDB_FL_CLEAN))
		ret = ibdb_clean(ibdb_ctx, THR_NO_LOCK);
	/* fall through for unlocking */

  error:
	if ((!sm_is_err(ret) && thr_unl_no_err(locktype))
	    || (sm_is_err(ret) && thr_unl_if_err(locktype)))
	{
		r = pthread_mutex_unlock(&ibdb_ctx->ibdb_mutex);
		SM_ASSERT(0 == r);
		if (r != 0 && sm_is_success(ret))
			ret = sm_error_perm(SM_EM_IBDB, r);
	}

	/* cleanup? */
	return ret;
}

/*
**  IBDB_RCPT_STATUS -- write recipient status
**
**	Parameters:
**		ibdb_ctx -- INCEDB context
**		ibdb_rcpt -- recipient data
**		status -- recipient status
**		flags -- flags
**		locktype -- kind of locking
**
**	Side Effects:
**		writes directly to IBDB (not undone on error; theoretically
**		this could be tried with fgetpos(), fsetpos())
**
**	Returns:
**		usual sm_error code; ENOMEM (for new rcpt), I/O errors
**
**	Last code review: 2005-03-31 05:38:59
**	Last code change: 2005-03-30 23:08:35
*/

sm_ret_T
ibdb_rcpt_status(ibdb_ctx_P ibdb_ctx, ibdb_rcpt_P ibdb_rcpt, int status, uint flags, thr_lock_T locktype)
{
	sm_file_T *fp;
	sm_ret_T ret;
	int r;

	SM_ASSERT(ibdb_ctx != NULL);
	SM_ASSERT(ibdb_rcpt != NULL);

	if (thr_lock_it(locktype)) {
		r = pthread_mutex_lock(&ibdb_ctx->ibdb_mutex);
		SM_LOCK_OK(r);
		if (r != 0) {
			/* LOG? */
			return sm_error_perm(SM_EM_IBDB, r);
		}
	}
	if (!ibdb_is_noroll(flags) &&
	    ibdb_ctx->ibdb_cursize >= ibdb_ctx->ibdb_maxsize)
	{
		/* roll-over */
		ret = ibdb_next(ibdb_ctx);
		if (sm_is_err(ret))
			goto error;
	}
	fp = ibdb_ctx->ibdb_fp;

	/* first: record type */
	ret = sm_io_fputuint32(fp, RT_IBDB_RCPT);
	if (sm_is_err(ret))
		goto error;

	/* recipient status */
	ret = sm_io_fputv(fp,
		SM_RCBV_INT, RT_IBDB_RCPT_ST, status,
		SM_RCBV_INT, RT_IBDB_RCPT_IDX, ibdb_rcpt->ibr_idx,
		SM_RCBV_BUF, RT_IBDB_TAID, (uchar *) ibdb_rcpt->ibr_ta_id,
		strlen(ibdb_rcpt->ibr_ta_id), /* use constant?? */
	/* this may exceed the record size??? */
		SM_RCBV_STR, RT_IBDB_RCPT_PA, ibdb_rcpt->ibr_pa,
		SM_RCBV_END);
	if (sm_is_err(ret))
		goto error;
	ret = sm_io_falign(fp, IBDB_REC_SIZE);
	if (sm_is_err(ret))
		goto error;

#if 0
	ibdb_ctx->ibdb_touched = now;
#endif
	ibdb_ctx->ibdb_cursize += IBDB_REC_SIZE;
	ibdb_ctx->ibdb_changes++;

	if (IBDB_RCPT_NEW == status)
		ret = sm_ibc_rcpt_add(ibdb_ctx->ibdb_ct, ibdb_rcpt->ibr_ta_id,
			ibdb_rcpt->ibr_idx, ibdb_ctx->ibdb_sequence);
	else
		ret = sm_ibc_rcpt_rm(ibdb_ctx->ibdb_ct, ibdb_rcpt->ibr_ta_id,
			ibdb_rcpt->ibr_idx);
	if (sm_is_err(ret))
		goto error;

#if SM_IBDB_STATS
	if (IBDB_RCPT_NEW == status)
		ibdb_ctx->ibdb_openrcpts++;
	else {
		SM_ASSERT(ibdb_ctx->ibdb_openrcpts > 0);
		ibdb_ctx->ibdb_openrcpts--;
	}
#endif /* SM_IBDB_STATS */

	ret = SM_SUCCESS;
	if (IBDB_IS_FLAG(flags, IBDB_FL_CLEAN))
		ret = ibdb_clean(ibdb_ctx, THR_NO_LOCK);
	/* fall through for unlocking */

  error:
	if ((!sm_is_err(ret) && thr_unl_no_err(locktype))
	    || (sm_is_err(ret) && thr_unl_if_err(locktype)))
	{
		r = pthread_mutex_unlock(&ibdb_ctx->ibdb_mutex);
		SM_ASSERT(0 == r);
		if (r != 0 && sm_is_success(ret))
			ret = sm_error_perm(SM_EM_IBDB, r);
	}
	return ret;
}

/*
**  IBDB_UNLINK -- unlink ibdb files
**
**	Parameters:
**		ibdb_ctx -- ibdb context
**		first -- first sequence number to unlink
**		last -- last sequence number to unlink
**
**	Returns:
**		SM_SUCCESS (unlink() errors are ignored)
**
**	Locking: ibdb_ctx must be locked by caller.
**
**	Last code review: 2005-03-31 05:00:22
**	Last code change:
*/

static sm_ret_T
ibdb_unlink(ibdb_ctx_P ibdb_ctx, size_t first, size_t last)
{
	size_t i;

	SM_ASSERT(ibdb_ctx != NULL);
	for (i = first; i <= last; i++) {
		sm_str_clr(ibdb_ctx->ibdb_name_rm);
		crt_ibdb_path(ibdb_ctx->ibdb_name_rm, ibdb_ctx->ibdb_dir,
			ibdb_ctx->ibdb_name, i);
		(void) unlink((const char *) sm_str_getdata(ibdb_ctx->ibdb_name_rm));
		sm_str_clr(ibdb_ctx->ibdb_name_rm);
	}

	if (ibdb_ctx->ibdb_fs_ctx != NULL && last >= first) {
		ulong kbfree;

		/* This is just an estimate! The actual size may vary. */
		(void) fs_chgfree(ibdb_ctx->ibdb_fs_ctx, ibdb_ctx->ibdb_fs_idx,
			(ibdb_ctx->ibdb_maxsize * (last - first + 1)) / ONEKB,
			&kbfree);
	}
	return SM_SUCCESS;
}

/*
**  IBDB_CLOSE -- close ibdb
**
**	Parameters:
**		ibdb_ctx -- ibdb context
**
**	Returns:
**		usual sm_error code, only from sm_io_close()
**
**	Locking:
**		Does NOT lock ibdb_ctx because it shuts down IBDB.
**		Should it try to get the lock at least?
**
**	Last code review: 2005-04-01 00:03:17
**	Last code change:
*/

sm_ret_T
ibdb_close(ibdb_ctx_P ibdb_ctx)
{
	sm_ret_T ret;
	sm_file_T *fp;

	SM_ASSERT(ibdb_ctx != NULL);
	(void) ibdb_reql_free(ibdb_ctx);
	fp = ibdb_ctx->ibdb_fp;
	if (fp != NULL) {
		ret = sm_io_close(fp, SM_IO_CF_NONE);
		ibdb_ctx->ibdb_fp = NULL;
		if (sm_is_err(ret))
			return ret;
	}
#if SM_IBDB_STATS
	/*
	**  If there are no open transactions and no open recipients
	**  then unlink all ibdb files except for the last one
	**  because one file is needed to keep track of the last used id
	**  (see qm_rdibdb()).
	*/

	if (ibdb_ctx->ibdb_opentas == 0 && ibdb_ctx->ibdb_openrcpts == 0
	    && ibdb_ctx->ibdb_first + 1 <= ibdb_ctx->ibdb_sequence)
	{
		ibdb_unlink(ibdb_ctx, ibdb_ctx->ibdb_first,
			ibdb_ctx->ibdb_sequence - 1);
	}
#endif /* SM_IBDB_STATS */
	(void) pthread_mutex_destroy(&ibdb_ctx->ibdb_mutex);

	ibdb_ctx_free(ibdb_ctx);
	return SM_SUCCESS;
}

/*
**  IBDB_HDRMOD_APP -- append header modification data to request list
**
**	Parameters:
**		ibdb_ctx -- IBDB context
**		ibdb_hdrmod -- header modification data
**		ibdb_req_hd -- request list
**
**	Side Effects: none on error (except if unlock fails)
**
**	Returns:
**		usual sm_error code; ENOMEM, (un)lock
**
**	Locking:
**		locks ibdb_ctx during operation
**
**	Last code review:
**	Last code change:
*/

sm_ret_T
ibdb_hdrmod_app(ibdb_ctx_P ibdb_ctx, ibdb_hdrmod_P ibdb_hdrmod, ibdb_req_hd_P ibdb_req_hd)
{
	sm_ret_T ret;
	int r;
	ibdb_req_P ibdb_req;

	SM_ASSERT(ibdb_ctx != NULL);
	SM_ASSERT(ibdb_hdrmod != NULL);

	r = pthread_mutex_lock(&ibdb_ctx->ibdb_mutex);
	SM_LOCK_OK(r);
	if (r != 0) {
		/* LOG? */
		return sm_error_perm(SM_EM_IBDB, r);
	}

	ret = ibdb_req_new(ibdb_ctx, &ibdb_req);
	if (sm_is_err(ret))
		goto error;
	/* note: We could unlock here, except for releasing req in error case */

	SESSTA_COPY(ibdb_req->ibdb_req_ss_ta_id, ibdb_hdrmod->ibh_ta_id);
	ibdb_req->ibh_req_hdr = SM_CSTR_DUP(ibdb_hdrmod->ibh_hdr);
	if (NULL == ibdb_req->ibh_req_hdr) {
		(void) ibdb_req_free(ibdb_req);	/* always ok */
		ret = sm_error_temp(SM_EM_IBDB, ENOMEM);
		goto error;
	}
	ibdb_req->ibh_req_pos = ibdb_hdrmod->ibh_pos;
	ibdb_req->ibh_req_type = ibdb_hdrmod->ibh_type;
	ibdb_req->ibdb_req_type = IBDB_REQ_HDRMOD;
	IBDBREQL_APP(ibdb_req_hd, ibdb_req);

	/* fall through for unlocking */
  error:
	r = pthread_mutex_unlock(&ibdb_ctx->ibdb_mutex);
	SM_ASSERT(0 == r);
	if (r != 0 && sm_is_success(ret))
		ret = sm_error_perm(SM_EM_IBDB, r);
	return ret;
}

/*
**  IBDB_TA_APP -- append transaction (status) to request list
**
**	Parameters:
**		ibdb_ctx -- IBDB context
**		ibdb_ta -- transaction (sender) data
**		ibdb_req_hd -- request list
**		status -- transaction status
**
**	Side Effects: none on error (except if unlock fails)
**
**	Returns:
**		usual sm_error code; ENOMEM, (un)lock
**
**	Locking:
**		locks ibdb_ctx during operation
**
**	Last code review: 2005-03-31 18:33:26
**	Last code change:
*/

sm_ret_T
ibdb_ta_app(ibdb_ctx_P ibdb_ctx, ibdb_ta_P ibdb_ta, ibdb_req_hd_P ibdb_req_hd, int status)
{
	sm_ret_T ret;
	int r;
	ibdb_req_P ibdb_req;

	SM_ASSERT(ibdb_ctx != NULL);
	SM_ASSERT(ibdb_ta != NULL);

	r = pthread_mutex_lock(&ibdb_ctx->ibdb_mutex);
	SM_LOCK_OK(r);
	if (r != 0) {
		/* LOG? */
		return sm_error_perm(SM_EM_IBDB, r);
	}

	ret = ibdb_req_new(ibdb_ctx, &ibdb_req);
	if (sm_is_err(ret))
		goto error;
	/* note: We could unlock here, except for releasing req in error case */

	SESSTA_COPY(ibdb_req->ibdb_req_ss_ta_id, ibdb_ta->ibt_ta_id);
	ibdb_req->ibdb_req_addr_pa = sm_str_dup(NULL, ibdb_ta->ibt_mail_pa);
	if (NULL == ibdb_req->ibdb_req_addr_pa) {
		(void) ibdb_req_free(ibdb_req);	/* always ok */
		ret = sm_error_temp(SM_EM_IBDB, ENOMEM);
		goto error;
	}
	ibdb_req->ibdb_req_cdb_id = SM_CSTR_DUP(ibdb_ta->ibt_cdb_id);
	ibdb_req->ibdb_req_nrcpts = ibdb_ta->ibt_nrcpts;
	ibdb_req->ibdb_req_status = status;
	ibdb_req->ibdb_req_type = IBDB_REQ_TA;
	IBDBREQL_APP(ibdb_req_hd, ibdb_req);

	/* fall through for unlocking */
  error:
	r = pthread_mutex_unlock(&ibdb_ctx->ibdb_mutex);
	SM_ASSERT(0 == r);
	if (r != 0 && sm_is_success(ret))
		ret = sm_error_perm(SM_EM_IBDB, r);
	return ret;
}

/*
**  IBDB_RCPT_APP -- append recipient (status) to request list
**
**	Parameters:
**		ibdb_ctx -- IBDB context
**		ibdb_rcpt -- recipient data
**		ibdb_req_hd -- request list
**		status -- recipient status
**
**	Side Effects: none on error (except if unlock fails)
**
**	Returns:
**		usual sm_error code; ENOMEM
**
**	Locking:
**		locks ibdb_ctx during operation
**
**	Last code review: 2005-04-01 00:05:27
**	Last code change:
*/

sm_ret_T
ibdb_rcpt_app(ibdb_ctx_P ibdb_ctx, ibdb_rcpt_P ibdb_rcpt, ibdb_req_hd_P ibdb_req_hd, int status)
{
	ibdb_req_P ibdb_req;
	sm_ret_T ret;
	int r;

	SM_ASSERT(ibdb_ctx != NULL);
	SM_ASSERT(ibdb_rcpt != NULL);

	r = pthread_mutex_lock(&ibdb_ctx->ibdb_mutex);
	SM_LOCK_OK(r);
	if (r != 0) {
		/* LOG? */
		return sm_error_perm(SM_EM_IBDB, r);
	}

	ret = ibdb_req_new(ibdb_ctx, &ibdb_req);
	if (sm_is_err(ret))
		goto error;
	/* note: We could unlock here, except for releasing req in error case */

	SESSTA_COPY(ibdb_req->ibdb_req_ss_ta_id, ibdb_rcpt->ibr_ta_id);
	ibdb_req->ibdb_req_addr_pa = sm_str_dup(NULL, ibdb_rcpt->ibr_pa);
	if (NULL == ibdb_req->ibdb_req_addr_pa) {
		ibdb_req_free(ibdb_req);
		ret = sm_error_temp(SM_EM_IBDB, ENOMEM);
		goto error;
	}
	ibdb_req->ibdb_req_rcpt_idx = ibdb_rcpt->ibr_idx;
	ibdb_req->ibdb_req_status = status;
	ibdb_req->ibdb_req_type = IBDB_REQ_RCPT;
	IBDBREQL_APP(ibdb_req_hd, ibdb_req);

  error:
	r = pthread_mutex_unlock(&ibdb_ctx->ibdb_mutex);
	SM_ASSERT(0 == r);
	if (r != 0 && sm_is_success(ret))
		ret = sm_error_perm(SM_EM_IBDB, r);

	/* cleanup? */
	return ret;
}

/*
**  IBDB_WR_STATUS -- write status (request list)
**
**	Parameters:
**		ibdb_ctx -- IBDB context
**		ibdb_req_hd -- request list
**
**	Returns:
**		usual sm_error code; ENOMEM (for new rcpt/ta), I/O errors
**
**	Side Effects:
**		writes to IBDB, does not undo changes when an error occurs,
**		see ibdb_{ta,rcpt}_status()
**
**	Locking:
**		locks ibdb_ctx during operation
**
**	Last code review: 2005-04-01 00:54:12
**	Last code change:
*/

sm_ret_T
ibdb_wr_status(ibdb_ctx_P ibdb_ctx, ibdb_req_hd_P ibdb_req_hd)
{
	sm_ret_T ret;
	int r;
	ibdb_req_P ibdb_req;
	ibdb_rcpt_T ibdb_rcpt;
	ibdb_ta_T ibdb_ta;
	ibdb_hdrmod_T ibdb_hdrmod;

	SM_ASSERT(ibdb_ctx != NULL);
	SM_ASSERT(ibdb_req_hd != NULL);
	if  (IBDBREQL_EMPTY(ibdb_req_hd))
		return SM_SUCCESS;

	r = pthread_mutex_lock(&ibdb_ctx->ibdb_mutex);
	SM_LOCK_OK(r);
	if (r != 0) {
		/* LOG? */
		return sm_error_perm(SM_EM_IBDB, r);
	}

	ret = SM_SUCCESS;
	if  (IBDBREQL_EMPTY(ibdb_req_hd))
		goto done;

	/* walk through request list and write them to IBDB */
	for (ibdb_req = IBDBREQL_FIRST(ibdb_req_hd);
	     ibdb_req != IBDBREQL_END(ibdb_req_hd);
	     ibdb_req = IBDBREQL_NEXT(ibdb_req))
	{
		/* Check type */
		switch (ibdb_req->ibdb_req_type) {
		   case IBDB_REQ_RCPT:
			ibdb_rcpt.ibr_ta_id = ibdb_req->ibdb_req_ss_ta_id;
			ibdb_rcpt.ibr_pa = ibdb_req->ibdb_req_addr_pa;
			ibdb_rcpt.ibr_idx = ibdb_req->ibdb_req_rcpt_idx;
			ret = ibdb_rcpt_status(ibdb_ctx, &ibdb_rcpt,
				ibdb_req->ibdb_req_status, IBDB_FL_NOROLL,
				THR_NO_LOCK);
			break;

		   case IBDB_REQ_TA:
			ibdb_ta.ibt_ta_id = ibdb_req->ibdb_req_ss_ta_id;
			ibdb_ta.ibt_mail_pa = ibdb_req->ibdb_req_addr_pa;
			ibdb_ta.ibt_cdb_id = ibdb_req->ibdb_req_cdb_id;
			ibdb_ta.ibt_nrcpts = ibdb_req->ibdb_req_nrcpts;
			ret = ibdb_ta_status(ibdb_ctx, &ibdb_ta,
				ibdb_req->ibdb_req_status, IBDB_FL_NOROLL,
				0, THR_NO_LOCK);
			break;

		   case IBDB_REQ_HDRMOD:
			ibdb_hdrmod.ibh_ta_id = ibdb_req->ibdb_req_ss_ta_id;
			ibdb_hdrmod.ibh_hdr = ibdb_req->ibh_req_hdr;
			ibdb_hdrmod.ibh_pos = ibdb_req->ibh_req_pos;
			ibdb_hdrmod.ibh_type = ibdb_req->ibh_req_type;
			ret = ibdb_hdrmod_wr(ibdb_ctx, &ibdb_hdrmod,
				IBDB_FL_NOROLL, THR_NO_LOCK);
			break;

		   default:
			/*
			**  XXX Really abort? May screw up DB...
			**  How about logging a fatal error and returning
			**  that to the caller?
			*/

			sm_abort("wrong ibdb_req_type %d",
				ibdb_req->ibdb_req_type);
			ret = sm_error_perm(SM_EM_IBDB, SM_E_UNEXPECTED);
			break;
		}
		if (sm_is_err(ret))
			goto error;
	}
	/* Commit it? Not really necessary... */

	/* Everything is ok: remove requests from list */
	while (!IBDBREQL_EMPTY(ibdb_req_hd)) {
		ibdb_req = IBDBREQL_FIRST(ibdb_req_hd);
		IBDBREQL_REMOVE(ibdb_req_hd);
		(void) ibdb_req_rel(ibdb_ctx, ibdb_req); /* always ok... */
	}

	/* Always try to clean? */
	ret = ibdb_clean(ibdb_ctx, THR_NO_LOCK);

	/* Fall through for unlocking */
  done:
  error:
	r = pthread_mutex_unlock(&ibdb_ctx->ibdb_mutex);
	SM_ASSERT(0 == r);
	if (r != 0 && sm_is_success(ret))
		ret = sm_error_perm(SM_EM_IBDB, r);

	/* cleanup? */
	return ret;
}

/*
**  IBDB_CLEAN -- cleanup IBDB
**	Determine lowest sequence number that is still in use and remove
**	all "older" files.
**
**	Parameters:
**		ibdb_ctx -- ibdb context
**		locktype -- kind of locking
**
**	Returns:
**		usual sm_error code; only (un)lock
**
**	Locking: locks ibdb_ctx if requested.
**
**	Side Effects:
**		unlinks ibdb files that are no longer needed.
**
**	Last code review: 2005-03-31 05:05:02
**	Last code change:
*/

sm_ret_T
ibdb_clean(ibdb_ctx_P ibdb_ctx, thr_lock_T locktype)
{
	sm_ret_T ret;
	int r;
	uint32_t seq;

	SM_ASSERT(ibdb_ctx != NULL);
	if (thr_lock_it(locktype)) {
		r = pthread_mutex_lock(&ibdb_ctx->ibdb_mutex);
		SM_LOCK_OK(r);
		if (r != 0) {
			/* LOG? */
			return sm_error_perm(SM_EM_IBDB, r);
		}
	}
	ret = SM_SUCCESS;

	/* Not worth to do anything? At least SEQ_MIN_PURGE files? */
	if (ibdb_ctx->ibdb_sequence - ibdb_ctx->ibdb_first <= SEQ_MIN_PURGE ||
	    ibdb_ctx->ibdb_sequence - ibdb_ctx->ibdb_cleanrun <= SEQ_MIN_CLEAN)
		goto unlock;
	ibdb_ctx->ibdb_cleanrun = ibdb_ctx->ibdb_sequence;
	ret = sm_ibc_low_seq(ibdb_ctx->ibdb_ct, &seq); /* always OK */
	if (sm_is_err(ret))
		goto unlock;
#if 0
	sm_io_fprintf(smioerr, "ibdb_clean: first=%u, seq=%u, cur=%u\n",
		ibdb_ctx->ibdb_first, seq, ibdb_ctx->ibdb_sequence);
#endif

	/* no reference at all? remove everything but current file */
	if (0 == seq) {
		/* make sure seq is > 0 */
		if (ibdb_ctx->ibdb_sequence <= 1)
			goto unlock;
		seq = ibdb_ctx->ibdb_sequence - 1;
	}
	if (seq + 1 <= ibdb_ctx->ibdb_first || seq >= ibdb_ctx->ibdb_sequence
	    || seq == 0)
		goto unlock;
	ret = ibdb_unlink(ibdb_ctx, ibdb_ctx->ibdb_first, seq - 1);
	if (sm_is_err(ret))			/* always OK */
		goto unlock;
#if 0
	sm_io_fprintf(smioerr, "ibdb_clean ok: first=%u, seq=%u, cur=%u\n",
		ibdb_ctx->ibdb_first, seq, ibdb_ctx->ibdb_sequence);
#endif
	ibdb_ctx->ibdb_first = seq;

  unlock:
	if ((!sm_is_err(ret) && thr_unl_no_err(locktype))
	    || (sm_is_err(ret) && thr_unl_if_err(locktype)))
	{
		r = pthread_mutex_unlock(&ibdb_ctx->ibdb_mutex);
		SM_ASSERT(0 == r);
		if (r != 0 && sm_is_success(ret))
			ret = sm_error_perm(SM_EM_IBDB, r);
	}
	return ret;
}

/*
**  IBDB_FS_GETFREE -- get free disk space in IBDB
**
**	Parameters:
**		ibdb_ctx -- ibdb context
**		pkbfree -- (pointer to) free space (KB) (output)
**
**	Returns:
**		usual sm_error code; fs_getfree(): errno from stat(); ENXIO
**
**	Last code review: 2005-04-21 23:50:08
**	Last code change:
*/

sm_ret_T
ibdb_fs_getfree(ibdb_ctx_P ibdb_ctx, ulong *pkbfree)
{
	sm_ret_T ret;

	SM_REQUIRE(pkbfree != NULLPTR);
	if (ibdb_ctx->ibdb_fs_ctx != NULL) {
		ret = fs_getfree(ibdb_ctx->ibdb_fs_ctx, ibdb_ctx->ibdb_fs_idx, pkbfree);
	}
	else {
		ret = sm_error_perm(SM_EM_IBDB, ENXIO);
		*pkbfree = 0;
	}
	return ret;
}

/*
**  IBDB_STATS -- print IBDB stats
**
**	Parameters:
**		ibdb_ctx -- ibdb context
**		fp -- output file
**
**	Returns:
**		SM_SUCCESS
**
**	Last code review: 2005-04-01 00:55:05
**	Last code change:
*/

sm_ret_T
ibdb_stats(ibdb_ctx_P ibdb_ctx, sm_file_T *fp)
{
#if SM_IBDB_STATS
	SM_REQUIRE(ibdb_ctx != NULL);
	sm_io_fprintf(fp,
		"IBDB open TAs   =%u\n"
		"IBDB open RCPTs =%u\n"
		"IBDB commits    =%lu\n"
		"IBDB seq nr     =%u\n"
		"IBDB changes    =%u\n"
		"IBDB first      =%u\n"
		"IBDB cleanrun   =%u\n"
		, ibdb_ctx->ibdb_opentas
		, ibdb_ctx->ibdb_openrcpts
		, ibdb_ctx->ibdb_commits
		, ibdb_ctx->ibdb_sequence
		, ibdb_ctx->ibdb_changes
		, ibdb_ctx->ibdb_first
		, ibdb_ctx->ibdb_cleanrun
		);
#endif /* SM_IBDB_STATS */
	return SM_SUCCESS;
}

typedef struct ibdbs_ctx_S	ibdbs_ctx_T, *ibdbs_ctx_P;

struct ibdbs_ctx_S
{
	sm_file_T	*ibdbs_fp;	/* file for output */
	uint32_t	 ibdbs_entries;	/* number of entries */
	uint32_t	 ibdbs_sequence;	/* current sequence number */
};

/*
**  SM_IBC_SHOW -- Callback function: print bht entries
**
**	Parameters:
**		ibdb_ctx -- ibdb context
**		ctx -- ibdbs_context
**
**	Returns:
**		SM_SUCCESS
*/

static sm_ret_T
sm_ibc_show(bht_entry_P bhte, void *ctx)
{
	ibc_e_P ibc_e;
	ibdbs_ctx_P ibdbs_ctx;

	SM_REQUIRE(bhte != NULL);
	SM_REQUIRE(ctx != NULL);

	ibc_e = (ibc_e_P) bhte->bhe_value;
	ibdbs_ctx = (ibdbs_ctx_P) ctx;
	++ibdbs_ctx->ibdbs_entries;
	if (ibdbs_ctx->ibdbs_sequence > ibc_e->ibc_sequence) {
		sm_io_fprintf(ibdbs_ctx->ibdbs_fp,
			"ibc: id=%" SM_XSTR(SMTP_RCPTID_LEN) "s, type=%d, seq=%u\n"
			, ibc_e->ibc_id
			, ibc_e->ibc_type
			, ibc_e->ibc_sequence
			);
		ibdbs_ctx->ibdbs_sequence = ibc_e->ibc_sequence;
	}
	return SM_SUCCESS;
}

/*
**  IBDBC_SHOW_SEQ -- print IBDB Cache content
**
**	Parameters:
**		bht -- Hash table
**		fp -- file pointer
**
**	Returns:
**		usual sm_error code; only (un)lock
**
**	Locking:
**		must be done by caller (ibdb_ctx->ibdb_mutex)
*/

sm_ret_T
ibdbc_show_seq(ibdb_ctx_P ibdb_ctx, thr_lock_T locktype, sm_file_T *fp)
{
	sm_ret_T ret;
	int r;
	ibdbs_ctx_T ibdbs_ctx;

	SM_ASSERT(ibdb_ctx != NULL);
	if (thr_lock_it(locktype)) {
		r = pthread_mutex_lock(&ibdb_ctx->ibdb_mutex);
		SM_LOCK_OK(r);
		if (r != 0) {
			/* LOG? */
			return sm_error_perm(SM_EM_IBDB, r);
		}
	}
	ret = SM_SUCCESS;

	ibdbs_ctx.ibdbs_fp = fp;
	ibdbs_ctx.ibdbs_sequence = UINT32_MAX;
	ibdbs_ctx.ibdbs_entries = 0;
	ret = bht_walk(ibdb_ctx->ibdb_ct, sm_ibc_show, &ibdbs_ctx);
	sm_io_fprintf(fp, "ibdbc: lowest_seq=%u\n", ibdbs_ctx.ibdbs_sequence);
	sm_io_fprintf(fp, "ibdbc: #entries  =%u\n", ibdbs_ctx.ibdbs_entries);

	if ((!sm_is_err(ret) && thr_unl_no_err(locktype))
	    || (sm_is_err(ret) && thr_unl_if_err(locktype)))
	{
		r = pthread_mutex_unlock(&ibdb_ctx->ibdb_mutex);
		if (r != 0 && sm_is_success(ret))
			ret = sm_error_perm(SM_EM_IBDB, r);
	}
	return ret;
}


#if 0
///*
//**  SM_IDBSETINFO -- set/modify information for a file
//**
//**	Parameters:
//**		fp -- file to set info for
//**		what -- type of info to set
//**		valp -- location of data used for setting
//**
//**	Returns:
//**		usual sm_error code.
//*/
//
//sm_ret_T
//sm_idbsetinfo(sm_file_T *fp, int what, void *valp)
//{
//	int r;
//
//	switch (what)
//	{
//	  case SM_IO_WHAT_COMMIT:
//		SM_ASSERT(is_valid_fd(f_fd(*fp)));
//		r = fsync(f_fd(*fp));
//		if (r == -1)
//			return sm_error_perm(SM_EM_IBDB, errno);
//		return SM_SUCCESS;
//
//	  default:
//		return sm_error_perm(SM_EM_IBDB, EINVAL);
//	}
//}
//
///*
//**  SM_IDBGETINFO -- get information about the open file
//**
//**	Parameters:
//**		fp -- file to get info for
//**		what -- type of info to get
//**		valp -- location to place found info
//**
//**	Returns:
//**		Success: may or may not place info in 'valp' depending
//**			on 'what' value, and returns values >=0. Return
//**			value may be the obtained info
//**		Failure: usual sm_error code.
//*/
//
//sm_ret_T
//sm_idbgetinfo(sm_file_T *fp, int what, void *valp)
//{
//	switch (what)
//	{
//	  case SM_IO_WHAT_FD:
//		return f_fd(*fp);
//
//	  case SM_IO_WHAT_SIZE:
//	  {
//		struct stat st;
//
//		if (fstat(f_fd(*fp), &st) < 0)
//			return sm_error_perm(SM_EM_IBDB, errno);
//		return st.st_size;
//	  }
//
//	  case SM_IO_IS_READABLE:
//	  {
//		fd_set readfds;
//		struct timeval timeout;
//
//		FD_ZERO(&readfds);
//		SM_FD_SET(f_fd(*fp), &readfds);
//		timeout.tv_sec = 0;
//		timeout.tv_usec = 0;
//		if (select(f_fd(*fp) + 1, FDSET_CAST &readfds, NULL, NULL,
//			   &timeout) > 0 &&
//		    SM_FD_ISSET(f_fd(*fp), &readfds))
//			return 1;
//		return 0;
//	  }
//
//	  default:
//		return sm_error_perm(SM_EM_IBDB, EINVAL);
//	}
//}
//
///*
//**  SM_IDBREAD -- read from the file
//**
//**	Parameters:
//**
//**	Returns:
//**		usual sm_error code
//*/
//
//sm_ret_T
//sm_idbread(sm_file_T *fp, uchar *buf, size_t n, ssize_t *bytesread)
//{
//	ssize_t ret;
//
//	*bytesread = 0;
//	do
//	{
//		errno = 0;
//		ret = read(f_fd(*fp), buf, n);
//	} while (ret == -1 && EINTR == errno);
//	if (ret == -1)
//		return sm_error_perm(SM_EM_IBDB, errno);
//	*bytesread = ret;
//
//	/* if the read succeeded, update the current offset */
//	if (ret > 0)
//		fp->f_lseekoff += ret;
//	return SM_SUCCESS;
//}
#endif /* 0 */