<commit_before>/*
 *  The OpenDiamond Platform for Interactive Search
 *  Version 3
 *
 *  Copyright (c) 2002-2005 Intel Corporation
 *  Copyright (c) 2006 <NAME> <<EMAIL>>
 *  Copyright (c) 2007 Carnegie Mellon University
 *  All rights reserved.
 *
 *  This software is distributed under the terms of the Eclipse Public
 *  License, Version 1.0 which can be found in the file named LICENSE.
 *  ANY USE, REPRODUCTION OR DISTRIBUTION OF THIS SOFTWARE CONSTITUTES
 *  RECIPIENT'S ACCEPTANCE OF THIS AGREEMENT
 */

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <stddef.h>
#include <sys/stat.h>
#include <sys/file.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdint.h>
#include <dirent.h>
#include <assert.h>
#include <sys/uio.h>
#include <limits.h>

#include "diamond_consts.h"
#include "diamond_types.h"
#include "lib_tools.h"
#include "lib_log.h"
#include "lib_dctl.h"
#include "dctl_common.h"
#include "ocache_priv.h"
#include "obj_attr.h"
#include "obj_attr.h"
#include "lib_filter.h"
#include "lib_filter_sys.h"
#include "lib_odisk.h"
#include "lib_filterexec.h"
#include "lib_ocache.h"
#include "lib_dconfig.h"
#include "filter_priv.h"


#define	MAX_FNAME	128
#define TEMP_ATTR_BUF_SIZE	1024

/*
 * dctl variables 
 */
static unsigned int    if_cache_table = 1;
static unsigned int    if_cache_oattr = 0;

static int      search_active = 0;
static int      search_done = 0;

static GAsyncQueue *ocache_queue;
static GAsyncQueue *oattr_queue;

struct ocache_start_entry {
	void *cache_table;
};

struct ocache_end_entry {
	int result;
	query_info_t qid;		/* search that created entry */
	filter_exec_mode_t exec_mode;	/* mode when entry was created */
};

struct ocache_ring_entry {
	int type;
	sig_val_t id_sig;
	union {
		struct ocache_start_entry start;/* INSERT_START */
		cache_attr_entry attr;		/* INSERT_IATTR/INSERT_OATTR */
		struct ocache_end_entry end;	/* INSERT_END */
	} u;
};

struct cache_attr {
	attr_record_t*	arec;
	obj_data_t *	obj;
};

struct oattr_ring_entry {
	int type;
	sig_val_t id_sig;
	union {
		sig_val_t fsig;		/* INSERT_START filter signature */
		cache_attr_entry iattr;	/* INSERT_IATTR */
		struct cache_attr oattr; /* INSERT_OATTR add output attr*/
	} u;
};

static pthread_mutex_t shared_mutex = PTHREAD_MUTEX_INITIALIZER;
/*
 * active 
 */
static pthread_cond_t bg_active_cv = PTHREAD_COND_INITIALIZER;


#define SIG_BUF_SIZE	256
#define MAX_FILTER_ARG_NAME 256
#define CACHE_ENTRY_NUM 4096
#define FCACHE_NUM 50
#define MAX_CACHE_ENTRY_NUM 0X1000000
#define MAX_ENTRY_NUM	(2 * MAX_CACHE_ENTRY_NUM)
#define MAX_IATTR_SIZE	4096

static cache_init_obj **init_table;
static fcache_t *filter_cache_table[FCACHE_NUM];
static int      cache_entry_num = 0;	/* for debug purpose */

static sig_val_t ocache_sig = { {0,} };
static sig_val_t oattr_sig = { {0,} };

/* wrappers around g_async_queue_push and pop to get type checking */
static void ocache_queue_push(struct ocache_ring_entry *data)
{
	g_async_queue_push(ocache_queue, (gpointer)data);
}

static struct ocache_ring_entry *ocache_queue_pop(void)
{
	return (struct ocache_ring_entry *)g_async_queue_pop(ocache_queue);
}

static void oattr_queue_push(struct oattr_ring_entry *data)
{
	g_async_queue_push(oattr_queue, (gpointer)data);
}

static struct oattr_ring_entry *oattr_queue_pop(void)
{
	return (struct oattr_ring_entry *)g_async_queue_pop(oattr_queue);
}

/*
 * This could be moved to a support library XXX
 */
int
digest_cal(filter_data_t * fdata, char *fn_name, int numarg, char **filt_args,
	   int blob_len, void *blob, sig_val_t * signature)
{
	struct ciovec *iov;
	unsigned int i, len, n = 0;

	len =	fdata->num_libs +	/* library_signatures */
		1 +			/* filter name */
		numarg +		/* filter arguments */
		1;			/* optional binary blob */

	iov = (struct ciovec *)malloc(len * sizeof(struct ciovec));
	assert(iov != NULL);

	/* include the library signatures */
	for (i = 0; i < fdata->num_libs; i++) {
		iov[n].iov_base = &fdata->lib_info[i].lib_sig;
		iov[n].iov_len = sizeof(sig_val_t);
		n++;
	}

	iov[n].iov_base = fn_name;
	iov[n].iov_len = strlen(fn_name);
	n++;

	/* include the args */
	for (i = 0; i < numarg; i++) {
		if (filt_args[i] == NULL)
			continue;

		len = strlen(filt_args[i]);
		if (len >= MAX_FILTER_ARG_NAME)
			return (EINVAL);

		iov[n].iov_base = filt_args[i];
		iov[n].iov_len = len;
		n++;
	}

	/* include binary blob */
	if (blob_len > 0) {
		iov[n].iov_base = blob;
		iov[n].iov_len = blob_len;
		n++;
	}

	sig_cal_vec(iov, n, signature);
	free(iov);
	return 0;
}

/*
 * This is only used in this file, maybe there is a duplicate implementation
 * in another places?
 */
