389-ds-base/ldap/servers/plugins/replication/cl5_clcache.c

/** BEGIN COPYRIGHT BLOCK
 * Copyright (C) 2005 Red Hat, Inc.
 * All rights reserved.
 *
 * License: GPL (version 3 or any later version).
 * See LICENSE for details. 
 * END COPYRIGHT BLOCK **/

#ifdef HAVE_CONFIG_H
#  include <config.h>
#endif


#include "errno.h"			/* ENOMEM, EVAL used by Berkeley DB */
#include "db.h"				/* Berkeley DB */
#include "cl5.h"			/* changelog5Config */
#include "cl5_clcache.h"

/* newer bdb uses DB_BUFFER_SMALL instead of ENOMEM as the
   error return if the given buffer in which to load a
   key or value is too small - if it is not defined, define
   it here to ENOMEM
*/
#ifndef DB_BUFFER_SMALL
#define DB_BUFFER_SMALL ENOMEM
#endif

/*
 * Constants for the buffer pool:
 *
 * DEFAULT_CLC_BUFFER_PAGE_COUNT
 *		Little performance boost if it is too small.
 *
 * DEFAULT_CLC_BUFFER_PAGE_SIZE
 * 		Its value is determined based on the DB requirement that
 *		the buffer size should be the multiple of 1024.
 */
#define DEFAULT_CLC_BUFFER_COUNT_MIN		10
#define DEFAULT_CLC_BUFFER_COUNT_MAX		0
#define DEFAULT_CLC_BUFFER_PAGE_COUNT		32
#define DEFAULT_CLC_BUFFER_PAGE_SIZE		1024
#define WORK_CLC_BUFFER_PAGE_SIZE 8*DEFAULT_CLC_BUFFER_PAGE_SIZE

enum {
	CLC_STATE_READY = 0,		/* ready to iterate */
	CLC_STATE_UP_TO_DATE,		/* remote RUV already covers the CSN */
	CLC_STATE_CSN_GT_RUV,		/* local RUV doesn't conver the CSN */
	CLC_STATE_NEW_RID,			/* unknown RID to local RUVs */
	CLC_STATE_UNSAFE_RUV_CHANGE,/* (RUV1 < maxcsn-in-buffer) && (RUV1 < RUV1') */
	CLC_STATE_DONE,				/* no more change */
	CLC_STATE_ABORTING			/* abort replication session */
};

typedef struct clc_busy_list CLC_Busy_List;

struct csn_seq_ctrl_block {
	ReplicaId	rid;				/* RID this block serves */
	CSN			*consumer_maxcsn;	/* Don't send CSN <= this */
	CSN			*local_maxcsn;		/* Don't send CSN > this */
	CSN			*prev_local_maxcsn;	/* Copy of last state at buffer loading */
	CSN			*local_mincsn;		/* Used to determin anchor csn*/
	int			state;			/* CLC_STATE_* */
};

/*
 * Each cl5replayiterator acquires a buffer from the buffer pool
 * at the beginning of a replication session, and returns it back
 * at the end.
 */
struct clc_buffer {
	char		*buf_agmt_name;		/* agreement acquired this buffer */
	ReplicaId	 buf_consumer_rid;	/* help checking threshold csn */
	const RUV	*buf_consumer_ruv;	/* used to skip change */
	const RUV	*buf_local_ruv;		/* used to refresh local_maxcsn */
	int		buf_ignoreConsumerRID;	/* how to handle updates from consumer */
	int	 	buf_load_cnt;		/* number of loads for session */

	/*
	 * fields for retriving data from DB
	 */
	int			 buf_state;
	CSN			*buf_current_csn;
	int			 buf_load_flag;		/* db flag DB_MULTIPLE_KEY, DB_SET, DB_NEXT */
	DBC			*buf_cursor;
	DBT			 buf_key;			/* current csn string */
	DBT			 buf_data;			/* data retrived from db */
	void		*buf_record_ptr;	/* ptr to the current record in data */
	CSN			*buf_missing_csn;	/* used to detect persistent missing of CSN */
	CSN			*buf_prev_missing_csn;	/* used to surpress the repeated messages */

	/* fields for control the CSN sequence sent to the consumer */
	struct csn_seq_ctrl_block **buf_cscbs;
	int			 buf_num_cscbs;		/* number of csn sequence ctrl blocks */
	int			 buf_max_cscbs;

	/* fields for debugging stat */
	int		 	 buf_record_cnt;	/* number of changes for session */
	int		 	 buf_record_skipped;	/* number of changes skipped */
	int		 	 buf_skipped_new_rid;	/* number of changes skipped due to new_rid */
	int		 	 buf_skipped_csn_gt_cons_maxcsn;	/* number of changes skipped due to csn greater than consumer maxcsn */
	int		 	 buf_skipped_up_to_date;	/* number of changes skipped due to consumer being up-to-date for the given rid */
	int		 	 buf_skipped_csn_gt_ruv;	/* number of changes skipped due to preceedents are not covered by local RUV snapshot */
	int		 	 buf_skipped_csn_covered;	/* number of changes skipped due to CSNs already covered by consumer RUV */

	/*
	 * fields that should be accessed via bl_lock or pl_lock
	 */
	CLC_Buffer	*buf_next;			/* next buffer in the same list */
	CLC_Busy_List *buf_busy_list;	/* which busy list I'm in */
};

/*
 * Each changelog has a busy buffer list
 */
struct clc_busy_list {
	PRLock			*bl_lock;
	DB				*bl_db;				/* changelog db handle */
	CLC_Buffer		*bl_buffers;		/* busy buffers of this list */
	CLC_Busy_List	*bl_next;			/* next busy list in the pool */
};

/*
 * Each process has a buffer pool
 */ 
