<commit_before>/*
 * ZS Logging Container
 *
 * Copyright 2015 SanDisk Corporation.  All rights reserved.
 *
 * A small number of logging containers are supported.  Client threads
 * are tracked passively to determine their current and previous trx.
 * Reference to a new PG by a client causes a structure to be allocated
 * for tracking the state, and the structure is hashed into the container
 * and linked into a stream (selected in round-robin fashion).  The oldest
 * streaming object is deleted when all occupant PGs in the object have been
 * sufficiently trimmed: after reclaiming that space, the next streaming
 * object is read and applicable PG structures prepared afresh.
 *
 * Locking strategy.  For maximum reliabilty, locking is relatively simple
 * with the only optimized call path being lc_write/lc_mput, which are
 * engineered to promote write-combining at nvr level.  The locks in LC:
 *
 *	threadlist_lock		rwlock of local scope to change trx tracking list
 *	lc_lock			rwlock of global scope to change anything
 *	containerstructure.lock	rwlock for changes restricted to one container
 *	streamstructure.buflock	mutex to the stream buffer in fast-write mode
 *
 * Fast-write mode:
 *	This mode uses a specific protocol for locking and data access
 *	to maximize performance of multiple writers, including parallel
 *	calls into nvr services for batch-commit opportunities, by taking
 *	the containerstructure.lock for READING.  The mode mandates
 *	read-only access to LC structures, except those protected
 *	by locks for localized modifications.  Specifically, the two
 *	permitted modifications in fast-write mode are those to track
 *	trx and to append data to a stream.  Stream append in this mode
 *	is not allowed if the buffer is empty or would need flushing.
 *	When fast-write mode cannot proceed due to the these restrictions,
 *	the execution path must be abandoned, and normal write protocol
 *	(containerstructure.lock taken for WRITING) used.
 *
 *	Helpful tips for fast-write mode implementors: use
 *	pglookup_readonly(); access bufnexti only when holding
 *	WRITE containerstructure.lock, or when holding both READ
 *	containerstructure.lock and streamstructure.buflock.
 *
 * To do:
 *	- more messages: errors, init status/stats
 *
 * Reclamation notes:
 *	- if a stream is non-empty on flash, always track status of the oldest SO
 *	- when SO written is sole SO of a stream, calculate status on-the-spot
 *	- when (oldest) SO is reclaimed, immediately scan status of the next SO if present
 *	- reclamation of a stream is driven by writes to that stream as given by rate R
 *	- to reclaim a SO, every contained lrec must be obsolete
 *	- obsolete means lrec has been trimmed, and is not part of a possibly active trx
 *	- trx IDs are unique and increment monotonically forever
 */

#include 	<sys/types.h>
#include 	<sys/stat.h>
#include 	<fcntl.h>
#include	<sys/types.h>
#include	<unistd.h>
#include	<stdlib.h>
#include	<stdio.h>
#include	<pthread.h>
#include	"zs.h"
#include	"lc.h"
#include	"nvr_svc.h"
#include	"ssd/fifo/mcd_osd.h"
#include 	"utils/properties.h"
#include	"platform/logging.h"
#include	"utils/hash.h"
#include 	"common/sdftypes.h"
#include	"sdftcp/locks.h"
#include	"ssd/ssd.h"
#include	"api/fdf_internal.h"
#include	"protocol/action/recovery.h"
#include	"utils/rico.h"


/* resource control for one ZS instance
 */
#define	ZS_MAX_NUM_LC_DEFAULT	16
#define	bytes_per_logging_object		4000

//#define bytes_per_NVR				nvr_len
//#define placement_groups_per_flash_array	(1L << 13)
//#define bytes_per_stream_buffer			nvr_blksize
//#define streams_per_flash_array			(bytes_per_NVR / bytes_per_streaming_object)
//#define placement_groups_per_stream		(placement_groups_per_flash_array / streams_per_flash_array)
//#define bytes_of_stream_buffers			(streams_per_flash_array * bytes_per_stream_buffer)

#define	DIAGNOSTIC		PLAT_LOG_LEVEL_DIAGNOSTIC
#define DEBUG			PLAT_LOG_LEVEL_DEBUG
#define	INFO			PLAT_LOG_LEVEL_INFO
#define	ERROR			PLAT_LOG_LEVEL_ERROR
#define	FATAL			PLAT_LOG_LEVEL_FATAL
#define	INITIAL			PLAT_LOG_ID_INITIAL
#define	msg( id, lev, ...)	plat_log_msg( id, PLAT_LOG_CAT_SDF_APP_MEMCACHED_RECOVERY, lev, __VA_ARGS__)

#define	COUNTER_UNKNOWN		(~0)
#define	LC_MAGIC		0x81F191E1A2C2B292
#define	LOTABLESIZE		(1uL << 10)		/* buckets of LOs for enumeration */


/*
 * prevent tearing of shared scalers when assigned
 */
#define	atomic_assign( dst, val)	__sync_lock_test_and_set( (dst), (val))

ZS_ext_stat_t log_to_zs_stats_map[] = {
	{LOGSTAT_NUM_OBJS,		ZS_BTREE_NUM_OBJS,	ZS_STATS_TYPE_BTREE,0},
	{LOGSTAT_DELETE_CNT,		ZS_ACCESS_TYPES_DELETE,	ZS_STATS_TYPE_APP_REQ, 0},
	{LOGSTAT_FLUSH_CNT,		ZS_BTREE_FLUSH_CNT,	ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_IO_SAVED,		ZS_BTREE_MPUT_IO_SAVED,	ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_CNT,		ZS_ACCESS_TYPES_MPUT,	ZS_STATS_TYPE_APP_REQ, 0},
	{LOGSTAT_NUM_MPUT_OBJS,		ZS_BTREE_NUM_MPUT_OBJS,	ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_READ_CNT,		ZS_ACCESS_TYPES_READ,	ZS_STATS_TYPE_APP_REQ, 0},
	{LOGSTAT_WRITE_CNT,		ZS_ACCESS_TYPES_WRITE,	ZS_STATS_TYPE_APP_REQ, 0},
	{LOGSTAT_WRITE_SLOWPATH,	ZS_BTREE_LEAF_L1_MISSES, ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_SLOWPATH_TMPBUF,	ZS_BTREE_NONLEAF_L1_MISSES, ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_SLOWPATH_GTTMPBUF,ZS_BTREE_OVERFLOW_L1_MISSES, ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_SLOWPATH_FIRSTREC,ZS_BTREE_BACKUP_L1_MISSES, ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_SLOWPATH_NOSPC, 	ZS_BTREE_BACKUP_L1_HITS,ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_SLOWPATH_DIFFSTR,	ZS_BTREE_LEAF_L1_WRITES,ZS_STATS_TYPE_BTREE, 0},
};