static void
sig_iattr(cache_attr_set * iattr, sig_val_t * sig)
{
	struct ciovec *iov;
	int i;

	iov = (struct ciovec *)malloc(iattr->entry_num * sizeof(struct ciovec));
	assert(iov != NULL);

	for (i = 0; i < iattr->entry_num; i++) {
		iov[i].iov_base = iattr->entry_data[i];
		iov[i].iov_len = sizeof(cache_attr_entry);
	}

	sig_cal_vec(iov, iattr->entry_num, sig);
	free(iov);
}

static int
attr_in_set(cache_attr_entry * inattr, cache_attr_set * set)
{
	int             j;
	cache_attr_entry *tattr;

	for (j = 0; j < set->entry_num; j++) {
		tattr = set->entry_data[j];
		if (tattr == NULL) {
			printf("null temp_j, something wrong\n");
			continue;
		}
		if (memcmp(inattr, tattr, sizeof(*tattr)) == 0) {
			return (1);
		}
	}
	return (0);

}

/*
 * This to see if attr1 is a strict subset of attr2, if not, then we
 * return 1, otherwise return 0.
 */

static int
compare_attr_set(cache_attr_set * attr1, cache_attr_set * attr2)
{
	int             i;
	cache_attr_entry *temp_i;

	/*
	 * for each item in attr1, see if it exists in attr2.
	 */
	for (i = 0; i < attr1->entry_num; i++) {
		temp_i = attr1->entry_data[i];
		if (temp_i == NULL) {
			printf("null temp_i, something wrong\n");
			continue;
		}

		/*
		 * if this item isn't in the set then return 1 
		 */
		if (attr_in_set(temp_i, attr2) == 0) {
			return 1;
		}
	}
	return 0;
}

int
combine_attr_set(cache_attr_set * attr1, cache_attr_set * attr2)
{
	int             i,
	                j;
	int             found;
	cache_attr_entry *temp_i,
	               *temp_j;
	cache_attr_entry **tmp;

	for (i = 0; i < attr2->entry_num; i++) {
		temp_i = attr2->entry_data[i];
		if (temp_i == NULL) {
			printf("null temp_i, something wrong\n");
			assert(0);
		}
		found = 0;
		for (j = 0; j < attr1->entry_num; j++) {
			temp_j = attr1->entry_data[j];
			if (temp_j == NULL) {
				printf("null temp_j, something wrong\n");
				break;
			}
			if (memcmp(temp_i, temp_j, sizeof(*temp_j)) == 0) {
				attr1->entry_data[j] = temp_i;
				found = 1;
				break;
			}
		}
		/*
		 * no found, add to the tail 
		 */
		if (!found) {
			attr1->entry_data[attr1->entry_num] = temp_i;
			attr1->entry_num++;
			if ((attr1->entry_num % ATTR_ENTRY_NUM) == 0) {
				tmp = calloc(1, (attr1->entry_num +
						 ATTR_ENTRY_NUM) *
					     sizeof(char *));
				assert(tmp != NULL);
				memcpy(tmp, attr1->entry_data,
				       attr1->entry_num * sizeof(char *));
				free(attr1->entry_data);
				attr1->entry_data = tmp;
			}
		}
	}
	return 0;
}

static int
ocache_entry_free(cache_obj * cobj)
{
	int             i;

	if (cobj == NULL) {
		return (0);
	}

	for (i = 0; i < cobj->iattr.entry_num; i++) {
		if (cobj->iattr.entry_data[i] != NULL) {
			free(cobj->iattr.entry_data[i]);
		}
	}

	if (cobj->iattr.entry_data != NULL) {
		free(cobj->iattr.entry_data);
	}
	for (i = 0; i < cobj->oattr.entry_num; i++) {
		if (cobj->oattr.entry_data[i] != NULL) {
			free(cobj->oattr.entry_data[i]);
		}
	}
	if (cobj->oattr.entry_data != NULL) {
		free(cobj->oattr.entry_data);
	}
	free(cobj);
	return (0);
}

/*
 * Build state to keep track of initial object attributes.
 */
void
cache_set_init_attrs(sig_val_t * id_sig, obj_attr_t * init_attr)
{
	cache_init_obj *cobj;
	unsigned int    index;
	attr_record_t  *arec;
	unsigned char  *buf;
	size_t          len;
	void           *cookie;
	int             err;
	cache_attr_entry *attr_entry;


	pthread_mutex_lock(&shared_mutex);
	index = sig_hash(id_sig) % CACHE_ENTRY_NUM;

	for (cobj = init_table[index]; cobj != NULL; cobj = cobj->next) {
		if (sig_match(&cobj->id_sig, id_sig)) {
			pthread_mutex_unlock(&shared_mutex);
			return;
		}
	}

	cobj = (cache_init_obj *) calloc(1, sizeof(*cobj));
	memcpy(&cobj->id_sig, id_sig, sizeof(sig_val_t));
	cobj->attr.entry_num = 0;
	cobj->attr.entry_data = calloc(1, ATTR_ENTRY_NUM * sizeof(char *));
	assert(cobj->attr.entry_data != NULL);

	err = obj_get_attr_first(init_attr, &buf, &len, &cookie, 0);
	while (err != ENOENT) {
		if (buf == NULL) {
			printf("can not get attr\n");
			break;
		}
		arec = (attr_record_t *) buf;
		attr_entry = (cache_attr_entry *)
		    calloc(1, sizeof(cache_attr_entry));
		assert(attr_entry != NULL);

		attr_entry->name_len = arec->name_len;

		memcpy(attr_entry->the_attr_name, arec->data, arec->name_len);
		memcpy(&attr_entry->attr_sig, &arec->attr_sig,
		       sizeof(sig_val_t));
		cobj->attr.entry_data[cobj->attr.entry_num] = attr_entry;
		cobj->attr.entry_num++;
		if ((cobj->attr.entry_num % ATTR_ENTRY_NUM) == 0) {
			cache_attr_entry **tmp;
			tmp = calloc(1, (cobj->attr.entry_num +
					 ATTR_ENTRY_NUM) * sizeof(char *));
			assert(tmp != NULL);
			memcpy(tmp, cobj->attr.entry_data,
			       cobj->attr.entry_num * sizeof(char *));
			free(cobj->attr.entry_data);
			cobj->attr.entry_data = tmp;
		}
		err = obj_get_attr_next(init_attr, &buf, &len, &cookie, 0);
	}


	cobj->next = init_table[index];
	init_table[index] = cobj;
	pthread_mutex_unlock(&shared_mutex);
	return;
}