struct clc_pool {
	Slapi_RWLock		*pl_lock;				/* cl writer and agreements */
	DB_ENV			**pl_dbenv;				/* pointer to DB_ENV for all the changelog files */
	CLC_Busy_List	*pl_busy_lists;			/* busy buffer lists, one list per changelog file */
	int				 pl_buffer_cnt_now;		/* total number of buffers */
	int				 pl_buffer_cnt_min;		/* free a newly returned buffer if _now > _min */
	int				 pl_buffer_cnt_max;		/* no use */
	int				 pl_buffer_default_pages;	/* num of pages in a new buffer */
};

/* static variables */
static struct clc_pool *_pool = NULL;	/* process's buffer pool */

/* static prototypes */
static int	clcache_initial_anchorcsn ( CLC_Buffer *buf, int *flag );
static int	clcache_adjust_anchorcsn ( CLC_Buffer *buf, int *flag );
static void	clcache_refresh_consumer_maxcsns ( CLC_Buffer *buf );
static int	clcache_refresh_local_maxcsns ( CLC_Buffer *buf );
static int	clcache_skip_change ( CLC_Buffer *buf );
static int	clcache_load_buffer_bulk ( CLC_Buffer *buf, int flag );
static int	clcache_open_cursor ( DB_TXN *txn, CLC_Buffer *buf, DBC **cursor );
static int	clcache_cursor_get ( DBC *cursor, CLC_Buffer *buf, int flag );
static struct csn_seq_ctrl_block *clcache_new_cscb(void);
static void	clcache_free_cscb ( struct csn_seq_ctrl_block ** cscb );
static CLC_Buffer	*clcache_new_buffer ( ReplicaId consumer_rid );
static void	clcache_delete_buffer ( CLC_Buffer **buf );
static CLC_Busy_List *clcache_new_busy_list(void);
static void	clcache_delete_busy_list ( CLC_Busy_List **bl );
static int	clcache_enqueue_busy_list( DB *db, CLC_Buffer *buf );
static void csn_dup_or_init_by_csn ( CSN **csn1, CSN *csn2 );

/*
 * Initiates the process buffer pool. This should be done
 * once and only once when process starts.
 */
int
clcache_init ( DB_ENV **dbenv )
{
	if (_pool) {
		return 0; /* already initialized */
	}
	if (NULL == dbenv) {
		return -1;
	}
	_pool = (struct clc_pool*) slapi_ch_calloc ( 1, sizeof ( struct clc_pool ));
	_pool->pl_dbenv = dbenv;
	_pool->pl_buffer_cnt_min = DEFAULT_CLC_BUFFER_COUNT_MIN;
	_pool->pl_buffer_cnt_max = DEFAULT_CLC_BUFFER_COUNT_MAX;
	_pool->pl_buffer_default_pages = DEFAULT_CLC_BUFFER_COUNT_MAX;
	_pool->pl_lock = slapi_new_rwlock ();
	return 0;
}

/*
 * This is part of a callback function when changelog configuration
 * is read or updated.
 */
void
clcache_set_config ()
{
	slapi_rwlock_wrlock ( _pool->pl_lock );

	_pool->pl_buffer_cnt_max = CL5_DEFAULT_CONFIG_CACHESIZE;

	/*
	 * According to http://www.sleepycat.com/docs/api_c/dbc_get.html,
	 * data buffer should be a multiple of 1024 bytes in size
	 * for DB_MULTIPLE_KEY operation.
	 */
	_pool->pl_buffer_default_pages = CL5_DEFAULT_CONFIG_CACHEMEMSIZE / DEFAULT_CLC_BUFFER_PAGE_SIZE + 1;
	if ( _pool->pl_buffer_default_pages <= 0 ) { /* this never be true... */
		_pool->pl_buffer_default_pages = DEFAULT_CLC_BUFFER_PAGE_COUNT;
	}

	slapi_rwlock_unlock ( _pool->pl_lock );
}

/*
 * Gets the pointer to a thread dedicated buffer, or allocates
 * a new buffer if there is no buffer allocated yet for this thread.
 *
 * This is called when a cl5replayiterator is created for
 * a replication session.
 */ 
int
clcache_get_buffer ( CLC_Buffer **buf, DB *db, ReplicaId consumer_rid, const RUV *consumer_ruv, const RUV *local_ruv )
{
	int rc = 0;
	int need_new;

	if ( buf == NULL ) return CL5_BAD_DATA;

	*buf = NULL;

	/* if the pool was re-initialized, the thread private cache will be invalid,
	   so we must get a new one */
	need_new = (!_pool || !_pool->pl_busy_lists || !_pool->pl_busy_lists->bl_buffers);

	if ( (!need_new) && (NULL != ( *buf = (CLC_Buffer*) get_thread_private_cache())) ) {
		slapi_log_err(SLAPI_LOG_REPL, get_thread_private_agmtname(),
						  "clcache_get_buffer - found thread private buffer cache %p\n", *buf);
		slapi_log_err(SLAPI_LOG_REPL, get_thread_private_agmtname(),
						  "clcache_get_buffer - _pool is %p _pool->pl_busy_lists is %p _pool->pl_busy_lists->bl_buffers is %p\n",
						  _pool, _pool ? _pool->pl_busy_lists : NULL,
						  (_pool && _pool->pl_busy_lists) ? _pool->pl_busy_lists->bl_buffers : NULL);
		(*buf)->buf_state = CLC_STATE_READY;
		(*buf)->buf_load_cnt = 0;
		(*buf)->buf_record_cnt = 0;
		(*buf)->buf_record_skipped = 0;
		(*buf)->buf_cursor = NULL;
		(*buf)->buf_skipped_new_rid = 0;
		(*buf)->buf_skipped_csn_gt_cons_maxcsn = 0;
		(*buf)->buf_skipped_up_to_date = 0;
		(*buf)->buf_skipped_csn_gt_ruv = 0;
		(*buf)->buf_skipped_csn_covered = 0;
		(*buf)->buf_cscbs  = (struct csn_seq_ctrl_block **) slapi_ch_calloc(MAX_NUM_OF_MASTERS + 1,
			                 sizeof(struct csn_seq_ctrl_block *));
		(*buf)->buf_num_cscbs = 0;
		(*buf)->buf_max_cscbs = MAX_NUM_OF_MASTERS;
	}
	else {
		*buf = clcache_new_buffer ( consumer_rid );
		if ( *buf ) {
			if ( 0 == clcache_enqueue_busy_list ( db, *buf ) ) {
				set_thread_private_cache ( (void*) (*buf) );
			}
			else {
				clcache_delete_buffer ( buf );
			}
		}
	}

	if ( NULL != *buf ) {
		CSN *c_csn = NULL;
		CSN *l_csn = NULL;
		(*buf)->buf_consumer_ruv = consumer_ruv;
		(*buf)->buf_local_ruv = local_ruv;
		(*buf)->buf_load_flag = DB_MULTIPLE_KEY;
		ruv_get_largest_csn_for_replica (consumer_ruv, consumer_rid, &c_csn);
		ruv_get_largest_csn_for_replica (local_ruv, consumer_rid, &l_csn);
		if (l_csn && csn_compare(l_csn, c_csn) > 0) {
			/* the supplier has updates for the consumer RID and
			 * these updates are newer than on the consumer
			 */
			(*buf)->buf_ignoreConsumerRID = 0;
		} else {
			(*buf)->buf_ignoreConsumerRID = 1;
		}
		csn_free(&c_csn);
		csn_free(&l_csn);
	}
	else {
		slapi_log_err(SLAPI_LOG_ERR, get_thread_private_agmtname(),
			"clcache_get_buffer - Can't allocate new buffer\n" );
		rc = CL5_MEMORY_ERROR;
	}

	return rc;
}