typedef struct ZS_thread_state		ts_t;
typedef pthread_mutex_t			mutex_t;
typedef pthread_rwlock_t		lock_t;
typedef struct threadstructure		thread_t;
typedef struct streamstructure		stream_t;
typedef struct pgstructure		pg_t;
typedef struct pgnamestructure		pgname_t;
typedef struct containerstructure	container_t;
typedef struct lrecstructure		lrec_t;
typedef struct iterstructure		iter_t;
typedef struct lostructure		lo_t;
typedef struct sostructure		so_t;
typedef struct trxrevstructure		trxrev_t;
typedef struct nvstructure		nv_t;
typedef uint32_t			pghash_t;
typedef uint64_t			stream_seqno_t,
					trxid_t,
					counter_t;

/*
 * LC structures
 *
 * threadstructure:
 *	Used to track active trx of an incoming LC client.
 * sostructure:
 *	Structure of a streaming object as stored on flash as a regular
 *	ZS object, preferably of the same slab class as the raw objects
 *	used by Storm.	The open-ended buffer is packed with a whole
 *	number of logging records (lrec_t).  Size of the SO is recorded by
 *	the hosting facility (ZS hash table or nvr services).  Header is
 *	self-describing to allow full recovery of memory structures
 *	on restart.
 * streamstructure:
 *	Used to track stream activities, and associations with container
 *	and aging PGs.	Has a dedicated SO buffer, and an attached
 *	nvr_buffer for persisting LOs with low write amp.
 * pgnamestructure:
 *	Records the offset of counter, pg and osd parts of LO key.
 *	May store the key itself in an internal buffer.
 * pgstructure:
 *	Used to track activities on a PG. The first linked list allows
 *	lookup by the container.  A PG represented in the oldest SO
 *	of its stream is "aging" and linked into the second list for
 *	reclamation processing.
 * containerstructure:
 *	Allows PG lookup, and assignment of its dedicated streams to
 *	new PGs.
 * lrecstructure:
 *	A logging record efficiently encapsulates LOs written by the
 *	client, and also delete/trim operations.  Header is sufficiently
 *	descriptive to allow reclamation decisions.
 * iterstructure:
 *	Constructed and returned to client for object enumeration of a PG.
 * lostructure:
 *	Used to process LOs for enumeration.
 * trxrevstructure:
 *	Used to convey trx IDs into the LC for crash roll back.
 * struct nvstructure:
 *	Manages allocation of NVRAM to containers.  Buffers with live
 *	data are reserved until the owning container is opened and
 *	flushes them to flash.	Buffer with no data can be allocated.
 */
struct threadstructure {
	thread_t	*next;
	trxid_t		trxcurrent,
			trxprevious;
};
struct sostructure {
	ulong		magic;
	so_t		*next;
	ZS_cguid_t	cguid;
	uint		stream;
	stream_seqno_t	seqno;
	char		buffer[];
};
struct streamstructure {
	stream_t	*next;
	uint		stream;
	container_t	*container;
	nvr_buffer_t	*nb;
	stream_seqno_t	seqnext,
			seqoldest;
	pg_t		*pg;
	mutex_t		buflock;
	uint		bufnexti;
	trxid_t		newest;
	so_t		so;
};
struct pgnamestructure {
	uint		pgbase,
			pglen,
			len;
	char		buf[0];
};
struct pgstructure {
	bool		aging;
	pghash_t	hash;
	pg_t		*containerlink,
			*streamlink;
	stream_t	*stream;
	counter_t	trimposition,
			newest;
	pgname_t	name;
};
struct containerstructure {
	ZS_cguid_t	cguid;
	lock_t		lock;
	uint		stream;
	stream_t	*streams,
			*streamhand;
	char		*trxrevfile;
	pg_t		*pgtable[1000];
	log_stats_t lgstats;
};
struct lrecstructure {
	ushort		type,
			klen,
			dlen;
	trxid_t		trx;
	counter_t	counter;
	char		payload[bytes_per_logging_object];
};
struct iterstructure {
	ulong		magic;
	ZS_cguid_t	cguid;			/* must match offset within ZS_iterator */
	lo_t		*lo[LOTABLESIZE];
	so_t		*so;
	uint		objecti,
			nobject;
	lrec_t		**objects;
};
struct lostructure {
	lo_t		*next;
	lrec_t		*lrec;
};
struct trxrevstructure {
	trxid_t 	*vec;
	uint		num;
};
struct nvstructure {
	nvr_buffer_t	*nb;
	ZS_cguid_t	cguid;
};

/* lrecstructure.type
 */
enum {
	LR_TOMBSTONE,
	LR_WRITE,
	LR_DELETE,
	LR_TRIM,
};


extern trxid_t __thread		trx_bracket_id;
static uint			streams_per_container,
				logging_containers_per_instance,
				bytes_per_streaming_object,
				bytes_per_so_buffer,
				nvtablen;
static nv_t			*nvtab;
static thread_t __thread	*thread;
static thread_t			*threadlist;
static lock_t			lc_lock			= PTHREAD_RWLOCK_INITIALIZER,
				threadlist_lock		= PTHREAD_RWLOCK_INITIALIZER;
static container_t		*ctable;
static stream_t			*streamfreelist;

static container_t	*clookup( ZS_cguid_t),
			*callocate( );
static ZS_status_t	recover( ts_t *, container_t *),
			sostatusrecover( ts_t *t, container_t *),
			write_record( ts_t *, ZS_cguid_t, uint, char *, uint32_t, char *, uint64_t),
			write_records( ts_t *ts, lrec_t **lrecs, int num_recs, ZS_cguid_t cguid, int *written),
			streaminit( uint),
			streamwrite( ts_t *, stream_t *, lrec_t *),
			streamreclaim( ts_t *, stream_t *),
			soscan( container_t *, stream_t *, trxrev_t *, so_t *, ulong),
			pgstreamscan( ts_t *, pg_t *, char *, uint32_t, char **, uint64_t *),
			pgstreamenumerate( ts_t *, pg_t *, iter_t *),
			nvinit( ),
			nvflush( ts_t *, ZS_cguid_t),
			keydecode( char *, uint32_t, pgname_t *, pghash_t *, counter_t *);
static nvr_buffer_t	*nvalloc( ZS_cguid_t);
static pg_t		*pglookup( container_t *, char *, pgname_t, pghash_t),
			*pglookup_readonly( container_t *, char *, pgname_t, pghash_t),
			*pglookup2( container_t *, stream_t *, char *, pgname_t, pghash_t, counter_t);
static stream_t		*streamalloc( ZS_cguid_t),
			*streamlookup( container_t *, uint);
static bool		trxtrack( ),
			trxrevdistribute( ),
			sostatus( stream_t *, so_t *, uint),
			sostatusreclaimable( stream_t *),
			pgnameequal( pgname_t *, char *, pgname_t),
			lrecbuild( lrec_t *, uint, char *, uint32_t, char *, uint64_t, counter_t),
			keyequal( char *, uint, char *, uint),
			trimmed( counter_t, counter_t);