int
cache_get_init_attrs(sig_val_t * id_sig, cache_attr_set * init_set)
{
	cache_init_obj *cobj;
	int             found = 0;
	unsigned int    index;

	pthread_mutex_lock(&shared_mutex);
	index = sig_hash(id_sig) % CACHE_ENTRY_NUM;
	for (cobj = init_table[index]; cobj != NULL; cobj = cobj->next) {
		if (sig_match(&cobj->id_sig, id_sig)) {
			found = 1;
			init_set->entry_num = cobj->attr.entry_num;
			/*
			 * XXXLH this can bee too big ??? 
			 */
			memcpy(init_set->entry_data, cobj->attr.entry_data,
			       cobj->attr.entry_num * sizeof(char *));
			break;
		}
	}

	pthread_mutex_unlock(&shared_mutex);
	return found;
}


int
cache_lookup(sig_val_t * id_sig, sig_val_t * fsig, void *fcache_table,
	     cache_attr_set * change_attr, int *err,
	     cache_attr_set ** oattr_set, sig_val_t * iattr_sig, 
	     query_info_t *qinfo)
{
	cache_obj      *cobj;
	int             found = 0;
	unsigned int    index;
	cache_obj     **cache_table = (cache_obj **) fcache_table;

	if (cache_table == NULL)
		return found;
	if (search_done == 1) {
		return (ENOENT);
	}
	pthread_mutex_lock(&shared_mutex);
	index = sig_hash(id_sig) % CACHE_ENTRY_NUM;
	cobj = cache_table[index];
	sig_clear(iattr_sig);

	/*
	 * cache hit if there is a (id_sig, filter sig, input attr sig) match 
	 */
	while (cobj != NULL) {
		if (sig_match(&cobj->id_sig, id_sig)) {
			/*
			 * compare change_attr set with input attr set 
			 */
			if (!compare_attr_set(&cobj->iattr, change_attr)) {
				found = 1;
				*err = cobj->result;
				cobj->ahit_count++;

				memcpy(iattr_sig, &cobj->iattr_sig,
				       sizeof(sig_val_t));
				memcpy(qinfo, &cobj->qid, sizeof(query_info_t));
				
				/*
				 * pass back the output attr set 
				 */
				*oattr_set = &cobj->oattr;
				break;
			}
		}
		cobj = cobj->next;
	}

	pthread_mutex_unlock(&shared_mutex);

	return found;
}

static int
time_after(struct timeval *time1, struct timeval *time2)
{
	if (time1->tv_sec > time2->tv_sec)
		return (1);
	if (time1->tv_sec < time2->tv_sec)
		return (0);
	if (time1->tv_usec > time2->tv_usec)
		return (1);
	return (0);
}

static int
ocache_update(int fd, cache_obj ** cache_table, struct stat *stats)
{
	cache_obj      *cobj;
	int             i;
	off_t           size,
	                rsize;
	cache_obj      *p,
	               *q;
	unsigned int    index;
	int             duplicate;

	if (fd < 0) {
		printf("cache file does not exist\n");
		return (EINVAL);
	}
	size = stats->st_size;
	rsize = 0;

	pthread_mutex_lock(&shared_mutex);
	while (rsize < size) {
		cobj = (cache_obj *) calloc(1, sizeof(*cobj));
		assert(cobj != NULL);
		read(fd, &cobj->id_sig, sizeof(sig_val_t));
		read(fd, &cobj->iattr_sig, sizeof(sig_val_t));
		read(fd, &cobj->result, sizeof(int));

		read(fd, &cobj->eval_count, sizeof(unsigned short));
		cobj->aeval_count = 0;
		read(fd, &cobj->hit_count, sizeof(unsigned short));
		cobj->ahit_count = 0;
		rsize +=
		    (2 * sizeof(sig_val_t) + sizeof(int) +
		     2 * sizeof(unsigned short));
		     
		read(fd, &cobj->qid, sizeof(query_info_t));
		read(fd, &cobj->exec_mode, sizeof(filter_exec_mode_t));
		rsize += sizeof(query_info_t) + sizeof(filter_exec_mode_t);
		
		read(fd, &cobj->iattr.entry_num, sizeof(unsigned int));
		rsize += sizeof(unsigned int);
		cobj->iattr.entry_data =
		    calloc(1, cobj->iattr.entry_num * sizeof(char *));
		assert(cobj->iattr.entry_data != NULL);
		for (i = 0; i < cobj->iattr.entry_num; i++) {
			cobj->iattr.entry_data[i] =
			    calloc(1, sizeof(cache_attr_entry));
			assert(cobj->iattr.entry_data[i] != NULL);
			rsize += read(fd, cobj->iattr.entry_data[i],
				      sizeof(cache_attr_entry));
		}

		read(fd, &cobj->oattr.entry_num, sizeof(unsigned int));
		rsize += sizeof(unsigned int);
		cobj->oattr.entry_data =
		    calloc(1, cobj->oattr.entry_num * sizeof(char *));
		assert(cobj->oattr.entry_data != NULL);
		for (i = 0; i < cobj->oattr.entry_num; i++) {
			cobj->oattr.entry_data[i] =
			    calloc(1, sizeof(cache_attr_entry));
			rsize +=
			    read(fd, cobj->oattr.entry_data[i],
				 sizeof(cache_attr_entry));
		}
		cobj->next = NULL;
		/*
		 * insert it into the cache_table array 
		 */
		index = sig_hash(&cobj->id_sig) % CACHE_ENTRY_NUM;
		if (cache_table[index] == NULL) {
			cache_table[index] = cobj;
			/*
			 * for debug purpose 
			 */
			cache_entry_num++;
		} else {
			p = cache_table[index];
			q = p;
			duplicate = 0;
			while (p != NULL) {
				if (sig_match(&p->id_sig, &cobj->id_sig) &&
				    sig_match(&p->iattr_sig,
					      &cobj->iattr_sig)) {
					if (cobj->eval_count > p->eval_count) {
						p->eval_count +=
						    (cobj->eval_count -
						     p->eval_count);
					}
					if (cobj->hit_count > p->hit_count) {
						p->hit_count +=
						    (cobj->hit_count -
						     p->hit_count);
					}
					duplicate = 1;
					break;
				}
				q = p;
				p = p->next;
			}
			if (duplicate) {
				ocache_entry_free(cobj);
			} else {
				q->next = cobj;
				/*
				 * for debug purpose 
				 */
				cache_entry_num++;
			}
		}
	}
	pthread_mutex_unlock(&shared_mutex);
	return (0);
}