/*
 * Returns a buffer back to the buffer pool.
 */
void
clcache_return_buffer ( CLC_Buffer **buf )
{
	int i;

	slapi_log_err(SLAPI_LOG_REPL, (*buf)->buf_agmt_name,
			  "clcache_return_buffer - session end: state=%d load=%d sent=%d skipped=%d skipped_new_rid=%d "
			  "skipped_csn_gt_cons_maxcsn=%d skipped_up_to_date=%d "
			  "skipped_csn_gt_ruv=%d skipped_csn_covered=%d\n",
			  (*buf)->buf_state,
			  (*buf)->buf_load_cnt,
			  (*buf)->buf_record_cnt - (*buf)->buf_record_skipped,
			  (*buf)->buf_record_skipped, (*buf)->buf_skipped_new_rid,
			  (*buf)->buf_skipped_csn_gt_cons_maxcsn,
			  (*buf)->buf_skipped_up_to_date, (*buf)->buf_skipped_csn_gt_ruv,
			  (*buf)->buf_skipped_csn_covered);

	for ( i = 0; i < (*buf)->buf_num_cscbs; i++ ) {
		clcache_free_cscb ( &(*buf)->buf_cscbs[i] );
	}
	slapi_ch_free((void **)&(*buf)->buf_cscbs);

	if ( (*buf)->buf_cursor ) {

		(*buf)->buf_cursor->c_close ( (*buf)->buf_cursor );
		(*buf)->buf_cursor = NULL;
	}
}

/*
 * Loads a buffer from DB.
 *
 * anchorcsn - passed in for the first load of a replication session;
 * flag	     - DB_SET to load in the key CSN record.
 * 		       DB_NEXT to load in the records greater than key CSN.
 * return    - DB error code instead of cl5 one because of the
 *		       historic reason.
 */
int
clcache_load_buffer ( CLC_Buffer *buf, CSN **anchorCSN, int *continue_on_miss )
{
	int rc = 0;
        int flag = DB_NEXT;

	if (anchorCSN) *anchorCSN = NULL;
	clcache_refresh_local_maxcsns ( buf );

	if (buf->buf_load_cnt == 0 ) {
		clcache_refresh_consumer_maxcsns ( buf );
		rc = clcache_initial_anchorcsn ( buf, &flag );
        } else {
		rc = clcache_adjust_anchorcsn ( buf, &flag );
	}

	if ( rc == 0 ) {

		buf->buf_state = CLC_STATE_READY;
		if (anchorCSN) *anchorCSN = buf->buf_current_csn;
		rc = clcache_load_buffer_bulk ( buf, flag );

		if (rc == DB_NOTFOUND && continue_on_miss && *continue_on_miss) {
			/* make replication going using next best startcsn */
			slapi_log_err(SLAPI_LOG_ERR, buf->buf_agmt_name,
					"clcache_load_buffer - Can't load changelog buffer starting at CSN %s with flag(%s). "
					"Trying to use an alterantive start CSN.\n",
					(char*)buf->buf_key.data,
					flag==DB_NEXT?"DB_NEXT":"DB_SET" );
			rc = clcache_load_buffer_bulk ( buf, DB_SET_RANGE );
			if (rc == 0) {
				slapi_log_err(SLAPI_LOG_ERR, buf->buf_agmt_name,
					"clcache_load_buffer - Using alternative start iteration csn: %s \n",
					(char*)buf->buf_key.data);
			}
			/* the use of alternative start csns can be limited, record its usage */
			(*continue_on_miss)--;
		}
		/* Reset some flag variables */
		if ( rc == 0 ) {
			int i;
			for ( i = 0; i < buf->buf_num_cscbs; i++ ) {
				buf->buf_cscbs[i]->state = CLC_STATE_READY;
			}
		}
		else {
			slapi_log_err(SLAPI_LOG_ERR, buf->buf_agmt_name,
					"clcache_load_buffer - Can't locate CSN %s in the changelog (DB rc=%d). "
					"If replication stops, the consumer may need to be reinitialized.\n",
					(char*)buf->buf_key.data, rc );
		}
	} else if (rc == CLC_STATE_DONE) {
		rc = DB_NOTFOUND;
	}

	if ( rc != 0 ) {
		slapi_log_err(SLAPI_LOG_REPL, buf->buf_agmt_name,
				"clcache_load_buffer - rc=%d\n", rc );
	}

	return rc;
}