static trxrev_t		*trxreversions( container_t *),
			*trxrevfree( trxrev_t *);
static uint		streamcount( stream_t *),
			lrecsize( lrec_t *);
static void		free_iterator( iter_t *),
			cdeallocate( container_t *),
			sostatusclear( stream_t *),
			dumpparams( );
//static void		streamdump();
static char		*trxrevfilename( ZS_cguid_t),
			*make_sokey( char [], uint, stream_seqno_t);
static int		compar( const void *, const void *);
//static char		*prettynumber( ulong);
static counter_t	newest( counter_t, counter_t);

/*
 * LC entrypoint - initialize LC system
 *
 * Called at end of ZSInit().
 */
ZS_status_t
lc_init( struct ZS_state *zs_state, bool reformat)
{

	unless (trxrevdistribute( ))
		return (ZS_FAILURE);
	ZS_status_t r = nvr_init( zs_state);
	unless (r == ZS_SUCCESS)
		return (r);
	if (reformat)
		nvr_reset( );
	nvinit( );
	dumpparams( );
	bytes_per_streaming_object = nvr_bytes_in_buffer( );
	if (bytes_per_streaming_object < sizeof( so_t)+sizeof( lrec_t))
		return (ZS_FAILURE);
	bytes_per_so_buffer = bytes_per_streaming_object - sizeof( so_t);
	uint streams_per_instance = nvr_buffer_count( );
	unless (logging_containers_per_instance = getProperty_Int( "ZS_MAX_NUM_LC", ZS_MAX_NUM_LC_DEFAULT))
		return (ZS_FAILURE);
	unless (ctable = calloc( logging_containers_per_instance, sizeof *ctable))
		return (ZS_OUT_OF_MEM);
	streams_per_container = divideup( streams_per_instance, logging_containers_per_instance);
	r = streaminit( streams_per_instance);
	unless (r == ZS_SUCCESS)
		return (r);
	return (ZS_SUCCESS);
}


/*
 * LC entrypoint - open a container
 *
 * On open, LC containers are recovered entirely from the regular ZS objects
 * in the underlying hash-mode container.  An object of this kind is called
 * a streaming object (SO).
 *
 * On first open after ZS restart, additional SOs for this container are
 * retrieved from NVRAM hardware, and persisted to flash.  Opening all
 * existing LCs in this way will empty the NVRAM, allowing the hardware
 * to be physically removed.  Also on first open after a ZS restart,
 * uncommitted outer trx are rolled back by means of a recovery packet
 * stored in the file system as generated earlier.  Until the LC is opened,
 * the respective packet must not be disturbed.  Crash recovery is itself
 * crash-safe because the steps are idempotent.
 *
 * After recovery, remaining structures and tracking state for the container
 * are initialized.  This includes the addition of new streams to reach the
 * specified total.  Newly encountered PGs will be assigned to a stream in
 * round-robin order.
 */