static int
ocache_write_file(char *disk_path, fcache_t * fcache)
{
	char            fpath[PATH_MAX];
	cache_obj      *cobj;
	cache_obj      *tmp;
	int             i,
	                j;
	int             fd;
	int             err;
	cache_obj     **cache_table;
	unsigned int    count;
	struct stat     stats;
	char           *s_str;

	assert(fcache != NULL);
	cache_table = (cache_obj **) fcache->cache_table;
	s_str = sig_string(&fcache->fsig);
	if (s_str == NULL) {
		return (0);
	}
	sprintf(fpath, "%s/%s.%s", disk_path, s_str, CACHE_EXT);
	free(s_str);

	fd = open(fpath, O_RDWR, 00777);
	if (fd >=  0) {
		err = flock(fd, LOCK_EX);
		if (err) {
			perror("failed to lock cache file\n");
			close(fd);
			return (0);
		}
		err = fstat(fd, &stats);
		if (err != 0) {
			perror("failed to stat cache file\n");
			close(fd);
			return (0);
		}
		if (memcmp(&stats.st_mtime, &fcache->mtime, sizeof(time_t))) {
			err = ocache_update(fd, cache_table, &stats);
		}
		close(fd);
	}

	fd = open(fpath, O_CREAT | O_RDWR | O_TRUNC, 00777);
	err = flock(fd, LOCK_EX);
	if (err) {
		perror("failed to lock cache file\n");
		close(fd);
		return (0);
	}

	pthread_mutex_lock(&shared_mutex);
	for (j = 0; j < CACHE_ENTRY_NUM; j++) {
		cobj = cache_table[j];
		while (cobj != NULL) {
			write(fd, &cobj->id_sig, sizeof(sig_val_t));
			write(fd, &cobj->iattr_sig, sizeof(sig_val_t));
			write(fd, &cobj->result, sizeof(int));

			count = cobj->eval_count + cobj->aeval_count;
			write(fd, &count, sizeof(unsigned short));
			count = cobj->hit_count + cobj->ahit_count;
			write(fd, &count, sizeof(unsigned short));
			
			write(fd, &cobj->qid, sizeof(query_info_t));
			write(fd, &cobj->exec_mode, sizeof(filter_exec_mode_t));
			
			write(fd, &cobj->iattr.entry_num,
			      sizeof(unsigned int));
			for (i = 0; i < cobj->iattr.entry_num; i++) {
				write(fd, cobj->iattr.entry_data[i],
				      sizeof(cache_attr_entry));
			}

			write(fd, &cobj->oattr.entry_num,
			      sizeof(unsigned int));

			for (i = 0; i < cobj->oattr.entry_num; i++) {
				write(fd, cobj->oattr.entry_data[i],
				      sizeof(cache_attr_entry));
			}

			tmp = cobj;
			cobj = cobj->next;
			/*
			 * free 
			 */
			ocache_entry_free(tmp);
			cache_entry_num--;
		}
	}
	for (i = 0; i < CACHE_ENTRY_NUM; i++) {
		cache_table[i] = NULL;
	}
	pthread_mutex_unlock(&shared_mutex);
	close(fd);
	return (0);
}

/*
 * free cache table for unused filters: a very simple LRU 
 */
static int
free_fcache_entry(char *disk_path)
{
	int             i;
	fcache_t       *oldest = NULL;
	int             found = -1;

	while (cache_entry_num >= MAX_CACHE_ENTRY_NUM) {
		for (i = 0; i < FCACHE_NUM; i++) {
			if (filter_cache_table[i] == NULL)
				continue;
			if (filter_cache_table[i]->running > 0)
				continue;
			if (oldest == NULL) {
				oldest = filter_cache_table[i];
				found = i;
			} else {
				if (time_after
				    (&oldest->atime,
				     &filter_cache_table[i]->atime)) {
					oldest = filter_cache_table[i];
					found = i;
				}
			}
		}
		if (oldest == NULL) {
			return (-1);
		} else {
			ocache_write_file(disk_path, oldest);
			free(oldest->cache_table);
			free(oldest);
			filter_cache_table[found] = NULL;
			oldest = NULL;
		}
	}
	return (found);
}
<commit_msg>Make sure free_fcache_entry actually frees and entry<commit_after>/*
 *  The OpenDiamond Platform for Interactive Search
 *  Version 3
 *
 *  Copyright (c) 2002-2005 Intel Corporation
 *  Copyright (c) 2006 <NAME> <<EMAIL>>
 *  Copyright (c) 2007 Carnegie Mellon University
 *  All rights reserved.
 *
 *  This software is distributed under the terms of the Eclipse Public
 *  License, Version 1.0 which can be found in the file named LICENSE.
 *  ANY USE, REPRODUCTION OR DISTRIBUTION OF THIS SOFTWARE CONSTITUTES
 *  RECIPIENT'S ACCEPTANCE OF THIS AGREEMENT
 */

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <stddef.h>
#include <sys/stat.h>
#include <sys/file.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdint.h>
#include <dirent.h>
#include <assert.h>
#include <sys/uio.h>
#include <limits.h>