static int
clcache_load_buffer_bulk ( CLC_Buffer *buf, int flag )
{
	DB_TXN *txn = NULL;
	DBC *cursor = NULL;
	int rc = 0;
	int tries = 0;
	int use_flag = flag;

#if 0 /* txn control seems not improving anything so turn it off */
	if ( *(_pool->pl_dbenv) ) {
		txn_begin( *(_pool->pl_dbenv), NULL, &txn, 0 );
	}
#endif

	if (NULL == buf) {
		slapi_log_err(SLAPI_LOG_ERR, buf->buf_agmt_name, 
				"clcache_load_buffer_bulk - NULL buf\n" );
		return rc;
	}
	if (NULL == buf->buf_busy_list) {
		slapi_log_err(SLAPI_LOG_ERR, buf->buf_agmt_name, "clcache_load_buffer_bulk - "
		                  "%s%sno buf_busy_list\n",
		                  buf->buf_agmt_name?buf->buf_agmt_name:"",
		                  buf->buf_agmt_name?": ":"" );
		return rc;
	}

	PR_Lock ( buf->buf_busy_list->bl_lock );
retry:
	if ( 0 == ( rc = clcache_open_cursor ( txn, buf, &cursor )) ) {

		if ( use_flag == DB_NEXT ) {
			/* For bulk read, position the cursor before read the next block */
			rc = cursor->c_get ( cursor,
								 & buf->buf_key,
								 & buf->buf_data,
								 DB_SET );
		}

		/*
		 * Continue if the error is no-mem since we don't need to
		 * load in the key record anyway with DB_SET.
		 */
		if ( 0 == rc || DB_BUFFER_SMALL == rc ) {
			rc = clcache_cursor_get ( cursor, buf, use_flag );
		}

	}

	/*
	 * Don't keep a cursor open across the whole replication session.
	 * That had caused noticeable DB resource contention.
	 */
	if ( cursor ) {
		cursor->c_close ( cursor );
		cursor = NULL;
	}
	if ((rc == DB_LOCK_DEADLOCK) && (tries < MAX_TRIALS)) {
		PRIntervalTime interval;

		tries++;
		slapi_log_err(SLAPI_LOG_TRACE, buf->buf_agmt_name, "clcache_load_buffer_bulk - "
		                  "deadlock number [%d] - retrying\n", tries );
		/* back off */
		interval = PR_MillisecondsToInterval(slapi_rand() % 100);
		DS_Sleep(interval);
		use_flag = flag;
		goto retry;
	}
	if ((rc == DB_LOCK_DEADLOCK) && (tries >= MAX_TRIALS)) {
		slapi_log_err(SLAPI_LOG_REPL, buf->buf_agmt_name, "clcache_load_buffer_bulk - "
		                  "could not load buffer from changelog after %d tries\n", tries );
	}

#if 0 /* txn control seems not improving anything so turn it off */
	if ( txn ) {
		txn->commit ( txn, DB_TXN_NOSYNC );
	}
#endif

	PR_Unlock ( buf->buf_busy_list->bl_lock );

	buf->buf_record_ptr = NULL;
	if ( 0 == rc ) {
		DB_MULTIPLE_INIT ( buf->buf_record_ptr, &buf->buf_data );
		if ( NULL == buf->buf_record_ptr )
			rc = DB_NOTFOUND;
		else
			buf->buf_load_cnt++;
	}

	return rc;
}

/*
 * Gets the next change from the buffer.
 * *key	: output - key of the next change, or NULL if no more change
 * *data: output - data of the next change, or NULL if no more change
 */
int
clcache_get_next_change ( CLC_Buffer *buf, void **key, size_t *keylen, void **data, size_t *datalen, CSN **csn )
{
	int skip = 1;
	int rc = 0;

	do {
		*key = *data = NULL;
		*keylen = *datalen = 0;

		if ( buf->buf_record_ptr ) {
			DB_MULTIPLE_KEY_NEXT ( buf->buf_record_ptr, &buf->buf_data,
								   *key, *keylen, *data, *datalen );
		}

		/*
		 * We're done with the current buffer. Now load the next chunk.
		 */
		if ( NULL == *key && CLC_STATE_READY == buf->buf_state ) {
			rc = clcache_load_buffer ( buf, NULL, NULL );
			if ( 0 == rc && buf->buf_record_ptr ) {
				DB_MULTIPLE_KEY_NEXT ( buf->buf_record_ptr, &buf->buf_data,
								   *key, *keylen, *data, *datalen );
			}
		}

		/* Compare the new change to the local and remote RUVs */
		if ( NULL != *key ) {
			buf->buf_record_cnt++;
			csn_init_by_string ( buf->buf_current_csn, (char*)*key );
			skip = clcache_skip_change ( buf );
			if (skip) buf->buf_record_skipped++;
		}
	}
	while ( rc == 0 && *key && skip );

	if ( NULL == *key ) {
		*key = NULL;
		*csn = NULL;
		rc = DB_NOTFOUND;
	}
	else {
		*csn = buf->buf_current_csn;
		slapi_log_err(SLAPI_LOG_REPL, buf->buf_agmt_name,
			"clcache_get_next_change - load=%d rec=%d csn=%s\n",
			buf->buf_load_cnt, buf->buf_record_cnt, (char*)*key );
	}

	return rc;
}

static void
clcache_refresh_consumer_maxcsns ( CLC_Buffer *buf )
{
	int i;

	for ( i = 0; i < buf->buf_num_cscbs; i++ ) {
		ruv_get_largest_csn_for_replica (
				buf->buf_consumer_ruv,
				buf->buf_cscbs[i]->rid,
				&buf->buf_cscbs[i]->consumer_maxcsn );
	}
}