ZS_status_t
lc_open( ts_t *t, ZS_cguid_t cguid)
{
	uint	i,
		j;

	char *a = trxrevfilename( cguid);
	unless (a)
		return (ZS_OUT_OF_MEM);
	pthread_rwlock_wrlock( &lc_lock);
	if (clookup( cguid)) {
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_SUCCESS);
	}
	container_t *c = callocate( );
	unless (c) {
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	c->cguid = cguid;
	c->trxrevfile = a;
	ZS_status_t r = nvflush( t, cguid);
	unless (r == ZS_SUCCESS) {
		cdeallocate( c);
		pthread_rwlock_unlock( &lc_lock);
		return (r);
	}
	r = recover( t, c);
	switch (r) {
	case ZS_SUCCESS:
	case ZS_CONTAINER_UNKNOWN:
		for (j=streamcount( c->streams); j<streams_per_container; ++j) {
			stream_t *s = streamalloc( cguid);
			unless (s) {
				unless (j) {
					pthread_rwlock_unlock( &lc_lock);
					return (ZS_FAILURE);
				}
				break;
			}
			s->stream = ++c->stream;
			s->next = c->streams;
			c->streams = s;
			s->container = c;
			s->so.cguid = cguid;
			s->so.stream = s->stream;
		}
		break;
	default:
		cdeallocate( c);
		pthread_rwlock_unlock( &lc_lock);
		return (r);
	}
	unless (c->streamhand = c->streams) {
		cdeallocate( c);
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	r = sostatusrecover( t, c);
	unless (r == ZS_SUCCESS) {
		cdeallocate( c);
		pthread_rwlock_unlock( &lc_lock);
		return (r);
	}
	pthread_rwlock_unlock( &lc_lock);
	if (c->trxrevfile)
		unlink( c->trxrevfile);
	return (ZS_SUCCESS);
}


/*
 * LC entrypoint - write a logging object
 *
 * LO is inserted or updated without distinction.  Fast-write mode is
 * used to improve CPU efficiency and, more importantly, to encourage
 * batch-commit optimization.  As mentioned earlier, fast-mode write is
 * possible under conditions that are quite common: PG already existing,
 * and lrec able to drop into the middle of the stream buffer.  The mutex
 * protecting stream access is held for a short time.
 */
ZS_status_t
lc_write( ts_t *t, ZS_cguid_t cguid, char *k, uint32_t kl, char *d, uint64_t dl)
{
	counter_t	counter;
	pghash_t	h;
	pgname_t	n;
	lrec_t		lr;

	atomic_inc(c->lgstats.stat[LOGSTAT_WRITE_CNT]);

	ZS_status_t r = keydecode( k, kl, &n, &h, &counter);
	unless (r == ZS_SUCCESS)
		return (r);
	unless (lrecbuild( &lr, LR_WRITE, k, kl, d, dl, counter))
		return (ZS_OBJECT_TOO_BIG);
	const uint i = lrecsize( &lr);
	pthread_rwlock_rdlock( &lc_lock);
	container_t *c = clookup( cguid);
	unless ((c)
	and (trxtrack( ))) {
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	pthread_rwlock_rdlock( &c->lock);
	pg_t *pg = pglookup_readonly( c, k, n, h);
	if (pg) {
		stream_t *s = pg->stream;
		pthread_mutex_lock( &s->buflock);
		unless ((s->bufnexti == 0)
		or (bytes_per_so_buffer < i + s->bufnexti)) {
			uint bufnexti = s->bufnexti;
			s->bufnexti += i;
			memcpy( s->so.buffer+bufnexti, &lr, i);
			pthread_mutex_unlock( &s->buflock);
			r = nvr_write_buffer_partial( s->nb, &s->so.buffer[bufnexti], i);
			if (r == ZS_SUCCESS) {
				atomic_inc(c->lgstats.stat[LOGSTAT_NUM_OBJS]);
			}
			pthread_rwlock_unlock( &c->lock);
			pthread_rwlock_unlock( &lc_lock);
			return (r);
		}
		pthread_mutex_unlock( &s->buflock);
	}
	pthread_rwlock_unlock( &c->lock);
	pthread_rwlock_unlock( &lc_lock);
	return (write_record( t, cguid, LR_WRITE, k, kl, d, dl));
}


/*
 * LC entrypoint - delete a logging object
 *
 * LO is deleted if it exists, otherwise no effect.  Flash space is never
 * reclaimed by this operation (see lc_trim).
 */
ZS_status_t
lc_delete( ts_t *t, ZS_cguid_t cguid, char *k, uint32_t kl)
{

	atomic_inc(c->lgstats.stat[LOGSTAT_DELETE_CNT]);
	return (write_record( t, cguid, LR_DELETE, k, kl, 0, 0));
}


/*
 * LC entrypoint - trim a PG
 *
 * The specified PG is trimmed at position given by the counter.  All LOs
 * before the trim position are considered deleted.  With sufficient trimming
 * of PGs, flash space will be freed without write amp for further ZS use.
 */
ZS_status_t
lc_trim( ts_t *t, ZS_cguid_t cguid, char *k, uint32_t kl)
{
	atomic_inc(c->lgstats.stat[LOGSTAT_DELETE_CNT]);

	return (write_record( t, cguid, LR_TRIM, k, kl, 0, 0));
}


/*
 * LC entrypoint - read a logging object
 *
 * Object is returned if present, otherwise ZS_FAILURE.
 */
ZS_status_t
lc_read( ts_t *t, ZS_cguid_t cguid, char *k, uint32_t kl, char **d, uint64_t *dl)
{
	counter_t	counter;
	pghash_t	h;
	pgname_t	n;

	pthread_rwlock_rdlock( &lc_lock);
	container_t *c = clookup( cguid);
	unless ((c)
	and (trxtrack( ))) {
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	ZS_status_t r = keydecode( k, kl, &n, &h, &counter);
	unless (r == ZS_SUCCESS) {
		pthread_rwlock_unlock( &lc_lock);
		return (r);
	}
	pthread_rwlock_rdlock( &c->lock);
	pg_t *pg = pglookup( c, k, n, h);
	unless (pg) {
		pthread_rwlock_unlock( &c->lock);
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	atomic_inc(c->lgstats.stat[LOGSTAT_READ_CNT]);
	r = pgstreamscan( t, pg, k, kl, d, dl);
	pthread_rwlock_unlock( &c->lock);
	pthread_rwlock_unlock( &lc_lock);
	return (r);
}


/*
 * LC entrypoint - start enumeration of LOs in a PG
 *
 * Use the returned iterator to enumerate LOs.
 */
ZS_status_t
lc_enum_start( ts_t *t, ZS_cguid_t cguid, void *iterator, char *k, uint32_t kl)
{
	pghash_t	h;
	pgname_t	n;

	pthread_rwlock_rdlock( &lc_lock);
	container_t *c = clookup( cguid);
	unless ((c)
	and (trxtrack( ))) {
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	ZS_status_t r = keydecode( k, kl, &n, &h, 0);
	unless (r == ZS_SUCCESS) {
		pthread_rwlock_unlock( &lc_lock);
		return (r);
	}
	pthread_rwlock_rdlock( &c->lock);
	pg_t *pg = pglookup( c, k, n, h);
	unless (pg) {
		pthread_rwlock_unlock( &c->lock);
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	iter_t *iter = calloc( sizeof *iter, 1);
	unless (iter) {
		pthread_rwlock_unlock( &c->lock);
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_OUT_OF_MEM);
	}
	r = pgstreamenumerate( t, pg, iter);
	unless (r == ZS_SUCCESS) {
		free_iterator( iter);
		pthread_rwlock_unlock( &c->lock);
		pthread_rwlock_unlock( &lc_lock);
		return (r);
	}
	pthread_rwlock_unlock( &c->lock);
	pthread_rwlock_unlock( &lc_lock);
	iter->magic = LC_MAGIC;
	iter->cguid = cguid;
	*(iter_t **)iterator = iter;
	return (ZS_SUCCESS);
}


/*
 * LC entrypoint - continue enumeration of a PG
 *
 * The next LO is returned, otherwise ZS_OBJECT_UNKNOWN.
 */
ZS_status_t
lc_enum_next( void *iterator, char **key, uint32_t *keylen, char **data, uint64_t *datalen)
{

	iter_t *iter = (iter_t *)iterator;
	unless (iter->magic == LC_MAGIC)
		return (ZS_FAILURE);
	if (iter->objecti < iter->nobject) {
		lrec_t *lr = iter->objects[iter->objecti++];
		unless (*key = malloc( lr->klen))
			return (ZS_OUT_OF_MEM);
		unless (*data = malloc( lr->dlen)) {
			free( *key);
			return (ZS_OUT_OF_MEM);
		}
		memcpy( *key, lr->payload, lr->klen);
		*keylen = lr->klen;
		memcpy( *data, &lr->payload[lr->klen], lr->dlen);
		*datalen = lr->dlen;
		return (ZS_SUCCESS);
	}
	return (ZS_OBJECT_UNKNOWN);
}


/*
 * LC entrypoint - finish enumeration of a PG
 *
 * The iterator is freed.
 */
ZS_status_t
lc_enum_finish( void *iterator)
{

	iter_t *iter = (iter_t *)iterator;
	unless (iter->magic == LC_MAGIC)
		return (ZS_FAILURE);
	iter->magic = 0;
	free_iterator( iter);
	return (ZS_SUCCESS);
}
ZS_status_t
lc_mput( ts_t *t, ZS_cguid_t cguid, uint32_t num_objs, ZS_obj_t *objs, uint32_t flags,
					        uint32_t *objs_written)
{

#define TMPBUFSZ	16000

	counter_t	counter;
	pghash_t	h;
	pgname_t	n;
	ZS_status_t	ret = ZS_SUCCESS;
	char		k[100], tmpbuf[TMPBUFSZ];
	lrec_t		lr, *lrecs[TMPBUFSZ / (sizeof(lrec_t) - bytes_per_logging_object)];
	uint		syncoff=0, sz;
	stream_t	*s = 0;
	int		i = 0, committed_keys = 0, szintmpbuf = 0, objsinbuf;
	pg_t		*pg, *spg = NULL;
	int		written;
	int		flag = 0;

	pthread_rwlock_rdlock( &lc_lock);
	container_t *c = clookup( cguid);
	if (c) {
		atomic_inc(c->lgstats.stat[LOGSTAT_MPUT_CNT]);
		atomic_add(c->lgstats.stat[LOGSTAT_NUM_MPUT_OBJS], num_objs);<commit_msg>#000: Reverted tmp buf size TESTS_RAN: Unit test<commit_after>/*
 * ZS Logging Container
 *
 * Copyright 2015 SanDisk Corporation.  All rights reserved.
 *
 * A small number of logging containers are supported.  Client threads
 * are tracked passively to determine their current and previous trx.
 * Reference to a new PG by a client causes a structure to be allocated
 * for tracking the state, and the structure is hashed into the container
 * and linked into a stream (selected in round-robin fashion).  The oldest
 * streaming object is deleted when all occupant PGs in the object have been
 * sufficiently trimmed: after reclaiming that space, the next streaming
 * object is read and applicable PG structures prepared afresh.
 *
 * Locking strategy.  For maximum reliabilty, locking is relatively simple
 * with the only optimized call path being lc_write/lc_mput, which are
 * engineered to promote write-combining at nvr level.  The locks in LC:
 *
 *	threadlist_lock		rwlock of local scope to change trx tracking list
 *	lc_lock			rwlock of global scope to change anything
 *	containerstructure.lock	rwlock for changes restricted to one container
 *	streamstructure.buflock	mutex to the stream buffer in fast-write mode
 *
 * Fast-write mode:
 *	This mode uses a specific protocol for locking and data access
 *	to maximize performance of multiple writers, including parallel
 *	calls into nvr services for batch-commit opportunities, by taking
 *	the containerstructure.lock for READING.  The mode mandates
 *	read-only access to LC structures, except those protected
 *	by locks for localized modifications.  Specifically, the two
 *	permitted modifications in fast-write mode are those to track
 *	trx and to append data to a stream.  Stream append in this mode
 *	is not allowed if the buffer is empty or would need flushing.
 *	When fast-write mode cannot proceed due to the these restrictions,
 *	the execution path must be abandoned, and normal write protocol
 *	(containerstructure.lock taken for WRITING) used.
 *
 *	Helpful tips for fast-write mode implementors: use
 *	pglookup_readonly(); access bufnexti only when holding
 *	WRITE containerstructure.lock, or when holding both READ
 *	containerstructure.lock and streamstructure.buflock.
 *
 * To do:
 *	- more messages: errors, init status/stats
 *
 * Reclamation notes:
 *	- if a stream is non-empty on flash, always track status of the oldest SO
 *	- when SO written is sole SO of a stream, calculate status on-the-spot
 *	- when (oldest) SO is reclaimed, immediately scan status of the next SO if present
 *	- reclamation of a stream is driven by writes to that stream as given by rate R
 *	- to reclaim a SO, every contained lrec must be obsolete
 *	- obsolete means lrec has been trimmed, and is not part of a possibly active trx
 *	- trx IDs are unique and increment monotonically forever
 */

#include 	<sys/types.h>
#include 	<sys/stat.h>
#include 	<fcntl.h>
#include	<sys/types.h>
#include	<unistd.h>
#include	<stdlib.h>
#include	<stdio.h>
#include	<pthread.h>
#include	"zs.h"
#include	"lc.h"
#include	"nvr_svc.h"
#include	"ssd/fifo/mcd_osd.h"
#include 	"utils/properties.h"
#include	"platform/logging.h"
#include	"utils/hash.h"
#include 	"common/sdftypes.h"
#include	"sdftcp/locks.h"
#include	"ssd/ssd.h"
#include	"api/fdf_internal.h"
#include	"protocol/action/recovery.h"
#include	"utils/rico.h"


/* resource control for one ZS instance
 */
#define	ZS_MAX_NUM_LC_DEFAULT	16
#define	bytes_per_logging_object		4000

//#define bytes_per_NVR				nvr_len
//#define placement_groups_per_flash_array	(1L << 13)
//#define bytes_per_stream_buffer			nvr_blksize
//#define streams_per_flash_array			(bytes_per_NVR / bytes_per_streaming_object)
//#define placement_groups_per_stream		(placement_groups_per_flash_array / streams_per_flash_array)
//#define bytes_of_stream_buffers			(streams_per_flash_array * bytes_per_stream_buffer)

#define	DIAGNOSTIC		PLAT_LOG_LEVEL_DIAGNOSTIC
#define DEBUG			PLAT_LOG_LEVEL_DEBUG
#define	INFO			PLAT_LOG_LEVEL_INFO
#define	ERROR			PLAT_LOG_LEVEL_ERROR
#define	FATAL			PLAT_LOG_LEVEL_FATAL
#define	INITIAL			PLAT_LOG_ID_INITIAL
#define	msg( id, lev, ...)	plat_log_msg( id, PLAT_LOG_CAT_SDF_APP_MEMCACHED_RECOVERY, lev, __VA_ARGS__)

#define	COUNTER_UNKNOWN		(~0)
#define	LC_MAGIC		0x81F191E1A2C2B292
#define	LOTABLESIZE		(1uL << 10)		/* buckets of LOs for enumeration */


/*
 * prevent tearing of shared scalers when assigned
 */
#define	atomic_assign( dst, val)	__sync_lock_test_and_set( (dst), (val))

ZS_ext_stat_t log_to_zs_stats_map[] = {
	{LOGSTAT_NUM_OBJS,		ZS_BTREE_NUM_OBJS,	ZS_STATS_TYPE_BTREE,0},
	{LOGSTAT_DELETE_CNT,		ZS_ACCESS_TYPES_DELETE,	ZS_STATS_TYPE_APP_REQ, 0},
	{LOGSTAT_FLUSH_CNT,		ZS_BTREE_FLUSH_CNT,	ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_IO_SAVED,		ZS_BTREE_MPUT_IO_SAVED,	ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_CNT,		ZS_ACCESS_TYPES_MPUT,	ZS_STATS_TYPE_APP_REQ, 0},
	{LOGSTAT_NUM_MPUT_OBJS,		ZS_BTREE_NUM_MPUT_OBJS,	ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_READ_CNT,		ZS_ACCESS_TYPES_READ,	ZS_STATS_TYPE_APP_REQ, 0},
	{LOGSTAT_WRITE_CNT,		ZS_ACCESS_TYPES_WRITE,	ZS_STATS_TYPE_APP_REQ, 0},
	{LOGSTAT_WRITE_SLOWPATH,	ZS_BTREE_LEAF_L1_MISSES, ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_SLOWPATH_TMPBUF,	ZS_BTREE_NONLEAF_L1_MISSES, ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_SLOWPATH_GTTMPBUF,ZS_BTREE_OVERFLOW_L1_MISSES, ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_SLOWPATH_FIRSTREC,ZS_BTREE_BACKUP_L1_MISSES, ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_SLOWPATH_NOSPC, 	ZS_BTREE_BACKUP_L1_HITS,ZS_STATS_TYPE_BTREE, 0},
	{LOGSTAT_MPUT_SLOWPATH_DIFFSTR,	ZS_BTREE_LEAF_L1_WRITES,ZS_STATS_TYPE_BTREE, 0},
};

typedef struct ZS_thread_state		ts_t;
typedef pthread_mutex_t			mutex_t;
typedef pthread_rwlock_t		lock_t;
typedef struct threadstructure		thread_t;
typedef struct streamstructure		stream_t;
typedef struct pgstructure		pg_t;
typedef struct pgnamestructure		pgname_t;
typedef struct containerstructure	container_t;
typedef struct lrecstructure		lrec_t;
typedef struct iterstructure		iter_t;
typedef struct lostructure		lo_t;
typedef struct sostructure		so_t;
typedef struct trxrevstructure		trxrev_t;
typedef struct nvstructure		nv_t;
typedef uint32_t			pghash_t;
typedef uint64_t			stream_seqno_t,
					trxid_t,
					counter_t;

/*
 * LC structures
 *
 * threadstructure:
 *	Used to track active trx of an incoming LC client.
 * sostructure:
 *	Structure of a streaming object as stored on flash as a regular
 *	ZS object, preferably of the same slab class as the raw objects
 *	used by Storm.	The open-ended buffer is packed with a whole
 *	number of logging records (lrec_t).  Size of the SO is recorded by
 *	the hosting facility (ZS hash table or nvr services).  Header is
 *	self-describing to allow full recovery of memory structures
 *	on restart.
 * streamstructure:
 *	Used to track stream activities, and associations with container
 *	and aging PGs.	Has a dedicated SO buffer, and an attached
 *	nvr_buffer for persisting LOs with low write amp.
 * pgnamestructure:
 *	Records the offset of counter, pg and osd parts of LO key.
 *	May store the key itself in an internal buffer.
 * pgstructure:
 *	Used to track activities on a PG. The first linked list allows
 *	lookup by the container.  A PG represented in the oldest SO
 *	of its stream is "aging" and linked into the second list for
 *	reclamation processing.
 * containerstructure:
 *	Allows PG lookup, and assignment of its dedicated streams to
 *	new PGs.
 * lrecstructure:
 *	A logging record efficiently encapsulates LOs written by the
 *	client, and also delete/trim operations.  Header is sufficiently
 *	descriptive to allow reclamation decisions.
 * iterstructure:
 *	Constructed and returned to client for object enumeration of a PG.
 * lostructure:
 *	Used to process LOs for enumeration.
 * trxrevstructure:
 *	Used to convey trx IDs into the LC for crash roll back.
 * struct nvstructure:
 *	Manages allocation of NVRAM to containers.  Buffers with live
 *	data are reserved until the owning container is opened and
 *	flushes them to flash.	Buffer with no data can be allocated.
 */
struct threadstructure {
	thread_t	*next;
	trxid_t		trxcurrent,
			trxprevious;
};
struct sostructure {
	ulong		magic;
	so_t		*next;
	ZS_cguid_t	cguid;
	uint		stream;
	stream_seqno_t	seqno;
	char		buffer[];
};
struct streamstructure {
	stream_t	*next;
	uint		stream;
	container_t	*container;
	nvr_buffer_t	*nb;
	stream_seqno_t	seqnext,
			seqoldest;
	pg_t		*pg;
	mutex_t		buflock;
	uint		bufnexti;
	trxid_t		newest;
	so_t		so;
};
struct pgnamestructure {
	uint		pgbase,
			pglen,
			len;
	char		buf[0];
};
struct pgstructure {
	bool		aging;
	pghash_t	hash;
	pg_t		*containerlink,
			*streamlink;
	stream_t	*stream;
	counter_t	trimposition,
			newest;
	pgname_t	name;
};
struct containerstructure {
	ZS_cguid_t	cguid;
	lock_t		lock;
	uint		stream;
	stream_t	*streams,
			*streamhand;
	char		*trxrevfile;
	pg_t		*pgtable[1000];
	log_stats_t lgstats;
};
struct lrecstructure {
	ushort		type,
			klen,
			dlen;
	trxid_t		trx;
	counter_t	counter;
	char		payload[bytes_per_logging_object];
};
struct iterstructure {
	ulong		magic;
	ZS_cguid_t	cguid;			/* must match offset within ZS_iterator */
	lo_t		*lo[LOTABLESIZE];
	so_t		*so;
	uint		objecti,
			nobject;
	lrec_t		**objects;
};
struct lostructure {
	lo_t		*next;
	lrec_t		*lrec;
};
struct trxrevstructure {
	trxid_t 	*vec;
	uint		num;
};
struct nvstructure {
	nvr_buffer_t	*nb;
	ZS_cguid_t	cguid;
};

/* lrecstructure.type
 */
enum {
	LR_TOMBSTONE,
	LR_WRITE,
	LR_DELETE,
	LR_TRIM,
};


extern trxid_t __thread		trx_bracket_id;
static uint			streams_per_container,
				logging_containers_per_instance,
				bytes_per_streaming_object,
				bytes_per_so_buffer,
				nvtablen;
static nv_t			*nvtab;
static thread_t __thread	*thread;
static thread_t			*threadlist;
static lock_t			lc_lock			= PTHREAD_RWLOCK_INITIALIZER,
				threadlist_lock		= PTHREAD_RWLOCK_INITIALIZER;
static container_t		*ctable;
static stream_t			*streamfreelist;

static container_t	*clookup( ZS_cguid_t),
			*callocate( );
static ZS_status_t	recover( ts_t *, container_t *),
			sostatusrecover( ts_t *t, container_t *),
			write_record( ts_t *, ZS_cguid_t, uint, char *, uint32_t, char *, uint64_t),
			write_records( ts_t *ts, lrec_t **lrecs, int num_recs, ZS_cguid_t cguid, int *written),
			streaminit( uint),
			streamwrite( ts_t *, stream_t *, lrec_t *),
			streamreclaim( ts_t *, stream_t *),
			soscan( container_t *, stream_t *, trxrev_t *, so_t *, ulong),
			pgstreamscan( ts_t *, pg_t *, char *, uint32_t, char **, uint64_t *),
			pgstreamenumerate( ts_t *, pg_t *, iter_t *),
			nvinit( ),
			nvflush( ts_t *, ZS_cguid_t),
			keydecode( char *, uint32_t, pgname_t *, pghash_t *, counter_t *);
static nvr_buffer_t	*nvalloc( ZS_cguid_t);
static pg_t		*pglookup( container_t *, char *, pgname_t, pghash_t),
			*pglookup_readonly( container_t *, char *, pgname_t, pghash_t),
			*pglookup2( container_t *, stream_t *, char *, pgname_t, pghash_t, counter_t);
static stream_t		*streamalloc( ZS_cguid_t),
			*streamlookup( container_t *, uint);
static bool		trxtrack( ),
			trxrevdistribute( ),
			sostatus( stream_t *, so_t *, uint),
			sostatusreclaimable( stream_t *),
			pgnameequal( pgname_t *, char *, pgname_t),
			lrecbuild( lrec_t *, uint, char *, uint32_t, char *, uint64_t, counter_t),
			keyequal( char *, uint, char *, uint),
			trimmed( counter_t, counter_t);
static trxrev_t		*trxreversions( container_t *),
			*trxrevfree( trxrev_t *);
static uint		streamcount( stream_t *),
			lrecsize( lrec_t *);
static void		free_iterator( iter_t *),
			cdeallocate( container_t *),
			sostatusclear( stream_t *),
			dumpparams( );
//static void		streamdump();
static char		*trxrevfilename( ZS_cguid_t),
			*make_sokey( char [], uint, stream_seqno_t);
static int		compar( const void *, const void *);
//static char		*prettynumber( ulong);
static counter_t	newest( counter_t, counter_t);

/*
 * LC entrypoint - initialize LC system
 *
 * Called at end of ZSInit().
 */
ZS_status_t
lc_init( struct ZS_state *zs_state, bool reformat)
{

	unless (trxrevdistribute( ))
		return (ZS_FAILURE);
	ZS_status_t r = nvr_init( zs_state);
	unless (r == ZS_SUCCESS)
		return (r);
	if (reformat)
		nvr_reset( );
	nvinit( );
	dumpparams( );
	bytes_per_streaming_object = nvr_bytes_in_buffer( );
	if (bytes_per_streaming_object < sizeof( so_t)+sizeof( lrec_t))
		return (ZS_FAILURE);
	bytes_per_so_buffer = bytes_per_streaming_object - sizeof( so_t);
	uint streams_per_instance = nvr_buffer_count( );
	unless (logging_containers_per_instance = getProperty_Int( "ZS_MAX_NUM_LC", ZS_MAX_NUM_LC_DEFAULT))
		return (ZS_FAILURE);
	unless (ctable = calloc( logging_containers_per_instance, sizeof *ctable))
		return (ZS_OUT_OF_MEM);
	streams_per_container = divideup( streams_per_instance, logging_containers_per_instance);
	r = streaminit( streams_per_instance);
	unless (r == ZS_SUCCESS)
		return (r);
	return (ZS_SUCCESS);
}


/*
 * LC entrypoint - open a container
 *
 * On open, LC containers are recovered entirely from the regular ZS objects
 * in the underlying hash-mode container.  An object of this kind is called
 * a streaming object (SO).
 *
 * On first open after ZS restart, additional SOs for this container are
 * retrieved from NVRAM hardware, and persisted to flash.  Opening all
 * existing LCs in this way will empty the NVRAM, allowing the hardware
 * to be physically removed.  Also on first open after a ZS restart,
 * uncommitted outer trx are rolled back by means of a recovery packet
 * stored in the file system as generated earlier.  Until the LC is opened,
 * the respective packet must not be disturbed.  Crash recovery is itself
 * crash-safe because the steps are idempotent.
 *
 * After recovery, remaining structures and tracking state for the container
 * are initialized.  This includes the addition of new streams to reach the
 * specified total.  Newly encountered PGs will be assigned to a stream in
 * round-robin order.
 */
ZS_status_t
lc_open( ts_t *t, ZS_cguid_t cguid)
{
	uint	i,
		j;

	char *a = trxrevfilename( cguid);
	unless (a)
		return (ZS_OUT_OF_MEM);
	pthread_rwlock_wrlock( &lc_lock);
	if (clookup( cguid)) {
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_SUCCESS);
	}
	container_t *c = callocate( );
	unless (c) {
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	c->cguid = cguid;
	c->trxrevfile = a;
	ZS_status_t r = nvflush( t, cguid);
	unless (r == ZS_SUCCESS) {
		cdeallocate( c);
		pthread_rwlock_unlock( &lc_lock);
		return (r);
	}
	r = recover( t, c);
	switch (r) {
	case ZS_SUCCESS:
	case ZS_CONTAINER_UNKNOWN:
		for (j=streamcount( c->streams); j<streams_per_container; ++j) {
			stream_t *s = streamalloc( cguid);
			unless (s) {
				unless (j) {
					pthread_rwlock_unlock( &lc_lock);
					return (ZS_FAILURE);
				}
				break;
			}
			s->stream = ++c->stream;
			s->next = c->streams;
			c->streams = s;
			s->container = c;
			s->so.cguid = cguid;
			s->so.stream = s->stream;
		}
		break;
	default:
		cdeallocate( c);
		pthread_rwlock_unlock( &lc_lock);
		return (r);
	}
	unless (c->streamhand = c->streams) {
		cdeallocate( c);
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	r = sostatusrecover( t, c);
	unless (r == ZS_SUCCESS) {
		cdeallocate( c);
		pthread_rwlock_unlock( &lc_lock);
		return (r);
	}
	pthread_rwlock_unlock( &lc_lock);
	if (c->trxrevfile)
		unlink( c->trxrevfile);
	return (ZS_SUCCESS);
}


/*
 * LC entrypoint - write a logging object
 *
 * LO is inserted or updated without distinction.  Fast-write mode is
 * used to improve CPU efficiency and, more importantly, to encourage
 * batch-commit optimization.  As mentioned earlier, fast-mode write is
 * possible under conditions that are quite common: PG already existing,
 * and lrec able to drop into the middle of the stream buffer.  The mutex
 * protecting stream access is held for a short time.
 */
ZS_status_t
lc_write( ts_t *t, ZS_cguid_t cguid, char *k, uint32_t kl, char *d, uint64_t dl)
{
	counter_t	counter;
	pghash_t	h;
	pgname_t	n;
	lrec_t		lr;

	atomic_inc(c->lgstats.stat[LOGSTAT_WRITE_CNT]);

	ZS_status_t r = keydecode( k, kl, &n, &h, &counter);
	unless (r == ZS_SUCCESS)
		return (r);
	unless (lrecbuild( &lr, LR_WRITE, k, kl, d, dl, counter))
		return (ZS_OBJECT_TOO_BIG);
	const uint i = lrecsize( &lr);
	pthread_rwlock_rdlock( &lc_lock);
	container_t *c = clookup( cguid);
	unless ((c)
	and (trxtrack( ))) {
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	pthread_rwlock_rdlock( &c->lock);
	pg_t *pg = pglookup_readonly( c, k, n, h);
	if (pg) {
		stream_t *s = pg->stream;
		pthread_mutex_lock( &s->buflock);
		unless ((s->bufnexti == 0)
		or (bytes_per_so_buffer < i + s->bufnexti)) {
			uint bufnexti = s->bufnexti;
			s->bufnexti += i;
			memcpy( s->so.buffer+bufnexti, &lr, i);
			pthread_mutex_unlock( &s->buflock);
			r = nvr_write_buffer_partial( s->nb, &s->so.buffer[bufnexti], i);
			if (r == ZS_SUCCESS) {
				atomic_inc(c->lgstats.stat[LOGSTAT_NUM_OBJS]);
			}
			pthread_rwlock_unlock( &c->lock);
			pthread_rwlock_unlock( &lc_lock);
			return (r);
		}
		pthread_mutex_unlock( &s->buflock);
	}
	pthread_rwlock_unlock( &c->lock);
	pthread_rwlock_unlock( &lc_lock);
	return (write_record( t, cguid, LR_WRITE, k, kl, d, dl));
}


/*
 * LC entrypoint - delete a logging object
 *
 * LO is deleted if it exists, otherwise no effect.  Flash space is never
 * reclaimed by this operation (see lc_trim).
 */
ZS_status_t
lc_delete( ts_t *t, ZS_cguid_t cguid, char *k, uint32_t kl)
{

	atomic_inc(c->lgstats.stat[LOGSTAT_DELETE_CNT]);
	return (write_record( t, cguid, LR_DELETE, k, kl, 0, 0));
}


/*
 * LC entrypoint - trim a PG
 *
 * The specified PG is trimmed at position given by the counter.  All LOs
 * before the trim position are considered deleted.  With sufficient trimming
 * of PGs, flash space will be freed without write amp for further ZS use.
 */
ZS_status_t
lc_trim( ts_t *t, ZS_cguid_t cguid, char *k, uint32_t kl)
{
	atomic_inc(c->lgstats.stat[LOGSTAT_DELETE_CNT]);

	return (write_record( t, cguid, LR_TRIM, k, kl, 0, 0));
}


/*
 * LC entrypoint - read a logging object
 *
 * Object is returned if present, otherwise ZS_FAILURE.
 */
ZS_status_t
lc_read( ts_t *t, ZS_cguid_t cguid, char *k, uint32_t kl, char **d, uint64_t *dl)
{
	counter_t	counter;
	pghash_t	h;
	pgname_t	n;

	pthread_rwlock_rdlock( &lc_lock);
	container_t *c = clookup( cguid);
	unless ((c)
	and (trxtrack( ))) {
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	ZS_status_t r = keydecode( k, kl, &n, &h, &counter);
	unless (r == ZS_SUCCESS) {
		pthread_rwlock_unlock( &lc_lock);
		return (r);
	}
	pthread_rwlock_rdlock( &c->lock);
	pg_t *pg = pglookup( c, k, n, h);
	unless (pg) {
		pthread_rwlock_unlock( &c->lock);
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	atomic_inc(c->lgstats.stat[LOGSTAT_READ_CNT]);
	r = pgstreamscan( t, pg, k, kl, d, dl);
	pthread_rwlock_unlock( &c->lock);
	pthread_rwlock_unlock( &lc_lock);
	return (r);
}


/*
 * LC entrypoint - start enumeration of LOs in a PG
 *
 * Use the returned iterator to enumerate LOs.
 */
ZS_status_t
lc_enum_start( ts_t *t, ZS_cguid_t cguid, void *iterator, char *k, uint32_t kl)
{
	pghash_t	h;
	pgname_t	n;

	pthread_rwlock_rdlock( &lc_lock);
	container_t *c = clookup( cguid);
	unless ((c)
	and (trxtrack( ))) {
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	ZS_status_t r = keydecode( k, kl, &n, &h, 0);
	unless (r == ZS_SUCCESS) {
		pthread_rwlock_unlock( &lc_lock);
		return (r);
	}
	pthread_rwlock_rdlock( &c->lock);
	pg_t *pg = pglookup( c, k, n, h);
	unless (pg) {
		pthread_rwlock_unlock( &c->lock);
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_FAILURE);
	}
	iter_t *iter = calloc( sizeof *iter, 1);
	unless (iter) {
		pthread_rwlock_unlock( &c->lock);
		pthread_rwlock_unlock( &lc_lock);
		return (ZS_OUT_OF_MEM);
	}
	r = pgstreamenumerate( t, pg, iter);
	unless (r == ZS_SUCCESS) {
		free_iterator( iter);
		pthread_rwlock_unlock( &c->lock);
		pthread_rwlock_unlock( &lc_lock);
		return (r);
	}
	pthread_rwlock_unlock( &c->lock);
	pthread_rwlock_unlock( &lc_lock);
	iter->magic = LC_MAGIC;
	iter->cguid = cguid;
	*(iter_t **)iterator = iter;
	return (ZS_SUCCESS);
}


/*
 * LC entrypoint - continue enumeration of a PG
 *
 * The next LO is returned, otherwise ZS_OBJECT_UNKNOWN.
 */
ZS_status_t
lc_enum_next( void *iterator, char **key, uint32_t *keylen, char **data, uint64_t *datalen)
{

	iter_t *iter = (iter_t *)iterator;
	unless (iter->magic == LC_MAGIC)
		return (ZS_FAILURE);
	if (iter->objecti < iter->nobject) {
		lrec_t *lr = iter->objects[iter->objecti++];
		unless (*key = malloc( lr->klen))
			return (ZS_OUT_OF_MEM);
		unless (*data = malloc( lr->dlen)) {
			free( *key);
			return (ZS_OUT_OF_MEM);
		}
		memcpy( *key, lr->payload, lr->klen);
		*keylen = lr->klen;
		memcpy( *data, &lr->payload[lr->klen], lr->dlen);
		*datalen = lr->dlen;
		return (ZS_SUCCESS);
	}
	return (ZS_OBJECT_UNKNOWN);
}


/*
 * LC entrypoint - finish enumeration of a PG
 *
 * The iterator is freed.
 */
ZS_status_t
lc_enum_finish( void *iterator)
{

	iter_t *iter = (iter_t *)iterator;
	unless (iter->magic == LC_MAGIC)
		return (ZS_FAILURE);
	iter->magic = 0;
	free_iterator( iter);
	return (ZS_SUCCESS);
}
ZS_status_t
lc_mput( ts_t *t, ZS_cguid_t cguid, uint32_t num_objs, ZS_obj_t *objs, uint32_t flags,
					        uint32_t *objs_written)
{

#define TMPBUFSZ	4072

	counter_t	counter;
	pghash_t	h;
	pgname_t	n;
	ZS_status_t	ret = ZS_SUCCESS;
	char		k[100], tmpbuf[TMPBUFSZ];
	lrec_t		lr, *lrecs[TMPBUFSZ / (sizeof(lrec_t) - bytes_per_logging_object)];
	uint		syncoff=0, sz;
	stream_t	*s = 0;
	int		i = 0, committed_keys = 0, szintmpbuf = 0, objsinbuf;
	pg_t		*pg, *spg = NULL;
	int		written;
	int		flag = 0;

	pthread_rwlock_rdlock( &lc_lock);
	container_t *c = clookup( cguid);
	if (c) {
		atomic_inc(c->lgstats.stat[LOGSTAT_MPUT_CNT]);
		atomic_add(c->lgstats.stat[LOGSTAT_NUM_MPUT_OBJS], num_objs);
{'fraction_non_alphanumeric': 0.10236670169507013, 'fraction_numerical': 0.0054827066295060995, 'mean_word_length': 4.46777416937297, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 145, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '27781', 'n_tokens_mistral': 17902, 'n_tokens_neox': 16872, 'n_words': 4652}