#include "diamond_consts.h"
#include "diamond_types.h"
#include "lib_tools.h"
#include "lib_log.h"
#include "lib_dctl.h"
#include "dctl_common.h"
#include "ocache_priv.h"
#include "obj_attr.h"
#include "obj_attr.h"
#include "lib_filter.h"
#include "lib_filter_sys.h"
#include "lib_odisk.h"
#include "lib_filterexec.h"
#include "lib_ocache.h"
#include "lib_dconfig.h"
#include "filter_priv.h"


#define	MAX_FNAME	128
#define TEMP_ATTR_BUF_SIZE	1024

/*
 * dctl variables 
 */
static unsigned int    if_cache_table = 1;
static unsigned int    if_cache_oattr = 0;

static int      search_active = 0;
static int      search_done = 0;

static GAsyncQueue *ocache_queue;
static GAsyncQueue *oattr_queue;

struct ocache_start_entry {
	void *cache_table;
};

struct ocache_end_entry {
	int result;
	query_info_t qid;		/* search that created entry */
	filter_exec_mode_t exec_mode;	/* mode when entry was created */
};

struct ocache_ring_entry {
	int type;
	sig_val_t id_sig;
	union {
		struct ocache_start_entry start;/* INSERT_START */
		cache_attr_entry attr;		/* INSERT_IATTR/INSERT_OATTR */
		struct ocache_end_entry end;	/* INSERT_END */
	} u;
};

struct cache_attr {
	attr_record_t*	arec;
	obj_data_t *	obj;
};

struct oattr_ring_entry {
	int type;
	sig_val_t id_sig;
	union {
		sig_val_t fsig;		/* INSERT_START filter signature */
		cache_attr_entry iattr;	/* INSERT_IATTR */
		struct cache_attr oattr; /* INSERT_OATTR add output attr*/
	} u;
};

static pthread_mutex_t shared_mutex = PTHREAD_MUTEX_INITIALIZER;
/*
 * active 
 */
static pthread_cond_t bg_active_cv = PTHREAD_COND_INITIALIZER;


#define SIG_BUF_SIZE	256
#define MAX_FILTER_ARG_NAME 256
#define CACHE_ENTRY_NUM 4096
#define FCACHE_NUM 50
#define MAX_CACHE_ENTRY_NUM 0X1000000
#define MAX_ENTRY_NUM	(2 * MAX_CACHE_ENTRY_NUM)
#define MAX_IATTR_SIZE	4096

static cache_init_obj **init_table;
static fcache_t *filter_cache_table[FCACHE_NUM];
static int      cache_entry_num = 0;	/* for debug purpose */

static sig_val_t ocache_sig = { {0,} };
static sig_val_t oattr_sig = { {0,} };

/* wrappers around g_async_queue_push and pop to get type checking */
static void ocache_queue_push(struct ocache_ring_entry *data)
{
	g_async_queue_push(ocache_queue, (gpointer)data);
}

static struct ocache_ring_entry *ocache_queue_pop(void)
{
	return (struct ocache_ring_entry *)g_async_queue_pop(ocache_queue);
}

static void oattr_queue_push(struct oattr_ring_entry *data)
{
	g_async_queue_push(oattr_queue, (gpointer)data);
}

static struct oattr_ring_entry *oattr_queue_pop(void)
{
	return (struct oattr_ring_entry *)g_async_queue_pop(oattr_queue);
}

/*
 * This could be moved to a support library XXX
 */
int
digest_cal(filter_data_t * fdata, char *fn_name, int numarg, char **filt_args,
	   int blob_len, void *blob, sig_val_t * signature)
{
	struct ciovec *iov;
	unsigned int i, len, n = 0;

	len =	fdata->num_libs +	/* library_signatures */
		1 +			/* filter name */
		numarg +		/* filter arguments */
		1;			/* optional binary blob */

	iov = (struct ciovec *)malloc(len * sizeof(struct ciovec));
	assert(iov != NULL);

	/* include the library signatures */
	for (i = 0; i < fdata->num_libs; i++) {
		iov[n].iov_base = &fdata->lib_info[i].lib_sig;
		iov[n].iov_len = sizeof(sig_val_t);
		n++;
	}

	iov[n].iov_base = fn_name;
	iov[n].iov_len = strlen(fn_name);
	n++;

	/* include the args */
	for (i = 0; i < numarg; i++) {
		if (filt_args[i] == NULL)
			continue;

		len = strlen(filt_args[i]);
		if (len >= MAX_FILTER_ARG_NAME)
			return (EINVAL);

		iov[n].iov_base = filt_args[i];
		iov[n].iov_len = len;
		n++;
	}

	/* include binary blob */
	if (blob_len > 0) {
		iov[n].iov_base = blob;
		iov[n].iov_len = blob_len;
		n++;
	}

	sig_cal_vec(iov, n, signature);
	free(iov);
	return 0;
}

/*
 * This is only used in this file, maybe there is a duplicate implementation
 * in another places?
 */