static int
clcache_refresh_local_maxcsn ( const ruv_enum_data *rid_data, void *data )
{
	struct clc_buffer *buf = (struct clc_buffer*) data;
	ReplicaId rid;
	int rc = 0;
	int i;

	rid = csn_get_replicaid ( rid_data->csn );
	/* we do not handle updates originated at the consumer if not required
	 * and we ignore RID which have been cleaned
	 */
	if ( (rid == buf->buf_consumer_rid && buf->buf_ignoreConsumerRID) ||
		is_cleaned_rid(rid) )
		return rc;

	for ( i = 0; i < buf->buf_num_cscbs; i++ ) {
		if ( buf->buf_cscbs[i]->rid == rid )
			break;
	}
	if ( i >= buf->buf_num_cscbs ) {
		if( i + 1 > buf->buf_max_cscbs){
			buf->buf_cscbs = (struct csn_seq_ctrl_block **) slapi_ch_realloc((char *)buf->buf_cscbs,
							 (i + 2) * sizeof(struct csn_seq_ctrl_block *));
			buf->buf_max_cscbs = i + 1;
		}
		buf->buf_cscbs[i] = clcache_new_cscb();
		if ( buf->buf_cscbs[i] == NULL ) {
			return -1;
		}
		buf->buf_cscbs[i]->rid = rid;
		buf->buf_num_cscbs++;
		/* this is the first time we have a local change for the RID
		 * we need to check what the consumer knows about it.
		 */
		ruv_get_largest_csn_for_replica (
				buf->buf_consumer_ruv,
				buf->buf_cscbs[i]->rid,
				&buf->buf_cscbs[i]->consumer_maxcsn );
	}

	if (buf->buf_cscbs[i]->local_maxcsn)
	    csn_dup_or_init_by_csn ( &buf->buf_cscbs[i]->prev_local_maxcsn, buf->buf_cscbs[i]->local_maxcsn );

	csn_dup_or_init_by_csn ( &buf->buf_cscbs[i]->local_maxcsn, rid_data->csn );
	csn_dup_or_init_by_csn ( &buf->buf_cscbs[i]->local_mincsn, rid_data->min_csn );

	if ( buf->buf_cscbs[i]->consumer_maxcsn &&
		 csn_compare (buf->buf_cscbs[i]->consumer_maxcsn, rid_data->csn) >= 0 ) {
		/* No change need to be sent for this RID */
		buf->buf_cscbs[i]->state = CLC_STATE_UP_TO_DATE;
	}

	return rc;
}

static int
clcache_refresh_local_maxcsns ( CLC_Buffer *buf )
{

	return ruv_enumerate_elements ( buf->buf_local_ruv, clcache_refresh_local_maxcsn, buf );
}

/*
 * Algorithm:
 *
 *	1. Determine anchorcsn for each RID:
 *	2. Determine anchorcsn for next load:
 *	   Anchor-CSN = min { all Next-Anchor-CSN, Buffer-MaxCSN }
 */
static int
clcache_initial_anchorcsn ( CLC_Buffer *buf, int *flag )
{
	PRBool hasChange = PR_FALSE;
	struct csn_seq_ctrl_block *cscb;
	int i;
	CSN *anchorcsn = NULL;

	if ( buf->buf_state == CLC_STATE_READY ) {
		for ( i = 0; i < buf->buf_num_cscbs; i++ ) {
			CSN *rid_anchor = NULL;
			int rid_flag = DB_NEXT;
			cscb = buf->buf_cscbs[i];

			if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
				char prevmax[CSN_STRSIZE];
				char local[CSN_STRSIZE];
				char curr[CSN_STRSIZE];
				char conmaxcsn[CSN_STRSIZE];
				csn_as_string(cscb->prev_local_maxcsn, 0, prevmax);
				csn_as_string(cscb->local_maxcsn, 0, local);
				csn_as_string(buf->buf_current_csn, 0, curr);
				csn_as_string(cscb->consumer_maxcsn, 0, conmaxcsn);
				slapi_log_err(SLAPI_LOG_REPL, buf->buf_agmt_name,
						"clcache_initial_anchorcsn - " 
						"%s - (cscb %d - state %d) - csnPrevMax (%s) "
						"csnMax (%s) csnBuf (%s) csnConsumerMax (%s)\n",
						buf->buf_agmt_name, i, cscb->state, prevmax, local,
						curr, conmaxcsn);
			}

			if (cscb->consumer_maxcsn == NULL) {
				/* the consumer hasn't seen changes for this RID */
				rid_anchor = cscb->local_mincsn;
				rid_flag = DB_SET;
			} else if ( csn_compare (cscb->local_maxcsn, cscb->consumer_maxcsn) > 0 ) {
				rid_anchor = cscb->consumer_maxcsn;
			}

			if (rid_anchor && (anchorcsn == NULL ||
			    ( csn_compare(rid_anchor, anchorcsn) < 0))) {
				anchorcsn = rid_anchor;
				*flag = rid_flag;
				hasChange = PR_TRUE;
			}
		}
	}

	if ( !hasChange ) {
		buf->buf_state = CLC_STATE_DONE;
	} else {
		csn_init_by_csn(buf->buf_current_csn, anchorcsn);
		csn_as_string(buf->buf_current_csn, 0, (char *)buf->buf_key.data);
		slapi_log_err(SLAPI_LOG_REPL, "clcache_initial_anchorcsn",
						"anchor is now: %s\n", (char *)buf->buf_key.data);
	}

	return buf->buf_state;
}