static void
sig_iattr(cache_attr_set * iattr, sig_val_t * sig)
{
	struct ciovec *iov;
	int i;

	iov = (struct ciovec *)malloc(iattr->entry_num * sizeof(struct ciovec));
	assert(iov != NULL);

	for (i = 0; i < iattr->entry_num; i++) {
		iov[i].iov_base = iattr->entry_data[i];
		iov[i].iov_len = sizeof(cache_attr_entry);
	}

	sig_cal_vec(iov, iattr->entry_num, sig);
	free(iov);
}

static int
attr_in_set(cache_attr_entry * inattr, cache_attr_set * set)
{
	int             j;
	cache_attr_entry *tattr;

	for (j = 0; j < set->entry_num; j++) {
		tattr = set->entry_data[j];
		if (tattr == NULL) {
			printf("null temp_j, something wrong\n");
			continue;
		}
		if (memcmp(inattr, tattr, sizeof(*tattr)) == 0) {
			return (1);
		}
	}
	return (0);

}

/*
 * This to see if attr1 is a strict subset of attr2, if not, then we
 * return 1, otherwise return 0.
 */

static int
compare_attr_set(cache_attr_set * attr1, cache_attr_set * attr2)
{
	int             i;
	cache_attr_entry *temp_i;

	/*
	 * for each item in attr1, see if it exists in attr2.
	 */
	for (i = 0; i < attr1->entry_num; i++) {
		temp_i = attr1->entry_data[i];
		if (temp_i == NULL) {
			printf("null temp_i, something wrong\n");
			continue;
		}

		/*
		 * if this item isn't in the set then return 1 
		 */
		if (attr_in_set(temp_i, attr2) == 0) {
			return 1;
		}
	}
	return 0;
}

int
combine_attr_set(cache_attr_set * attr1, cache_attr_set * attr2)
{
	int             i,
	                j;
	int             found;
	cache_attr_entry *temp_i,
	               *temp_j;
	cache_attr_entry **tmp;

	for (i = 0; i < attr2->entry_num; i++) {
		temp_i = attr2->entry_data[i];
		if (temp_i == NULL) {
			printf("null temp_i, something wrong\n");
			assert(0);
		}
		found = 0;
		for (j = 0; j < attr1->entry_num; j++) {
			temp_j = attr1->entry_data[j];
			if (temp_j == NULL) {
				printf("null temp_j, something wrong\n");
				break;
			}
			if (memcmp(temp_i, temp_j, sizeof(*temp_j)) == 0) {
				attr1->entry_data[j] = temp_i;
				found = 1;
				break;
			}
		}
		/*
		 * no found, add to the tail 
		 */
		if (!found) {
			attr1->entry_data[attr1->entry_num] = temp_i;
			attr1->entry_num++;
			if ((attr1->entry_num % ATTR_ENTRY_NUM) == 0) {
				tmp = calloc(1, (attr1->entry_num +
						 ATTR_ENTRY_NUM) *
					     sizeof(char *));
				assert(tmp != NULL);
				memcpy(tmp, attr1->entry_data,
				       attr1->entry_num * sizeof(char *));
				free(attr1->entry_data);
				attr1->entry_data = tmp;
			}
		}
	}
	return 0;
}

static int
ocache_entry_free(cache_obj * cobj)
{
	int             i;

	if (cobj == NULL) {
		return (0);
	}

	for (i = 0; i < cobj->iattr.entry_num; i++) {
		if (cobj->iattr.entry_data[i] != NULL) {
			free(cobj->iattr.entry_data[i]);
		}
	}

	if (cobj->iattr.entry_data != NULL) {
		free(cobj->iattr.entry_data);
	}
	for (i = 0; i < cobj->oattr.entry_num; i++) {
		if (cobj->oattr.entry_data[i] != NULL) {
			free(cobj->oattr.entry_data[i]);
		}
	}
	if (cobj->oattr.entry_data != NULL) {
		free(cobj->oattr.entry_data);
	}
	free(cobj);
	return (0);
}

/*
 * Build state to keep track of initial object attributes.
 */
void
cache_set_init_attrs(sig_val_t * id_sig, obj_attr_t * init_attr)
{
	cache_init_obj *cobj;
	unsigned int    index;
	attr_record_t  *arec;
	unsigned char  *buf;
	size_t          len;
	void           *cookie;
	int             err;
	cache_attr_entry *attr_entry;


	pthread_mutex_lock(&shared_mutex);
	index = sig_hash(id_sig) % CACHE_ENTRY_NUM;

	for (cobj = init_table[index]; cobj != NULL; cobj = cobj->next) {
		if (sig_match(&cobj->id_sig, id_sig)) {
			pthread_mutex_unlock(&shared_mutex);
			return;
		}
	}

	cobj = (cache_init_obj *) calloc(1, sizeof(*cobj));
	memcpy(&cobj->id_sig, id_sig, sizeof(sig_val_t));
	cobj->attr.entry_num = 0;
	cobj->attr.entry_data = calloc(1, ATTR_ENTRY_NUM * sizeof(char *));
	assert(cobj->attr.entry_data != NULL);

	err = obj_get_attr_first(init_attr, &buf, &len, &cookie, 0);
	while (err != ENOENT) {
		if (buf == NULL) {
			printf("can not get attr\n");
			break;
		}
		arec = (attr_record_t *) buf;
		attr_entry = (cache_attr_entry *)
		    calloc(1, sizeof(cache_attr_entry));
		assert(attr_entry != NULL);

		attr_entry->name_len = arec->name_len;

		memcpy(attr_entry->the_attr_name, arec->data, arec->name_len);
		memcpy(&attr_entry->attr_sig, &arec->attr_sig,
		       sizeof(sig_val_t));
		cobj->attr.entry_data[cobj->attr.entry_num] = attr_entry;
		cobj->attr.entry_num++;
		if ((cobj->attr.entry_num % ATTR_ENTRY_NUM) == 0) {
			cache_attr_entry **tmp;
			tmp = calloc(1, (cobj->attr.entry_num +
					 ATTR_ENTRY_NUM) * sizeof(char *));
			assert(tmp != NULL);
			memcpy(tmp, cobj->attr.entry_data,
			       cobj->attr.entry_num * sizeof(char *));
			free(cobj->attr.entry_data);
			cobj->attr.entry_data = tmp;
		}
		err = obj_get_attr_next(init_attr, &buf, &len, &cookie, 0);
	}


	cobj->next = init_table[index];
	init_table[index] = cobj;
	pthread_mutex_unlock(&shared_mutex);
	return;
}