static int
clcache_adjust_anchorcsn ( CLC_Buffer *buf, int *flag )
{
	PRBool hasChange = PR_FALSE;
	struct csn_seq_ctrl_block *cscb;
	int i;
	CSN *anchorcsn = NULL;

	if ( buf->buf_state == CLC_STATE_READY ) {
		for ( i = 0; i < buf->buf_num_cscbs; i++ ) {
			CSN *rid_anchor = NULL;
			int rid_flag = DB_NEXT;
			cscb = buf->buf_cscbs[i];

			if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
				char prevmax[CSN_STRSIZE];
				char local[CSN_STRSIZE];
				char curr[CSN_STRSIZE];
				char conmaxcsn[CSN_STRSIZE];
				csn_as_string(cscb->prev_local_maxcsn, 0, prevmax);
				csn_as_string(cscb->local_maxcsn, 0, local);
				csn_as_string(buf->buf_current_csn, 0, curr);
				csn_as_string(cscb->consumer_maxcsn, 0, conmaxcsn);
				slapi_log_err(SLAPI_LOG_REPL, buf->buf_agmt_name, "clcache_adjust_anchorcsn - "
								"%s - (cscb %d - state %d) - csnPrevMax (%s) "
								"csnMax (%s) csnBuf (%s) csnConsumerMax (%s)\n",
								buf->buf_agmt_name, i, cscb->state, prevmax, local,
								curr, conmaxcsn);
			}

			if (csn_compare(cscb->local_maxcsn, cscb->consumer_maxcsn) > 0) {
				/* We have something to send for this RID */

				if (csn_compare(cscb->local_maxcsn, cscb->prev_local_maxcsn) == 0 ||
					csn_compare(cscb->prev_local_maxcsn, buf->buf_current_csn) > 0) {
					/* No new changes or it remains, in the buffer, updates to send  */
					rid_anchor = buf->buf_current_csn;
				} else {
					/* prev local max csn < csnBuffer AND different from local maxcsn */
					if (cscb->consumer_maxcsn == NULL) {
						/* the consumer hasn't seen changes for this RID */
						rid_anchor = cscb->local_mincsn;
						rid_flag = DB_SET;
					} else {
						rid_anchor = cscb->consumer_maxcsn;
					}
				}
			}

			if (rid_anchor && (anchorcsn == NULL ||
			    ( csn_compare(rid_anchor, anchorcsn) < 0))) {
				anchorcsn = rid_anchor;
				*flag = rid_flag;
				hasChange = PR_TRUE;
			}
		}
	}

	if ( !hasChange ) {
		buf->buf_state = CLC_STATE_DONE;
	} else {
		csn_init_by_csn(buf->buf_current_csn, anchorcsn);
		csn_as_string(buf->buf_current_csn, 0, (char *)buf->buf_key.data);
		slapi_log_err(SLAPI_LOG_REPL, buf->buf_agmt_name, "clcache_adjust_anchorcsn - "
						"anchor is now: %s\n", (char *)buf->buf_key.data);
	}

	return buf->buf_state;
}

static int
clcache_skip_change ( CLC_Buffer *buf )
{
	struct csn_seq_ctrl_block *cscb = NULL;
	ReplicaId rid;
	int skip = 1;
	int i;
	char buf_cur_csn_str[CSN_STRSIZE];

	do {

		rid = csn_get_replicaid ( buf->buf_current_csn );

		/*
		 * Skip CSN that is originated from the consumer,
		 * unless the CSN is newer than the maxcsn.
		 * If RID==65535, the CSN is originated from a
		 * legacy consumer. In this case the supplier
		 * and the consumer may have the same RID.
		 */
		if (rid == buf->buf_consumer_rid && buf->buf_ignoreConsumerRID){
			if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
				csn_as_string(buf->buf_current_csn, 0, buf_cur_csn_str);
				slapi_log_err(SLAPI_LOG_REPL, buf->buf_agmt_name,
					"clcache_skip_change - Skipping update because the consumer with Rid: [%d] is ignored\n", rid);
				buf->buf_skipped_csn_gt_cons_maxcsn++;
			}
			break;
		}

		/* Skip helper entry (ENTRY_COUNT, PURGE_RUV and so on) */
		if ( cl5HelperEntry ( NULL, buf->buf_current_csn ) == PR_TRUE ) {
			slapi_log_err(SLAPI_LOG_REPL, buf->buf_agmt_name,
				"clcache_skip_change - Skip helper entry type=%ld\n", csn_get_time( buf->buf_current_csn ));
			break;
		}

		/* Find csn sequence control block for the current rid */
		for (i = 0; i < buf->buf_num_cscbs && buf->buf_cscbs[i]->rid != rid; i++);

		/* Skip CSN whose RID is unknown to the local RUV snapshot */
		if ( i >= buf->buf_num_cscbs ) {
			if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
				csn_as_string(buf->buf_current_csn, 0, buf_cur_csn_str);
				slapi_log_err(SLAPI_LOG_REPL, buf->buf_agmt_name,
					"clcache_skip_change - Skipping update because the changelog buffer current csn [%s] rid "
				        "[%d] is not in the list of changelog csn buffers (length %d)\n",
				        buf_cur_csn_str, rid, buf->buf_num_cscbs);
			}
			buf->buf_skipped_new_rid++;
			break;
		}

		cscb = buf->buf_cscbs[i];

		/* Skip if the consumer is already up-to-date for the RID */
		if ( cscb->state == CLC_STATE_UP_TO_DATE ) {
			buf->buf_skipped_up_to_date++;
			break;
		}

		/* Skip CSN whose preceedents are not covered by local RUV snapshot */
		if ( cscb->state == CLC_STATE_CSN_GT_RUV ) {
			buf->buf_skipped_csn_gt_ruv++;
			break;
		}

		/* Skip CSNs already covered by consumer RUV */
		if ( cscb->consumer_maxcsn &&
			 csn_compare ( buf->buf_current_csn, cscb->consumer_maxcsn ) <= 0 ) {
			buf->buf_skipped_csn_covered++;
				break;
		}

		/* Send CSNs that are covered by the local RUV snapshot */
		if ( csn_compare ( buf->buf_current_csn, cscb->local_maxcsn ) <= 0 ) {
			skip = 0;
			csn_dup_or_init_by_csn ( &cscb->consumer_maxcsn, buf->buf_current_csn );
			break;
		}

		/*
		 * Promote the local maxcsn to its next neighbor
		 * to keep the current session going. Skip if we
		 * are not sure if current_csn is the neighbor.
		 */
		if ( csn_time_difference(buf->buf_current_csn, cscb->local_maxcsn) == 0 &&
			 (csn_get_seqnum(buf->buf_current_csn) ==
				csn_get_seqnum(cscb->local_maxcsn) + 1) )
		{
			csn_init_by_csn ( cscb->local_maxcsn, buf->buf_current_csn );
			if(cscb->consumer_maxcsn){
				csn_init_by_csn ( cscb->consumer_maxcsn, buf->buf_current_csn );
			}
			skip = 0;
			break;
		}

		/* Skip CSNs not covered by local RUV snapshot */
		cscb->state = CLC_STATE_CSN_GT_RUV;
		buf->buf_skipped_csn_gt_ruv++;

	} while (0);

#ifdef DEBUG
	if (skip && cscb) {
		char consumer[24] = {'\0'};
		char local[24] = {'\0'};
		char current[24] = {'\0'};

		if ( cscb->consumer_maxcsn )
			csn_as_string ( cscb->consumer_maxcsn, PR_FALSE, consumer );
		if ( cscb->local_maxcsn )
			csn_as_string ( cscb->local_maxcsn, PR_FALSE, local );
		csn_as_string ( buf->buf_current_csn, PR_FALSE, current );
		slapi_log_err(SLAPI_LOG_REPL, buf->buf_agmt_name,
				"clcache_skip_change - Skip %s consumer=%s local=%s\n", current, consumer, local );
	}
#endif

	return skip;
}

static struct csn_seq_ctrl_block *
clcache_new_cscb(void)
{
	struct csn_seq_ctrl_block *cscb;

	cscb = (struct csn_seq_ctrl_block *) slapi_ch_calloc ( 1, sizeof (struct csn_seq_ctrl_block) );
	if (cscb == NULL) {
		slapi_log_err(SLAPI_LOG_ERR, NULL, "clcache: malloc failure\n" );
	}
	return cscb;
}

static void
clcache_free_cscb ( struct csn_seq_ctrl_block ** cscb )
{
	csn_free ( & (*cscb)->consumer_maxcsn );
	csn_free ( & (*cscb)->local_maxcsn );
	csn_free ( & (*cscb)->prev_local_maxcsn );
	csn_free ( & (*cscb)->local_mincsn );
	slapi_ch_free ( (void **) cscb );
}

/*
 * Allocate and initialize a new buffer
 * It is called when there is a request for a buffer while
 * buffer free list is empty.
 */
static CLC_Buffer *
clcache_new_buffer ( ReplicaId consumer_rid )
{
	CLC_Buffer *buf = NULL;
	int welldone = 0;

	do {

		buf = (CLC_Buffer*) slapi_ch_calloc (1, sizeof(CLC_Buffer));
		if ( NULL == buf )
			break;

		buf->buf_key.flags = DB_DBT_USERMEM;
		buf->buf_key.ulen = CSN_STRSIZE + 1;
		buf->buf_key.size = CSN_STRSIZE;
		buf->buf_key.data = slapi_ch_calloc( 1, buf->buf_key.ulen );
		if ( NULL == buf->buf_key.data )
			break;

		buf->buf_data.flags = DB_DBT_USERMEM;
		buf->buf_data.ulen = _pool->pl_buffer_default_pages * DEFAULT_CLC_BUFFER_PAGE_SIZE;
		buf->buf_data.data = slapi_ch_malloc( buf->buf_data.ulen );
		if ( NULL == buf->buf_data.data )
			break;

		if ( NULL == ( buf->buf_current_csn = csn_new()) )
			break;

		buf->buf_state = CLC_STATE_READY;
		buf->buf_agmt_name = get_thread_private_agmtname();
		buf->buf_consumer_rid = consumer_rid;
		buf->buf_num_cscbs = 0;
		buf->buf_max_cscbs = MAX_NUM_OF_MASTERS;
		buf->buf_cscbs  = (struct csn_seq_ctrl_block **) slapi_ch_calloc(MAX_NUM_OF_MASTERS + 1,
					sizeof(struct csn_seq_ctrl_block *));

		welldone = 1;

	} while (0);

	if ( !welldone ) {
		clcache_delete_buffer ( &buf );
	}

	return buf;
}

/*
 * Deallocates a buffer.
 * It is called when a buffer is returned to the buffer pool
 * and the pool size is over the limit.
 */
static void
clcache_delete_buffer ( CLC_Buffer **buf )
{
	if ( buf && *buf ) {
		slapi_ch_free (&( (*buf)->buf_key.data ));
		slapi_ch_free (&( (*buf)->buf_data.data ));
		csn_free (&( (*buf)->buf_current_csn ));
		csn_free (&( (*buf)->buf_missing_csn ));
		csn_free (&( (*buf)->buf_prev_missing_csn ));
		slapi_ch_free ( (void **) buf );
	}
}

static CLC_Busy_List *
clcache_new_busy_list(void)
{
	CLC_Busy_List *bl;
	int welldone = 0;

	do {
		if ( NULL == (bl = ( CLC_Busy_List* ) slapi_ch_calloc (1, sizeof(CLC_Busy_List)) ))
			break;

		if ( NULL == (bl->bl_lock = PR_NewLock ()) )
			break;

		/*
		if ( NULL == (bl->bl_max_csn = csn_new ()) )
			break;
		*/

		welldone = 1;
	}
	while (0);

	if ( !welldone ) {
		clcache_delete_busy_list ( &bl );
	}

	return bl;
}