int
cache_get_init_attrs(sig_val_t * id_sig, cache_attr_set * init_set)
{
	cache_init_obj *cobj;
	int             found = 0;
	unsigned int    index;

	pthread_mutex_lock(&shared_mutex);
	index = sig_hash(id_sig) % CACHE_ENTRY_NUM;
	for (cobj = init_table[index]; cobj != NULL; cobj = cobj->next) {
		if (sig_match(&cobj->id_sig, id_sig)) {
			found = 1;
			init_set->entry_num = cobj->attr.entry_num;
			/*
			 * XXXLH this can bee too big ??? 
			 */
			memcpy(init_set->entry_data, cobj->attr.entry_data,
			       cobj->attr.entry_num * sizeof(char *));
			break;
		}
	}

	pthread_mutex_unlock(&shared_mutex);
	return found;
}


int
cache_lookup(sig_val_t * id_sig, sig_val_t * fsig, void *fcache_table,
	     cache_attr_set * change_attr, int *err,
	     cache_attr_set ** oattr_set, sig_val_t * iattr_sig, 
	     query_info_t *qinfo)
{
	cache_obj      *cobj;
	int             found = 0;
	unsigned int    index;
	cache_obj     **cache_table = (cache_obj **) fcache_table;

	if (cache_table == NULL)
		return found;
	if (search_done == 1) {
		return (ENOENT);
	}
	pthread_mutex_lock(&shared_mutex);
	index = sig_hash(id_sig) % CACHE_ENTRY_NUM;
	cobj = cache_table[index];
	sig_clear(iattr_sig);

	/*
	 * cache hit if there is a (id_sig, filter sig, input attr sig) match 
	 */
	while (cobj != NULL) {
		if (sig_match(&cobj->id_sig, id_sig)) {
			/*
			 * compare change_attr set with input attr set 
			 */
			if (!compare_attr_set(&cobj->iattr, change_attr)) {
				found = 1;
				*err = cobj->result;
				cobj->ahit_count++;

				memcpy(iattr_sig, &cobj->iattr_sig,
				       sizeof(sig_val_t));
				memcpy(qinfo, &cobj->qid, sizeof(query_info_t));
				
				/*
				 * pass back the output attr set 
				 */
				*oattr_set = &cobj->oattr;
				break;
			}
		}
		cobj = cobj->next;
	}

	pthread_mutex_unlock(&shared_mutex);

	return found;
}

static int
time_after(struct timeval *time1, struct timeval *time2)
{
	if (time1->tv_sec > time2->tv_sec)
		return (1);
	if (time1->tv_sec < time2->tv_sec)
		return (0);
	if (time1->tv_usec > time2->tv_usec)
		return (1);
	return (0);
}

static int
ocache_update(int fd, cache_obj ** cache_table, struct stat *stats)
{
	cache_obj      *cobj;
	int             i;
	off_t           size,
	                rsize;
	cache_obj      *p,
	               *q;
	unsigned int    index;
	int             duplicate;

	if (fd < 0) {
		printf("cache file does not exist\n");
		return (EINVAL);
	}
	size = stats->st_size;
	rsize = 0;

	pthread_mutex_lock(&shared_mutex);
	while (rsize < size) {
		cobj = (cache_obj *) calloc(1, sizeof(*cobj));
		assert(cobj != NULL);
		read(fd, &cobj->id_sig, sizeof(sig_val_t));
		read(fd, &cobj->iattr_sig, sizeof(sig_val_t));
		read(fd, &cobj->result, sizeof(int));

		read(fd, &cobj->eval_count, sizeof(unsigned short));
		cobj->aeval_count = 0;
		read(fd, &cobj->hit_count, sizeof(unsigned short));
		cobj->ahit_count = 0;
		rsize +=
		    (2 * sizeof(sig_val_t) + sizeof(int) +
		     2 * sizeof(unsigned short));
		     
		read(fd, &cobj->qid, sizeof(query_info_t));
		read(fd, &cobj->exec_mode, sizeof(filter_exec_mode_t));
		rsize += sizeof(query_info_t) + sizeof(filter_exec_mode_t);
		
		read(fd, &cobj->iattr.entry_num, sizeof(unsigned int));
		rsize += sizeof(unsigned int);
		cobj->iattr.entry_data =
		    calloc(1, cobj->iattr.entry_num * sizeof(char *));
		assert(cobj->iattr.entry_data != NULL);
		for (i = 0; i < cobj->iattr.entry_num; i++) {
			cobj->iattr.entry_data[i] =
			    calloc(1, sizeof(cache_attr_entry));
			assert(cobj->iattr.entry_data[i] != NULL);
			rsize += read(fd, cobj->iattr.entry_data[i],
				      sizeof(cache_attr_entry));
		}

		read(fd, &cobj->oattr.entry_num, sizeof(unsigned int));
		rsize += sizeof(unsigned int);
		cobj->oattr.entry_data =
		    calloc(1, cobj->oattr.entry_num * sizeof(char *));
		assert(cobj->oattr.entry_data != NULL);
		for (i = 0; i < cobj->oattr.entry_num; i++) {
			cobj->oattr.entry_data[i] =
			    calloc(1, sizeof(cache_attr_entry));
			rsize +=
			    read(fd, cobj->oattr.entry_data[i],
				 sizeof(cache_attr_entry));
		}
		cobj->next = NULL;
		/*
		 * insert it into the cache_table array 
		 */
		index = sig_hash(&cobj->id_sig) % CACHE_ENTRY_NUM;
		if (cache_table[index] == NULL) {
			cache_table[index] = cobj;
			/*
			 * for debug purpose 
			 */
			cache_entry_num++;
		} else {
			p = cache_table[index];
			q = p;
			duplicate = 0;
			while (p != NULL) {
				if (sig_match(&p->id_sig, &cobj->id_sig) &&
				    sig_match(&p->iattr_sig,
					      &cobj->iattr_sig)) {
					if (cobj->eval_count > p->eval_count) {
						p->eval_count +=
						    (cobj->eval_count -
						     p->eval_count);
					}
					if (cobj->hit_count > p->hit_count) {
						p->hit_count +=
						    (cobj->hit_count -
						     p->hit_count);
					}
					duplicate = 1;
					break;
				}
				q = p;
				p = p->next;
			}
			if (duplicate) {
				ocache_entry_free(cobj);
			} else {
				q->next = cobj;
				/*
				 * for debug purpose 
				 */
				cache_entry_num++;
			}
		}
	}
	pthread_mutex_unlock(&shared_mutex);
	return (0);
}

static int
ocache_write_file(char *disk_path, fcache_t * fcache)
{
	char            fpath[PATH_MAX];
	cache_obj      *cobj;
	cache_obj      *tmp;
	int             i,
	                j;
	int             fd;
	int             err;
	cache_obj     **cache_table;
	unsigned int    count;
	struct stat     stats;
	char           *s_str;

	assert(fcache != NULL);
	cache_table = (cache_obj **) fcache->cache_table;
	s_str = sig_string(&fcache->fsig);
	if (s_str == NULL) {
		return (0);
	}
	sprintf(fpath, "%s/%s.%s", disk_path, s_str, CACHE_EXT);
	free(s_str);

	fd = open(fpath, O_RDWR, 00777);
	if (fd >=  0) {
		err = flock(fd, LOCK_EX);
		if (err) {
			perror("failed to lock cache file\n");
			close(fd);
			return (0);
		}
		err = fstat(fd, &stats);
		if (err != 0) {
			perror("failed to stat cache file\n");
			close(fd);
			return (0);
		}
		if (memcmp(&stats.st_mtime, &fcache->mtime, sizeof(time_t))) {
			err = ocache_update(fd, cache_table, &stats);
		}
		close(fd);
	}

	fd = open(fpath, O_CREAT | O_RDWR | O_TRUNC, 00777);
	err = flock(fd, LOCK_EX);
	if (err) {
		perror("failed to lock cache file\n");
		close(fd);
		return (0);
	}

	pthread_mutex_lock(&shared_mutex);
	for (j = 0; j < CACHE_ENTRY_NUM; j++) {
		cobj = cache_table[j];
		while (cobj != NULL) {
			write(fd, &cobj->id_sig, sizeof(sig_val_t));
			write(fd, &cobj->iattr_sig, sizeof(sig_val_t));
			write(fd, &cobj->result, sizeof(int));

			count = cobj->eval_count + cobj->aeval_count;
			write(fd, &count, sizeof(unsigned short));
			count = cobj->hit_count + cobj->ahit_count;
			write(fd, &count, sizeof(unsigned short));
			
			write(fd, &cobj->qid, sizeof(query_info_t));
			write(fd, &cobj->exec_mode, sizeof(filter_exec_mode_t));
			
			write(fd, &cobj->iattr.entry_num,
			      sizeof(unsigned int));
			for (i = 0; i < cobj->iattr.entry_num; i++) {
				write(fd, cobj->iattr.entry_data[i],
				      sizeof(cache_attr_entry));
			}

			write(fd, &cobj->oattr.entry_num,
			      sizeof(unsigned int));

			for (i = 0; i < cobj->oattr.entry_num; i++) {
				write(fd, cobj->oattr.entry_data[i],
				      sizeof(cache_attr_entry));
			}

			tmp = cobj;
			cobj = cobj->next;
			/*
			 * free 
			 */
			ocache_entry_free(tmp);
			cache_entry_num--;
		}
	}
	for (i = 0; i < CACHE_ENTRY_NUM; i++) {
		cache_table[i] = NULL;
	}
	pthread_mutex_unlock(&shared_mutex);
	close(fd);
	return (0);
}

/*
 * free cache table for unused filters: a very simple LRU 
 */
static int
free_fcache_entry(char *disk_path)
{
	int             i;
	fcache_t       *oldest = NULL;
	int             found = -1;

	do {
		for (i = 0; i < FCACHE_NUM; i++) {
			if (filter_cache_table[i] == NULL)
				continue;
			if (filter_cache_table[i]->running > 0)
				continue;
			if (oldest == NULL) {
				oldest = filter_cache_table[i];
				found = i;
			} else {
				if (time_after
				    (&oldest->atime,
				     &filter_cache_table[i]->atime)) {
					oldest = filter_cache_table[i];
					found = i;
				}
			}
		}
		if (oldest == NULL) {
			return (-1);
		} else {
			ocache_write_file(disk_path, oldest);
			free(oldest->cache_table);
			free(oldest);
			filter_cache_table[found] = NULL;
			oldest = NULL;
		}
	}
	while (cache_entry_num >= MAX_CACHE_ENTRY_NUM);

	return (found);
}

{'fraction_non_alphanumeric': 0.14683523154465206, 'fraction_numerical': 0.010080476734148311, 'mean_word_length': 3.2499408284023668, 'pattern_counts': {'":': 0, '<': 77, '<?xml version=': 0, '>': 347, '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': '7125485', 'n_tokens_mistral': 15877, 'n_tokens_neox': 14237, 'n_words': 3007}