static void
clcache_delete_busy_list ( CLC_Busy_List **bl )
{
	if ( bl && *bl ) {
		CLC_Buffer *buf = NULL;
		if ( (*bl)->bl_lock ) {
			PR_Lock ( (*bl)->bl_lock );
		}
		buf = (*bl)->bl_buffers;
		while (buf) {
			CLC_Buffer *next = buf->buf_next;
			clcache_delete_buffer(&buf);
			buf = next;
		}
		(*bl)->bl_buffers = NULL;
		(*bl)->bl_db = NULL;
		if ( (*bl)->bl_lock ) {
			PR_Unlock ( (*bl)->bl_lock );
			PR_DestroyLock ( (*bl)->bl_lock );
			(*bl)->bl_lock = NULL;
		}
		/* csn_free (&( (*bl)->bl_max_csn )); */
		slapi_ch_free ( (void **) bl );
	}
}

static int
clcache_enqueue_busy_list ( DB *db, CLC_Buffer *buf )
{
	CLC_Busy_List *bl;
	int rc = 0;

	slapi_rwlock_rdlock ( _pool->pl_lock );
	for ( bl = _pool->pl_busy_lists; bl && bl->bl_db != db; bl = bl->bl_next );
	slapi_rwlock_unlock ( _pool->pl_lock );

	if ( NULL == bl ) {
		if ( NULL == ( bl = clcache_new_busy_list ()) ) {
			rc = CL5_MEMORY_ERROR;
		}
		else {
			slapi_rwlock_wrlock ( _pool->pl_lock );
			bl->bl_db = db;
			bl->bl_next = _pool->pl_busy_lists;
			_pool->pl_busy_lists = bl;
			slapi_rwlock_unlock ( _pool->pl_lock );
		}
	}

	if ( NULL != bl ) {
		PR_Lock ( bl->bl_lock );
		buf->buf_busy_list = bl;
		buf->buf_next = bl->bl_buffers;
		bl->bl_buffers = buf;
		PR_Unlock ( bl->bl_lock );
	}

	return rc;
}

static int
clcache_open_cursor ( DB_TXN *txn, CLC_Buffer *buf, DBC **cursor )
{
	int rc;

	rc = buf->buf_busy_list->bl_db->cursor ( buf->buf_busy_list->bl_db, txn, cursor, 0 );
	if ( rc != 0 ) {
		slapi_log_err(SLAPI_LOG_ERR, get_thread_private_agmtname(),
			"clcache: failed to open cursor; db error - %d %s\n",
			rc, db_strerror(rc));
	}

	return rc;
}

static int
clcache_cursor_get ( DBC *cursor, CLC_Buffer *buf, int flag )
{
	int rc;

	if (buf->buf_data.ulen > WORK_CLC_BUFFER_PAGE_SIZE) {
		/*
		 * The buffer size had to be increased,
		 * reset it to a smaller working size,
		 * if not sufficient it will be increased again
		 */
		buf->buf_data.ulen = WORK_CLC_BUFFER_PAGE_SIZE;
	}

	rc = cursor->c_get ( cursor,
						 & buf->buf_key,
						 & buf->buf_data,
						 buf->buf_load_flag | flag );
	if ( DB_BUFFER_SMALL == rc ) {
		/*
		 * The record takes more space than the current size of the
		 * buffer. Fortunately, buf->buf_data.size has been set by
		 * c_get() to the actual data size needed. So we can
		 * reallocate the data buffer and try to read again.
		 */
		buf->buf_data.ulen = ( buf->buf_data.size / DEFAULT_CLC_BUFFER_PAGE_SIZE + 1 ) * DEFAULT_CLC_BUFFER_PAGE_SIZE;
		buf->buf_data.data = slapi_ch_realloc ( buf->buf_data.data, buf->buf_data.ulen );
		if ( buf->buf_data.data != NULL ) {
			rc = cursor->c_get ( cursor,
								 &( buf->buf_key ),
								 &( buf->buf_data ),
							 	 buf->buf_load_flag | flag );
			slapi_log_err(SLAPI_LOG_REPL, buf->buf_agmt_name,
				"clcache_cursor_get - clcache: (%d | %d) buf key len %d reallocated and retry returns %d\n", buf->buf_load_flag, flag, buf->buf_key.size, rc );
		}
	}

	switch ( rc ) {
		case EINVAL:
			slapi_log_err(SLAPI_LOG_ERR, buf->buf_agmt_name,
					"clcache_cursor_get - invalid parameter\n" );
			break;

		case DB_BUFFER_SMALL:
			slapi_log_err(SLAPI_LOG_ERR, buf->buf_agmt_name,
					"clcache_cursor_get - can't allocate %u bytes\n", buf->buf_data.ulen );
			break;

		default:
			break;
	}

	return rc;
}

static void
csn_dup_or_init_by_csn ( CSN **csn1, CSN *csn2 )
{
	if ( *csn1 == NULL )
		*csn1 = csn_new();
	csn_init_by_csn ( *csn1, csn2 );
}

void
clcache_destroy()
{
	if (_pool) {
		CLC_Busy_List *bl = NULL;
		if (_pool->pl_lock) {
			slapi_rwlock_wrlock (_pool->pl_lock);
		}

		bl = _pool->pl_busy_lists;
		while (bl) {
			CLC_Busy_List *next = bl->bl_next;
			clcache_delete_busy_list(&bl);
			bl = next;
		}
		_pool->pl_busy_lists = NULL;
		_pool->pl_dbenv = NULL;
		if (_pool->pl_lock) {
			slapi_rwlock_unlock(_pool->pl_lock);
			slapi_destroy_rwlock(_pool->pl_lock);
			_pool->pl_lock = NULL;
		}
		slapi_ch_free ( (void **) &_pool );
	}
}