<commit_before>/* $Id$ */

#include <time.h>

#include "psc_ds/dynarray.h"
#include "psc_ds/listcache.h"
#include "psc_ds/tree.h"
#include "psc_rpc/rpc.h"
#include "psc_rpc/rsx.h"
#include "psc_util/alloc.h"
#include "psc_util/atomic.h"
#include "psc_util/lock.h"
#include "psc_util/log.h"

#include "bmap_iod.h"
#include "fid.h"
#include "fidcache.h"
#include "sliod.h"
#include "slvr.h"

struct biod_infslvr_tree	 binfSlvrs;
struct pscrpc_nbreqset		*slvrNbReqSet;

__static SPLAY_GENERATE(crcup_reftree, biod_crcup_ref, bcr_tentry, bcr_cmp);

/*
 * Send an RPC containing CRC updates for slivers to the metadata server.  To avoid 
 * potential woes caused by out-of-order deliveries, we should allow at most one inflight
 * CRC update RPC at any time.  Note that this does not prevent us from having multiple 
 * threads to do the CRC calculation.
 */
__static int
slvr_worker_crcup_genrq(const struct dynarray *ref_array)
{
	struct biod_crcup_ref *bcrc_ref;
	struct srm_bmap_crcwrt_req *mq;
	struct srm_generic_rep *mp;
	struct pscrpc_request *req;
	struct pscrpc_bulk_desc *desc;
	struct iovec *iovs;
	size_t len;
	int rc;
	uint32_t i;

	rc = RSX_NEWREQ(rmi_csvc->csvc_import, SRMI_VERSION,
			SRMT_BMAPCRCWRT, req, mq, mp);
	if (rc) {
		return rc;
	}

	PSC_CRC64_INIT(&mq->crc);

	mq->ncrc_updates = dynarray_len(ref_array);
	req->rq_async_args.pointer_arg[0] = (void *)ref_array;

	len = (mq->ncrc_updates * sizeof(struct srm_bmap_crcup));
	iovs = PSCALLOC(sizeof(*iovs) * mq->ncrc_updates);

	for (i = 0; i < mq->ncrc_updates; i++) {
		bcrc_ref = dynarray_getpos(ref_array, i);

		DEBUG_BCR(PLL_NOTIFY, bcrc_ref, "ref_array pos=%d", i);

		iod_inode_getsize(&bcrc_ref->bcr_crcup.fg,
				  (off_t *)&bcrc_ref->bcr_crcup.fsize);

		mq->ncrcs_per_update[i] = bcrc_ref->bcr_crcup.nups;

		iovs[i].iov_base = &bcrc_ref->bcr_crcup;
		len += iovs[i].iov_len = ((mq->ncrcs_per_update[i] *
					   sizeof(struct srm_bmap_crcwire)) +
					   sizeof(struct srm_bmap_crcup));

		psc_crc64_add(&mq->crc, iovs[i].iov_base, iovs[i].iov_len);
	}
	psc_assert(len <= LNET_MTU);

	PSC_CRC64_FIN(&mq->crc);

	rc = rsx_bulkclient(req, &desc, BULK_GET_SOURCE, SRMI_BULK_PORTAL,
			     iovs, mq->ncrc_updates);
	PSCFREE(iovs);

	nbreqset_add(slvrNbReqSet, req);
	nbrequest_reap(slvrNbReqSet);

	return (rc);
}

__static void
slvr_worker_push_crcups(void)
{
	int			 i;
	int			 rc;
	struct timespec		 now;
	struct biod_crcup_ref	*bcrc_ref;
	struct dynarray		*ref_array;

	spinlock(&binfSlvrs.binfst_lock);
	if (binfSlvrs.binfst_inflight) {
		freelock(&binfSlvrs.binfst_lock);
		return;
	}
	binfSlvrs.binfst_inflight = 1;

	ref_array = PSCALLOC(sizeof(struct dynarray));

	/* First, try to gather full crcup's */
	SPLAY_FOREACH(bcrc_ref, crcup_reftree, &binfSlvrs.binfst_tree) {
		
		DEBUG_BCR(PLL_NOTIFY, bcrc_ref, "ref_array sz=%d", 
			  dynarray_len(ref_array));

		if (bcrc_ref->bcr_crcup.nups == MAX_BMAP_INODE_PAIRS) {
			bcrc_ref->bcr_flags |= BCR_SCANNED | BCR_UPDATED;
			dynarray_add(ref_array, bcrc_ref);
		}
		if (dynarray_len(ref_array) >= MAX_BMAP_NCRC_UPDATES)
			break;
	}
	/* Second, try to gather old crcups */
	if (dynarray_len(ref_array) >= MAX_BMAP_NCRC_UPDATES) {
		goto done;
	}
	clock_gettime(CLOCK_REALTIME, &now);
	SPLAY_FOREACH(bcrc_ref, crcup_reftree, &binfSlvrs.binfst_tree) {
		
		DEBUG_BCR(PLL_NOTIFY, bcrc_ref, "ref_array sz=%d now=%lu", 
			  dynarray_len(ref_array), now.tv_sec);

		if (bcrc_ref->bcr_flags & BCR_SCANNED)
			continue;

		if (now.tv_sec < (bcrc_ref->bcr_age.tv_sec + BIOD_CRCUP_MAX_AGE)) {
			bcrc_ref->bcr_flags |= BCR_SCANNED | BCR_UPDATED;
			dynarray_add(ref_array, bcrc_ref);
		}
		if (dynarray_len(ref_array) >= MAX_BMAP_NCRC_UPDATES)
			break;
	}
	if (!dynarray_len(ref_array)) {
		PSCFREE(ref_array);
		freelock(&binfSlvrs.binfst_lock);
		return;
	}
done:
	freelock(&binfSlvrs.binfst_lock);

	rc = slvr_worker_crcup_genrq(ref_array);

	/*
	 * If we fail to send an RPC, we must leave the reference in the tree for future attempt.  
	 * Otherwise, the callback function (i.e., slvr_nbreqset_cb()) should remove them from the tree.
	 */
	if (rc) {
		spinlock(&binfSlvrs.binfst_lock);
		for (i = 0; i < dynarray_len(ref_array); i++) {
			bcrc_ref = dynarray_getpos(ref_array, i);
			bcrc_ref->bcr_flags &= ~(BCR_SCANNED | BCR_UPDATED);
		}
		spinlock(&binfSlvrs.binfst_lock);
	}
}


int
slvr_nbreqset_cb(__unusedx struct pscrpc_request *req,
		 struct pscrpc_async_args *args)
{
	struct dynarray *a = args->pointer_arg[0];
	struct biod_crcup_ref *bcrc_ref=NULL;
	struct slvr_ref *s;
	struct srm_generic_rep *mp;
	int i, err=0;
	uint32_t j;

	ENTRY;

	mp = psc_msg_buf(req->rq_repmsg, 0, sizeof(*mp));
	if (req->rq_status || mp->rc)
		err = 1;

	spinlock(&binfSlvrs.binfst_lock);

	psc_assert(a);

	for (i=0; i < dynarray_len(a); i++) {
		bcrc_ref = dynarray_getpos(a, i);

		for (j=0; j < bcrc_ref->bcr_nups; j++) {
			s = bcrc_ref->bcr_slvrs[j];

			SLVR_LOCK(s);

			DEBUG_SLVR(err ? PLL_ERROR : PLL_INFO, s,
				   "crcup %s fid(%"PRId64":%"PRId64")"<commit_msg>style pick<commit_after>/* $Id$ */

#include <time.h>

#include "psc_ds/dynarray.h"
#include "psc_ds/listcache.h"
#include "psc_ds/tree.h"
#include "psc_rpc/rpc.h"
#include "psc_rpc/rsx.h"
#include "psc_util/alloc.h"
#include "psc_util/atomic.h"
#include "psc_util/lock.h"
#include "psc_util/log.h"

#include "bmap_iod.h"
#include "fid.h"
#include "fidcache.h"
#include "sliod.h"
#include "slvr.h"

struct biod_infslvr_tree	 binfSlvrs;
struct pscrpc_nbreqset		*slvrNbReqSet;

__static SPLAY_GENERATE(crcup_reftree, biod_crcup_ref, bcr_tentry, bcr_cmp);

/*
 * Send an RPC containing CRC updates for slivers to the metadata server.  To avoid 
 * potential woes caused by out-of-order deliveries, we should allow at most one inflight
 * CRC update RPC at any time.  Note that this does not prevent us from having multiple 
 * threads to do the CRC calculation.
 */
__static int
slvr_worker_crcup_genrq(const struct dynarray *ref_array)
{
	struct biod_crcup_ref *bcrc_ref;
	struct srm_bmap_crcwrt_req *mq;
	struct srm_generic_rep *mp;
	struct pscrpc_request *req;
	struct pscrpc_bulk_desc *desc;
	struct iovec *iovs;
	size_t len;
	int rc;
	uint32_t i;

	rc = RSX_NEWREQ(rmi_csvc->csvc_import, SRMI_VERSION,
			SRMT_BMAPCRCWRT, req, mq, mp);
	if (rc) {
		return rc;
	}

	PSC_CRC64_INIT(&mq->crc);

	mq->ncrc_updates = dynarray_len(ref_array);
	req->rq_async_args.pointer_arg[0] = (void *)ref_array;

	len = (mq->ncrc_updates * sizeof(struct srm_bmap_crcup));
	iovs = PSCALLOC(sizeof(*iovs) * mq->ncrc_updates);

	for (i = 0; i < mq->ncrc_updates; i++) {
		bcrc_ref = dynarray_getpos(ref_array, i);

		DEBUG_BCR(PLL_NOTIFY, bcrc_ref, "ref_array pos=%d", i);

		iod_inode_getsize(&bcrc_ref->bcr_crcup.fg,
				  (off_t *)&bcrc_ref->bcr_crcup.fsize);

		mq->ncrcs_per_update[i] = bcrc_ref->bcr_crcup.nups;

		iovs[i].iov_base = &bcrc_ref->bcr_crcup;
		len += iovs[i].iov_len = ((mq->ncrcs_per_update[i] *
					   sizeof(struct srm_bmap_crcwire)) +
					   sizeof(struct srm_bmap_crcup));

		psc_crc64_add(&mq->crc, iovs[i].iov_base, iovs[i].iov_len);
	}
	psc_assert(len <= LNET_MTU);

	PSC_CRC64_FIN(&mq->crc);

	rc = rsx_bulkclient(req, &desc, BULK_GET_SOURCE, SRMI_BULK_PORTAL,
			     iovs, mq->ncrc_updates);
	PSCFREE(iovs);

	nbreqset_add(slvrNbReqSet, req);
	nbrequest_reap(slvrNbReqSet);

	return (rc);
}

__static void
slvr_worker_push_crcups(void)
{
	int			 i;
	int			 rc;
	struct timespec		 now;
	struct biod_crcup_ref	*bcrc_ref;
	struct dynarray		*ref_array;

	spinlock(&binfSlvrs.binfst_lock);
	if (binfSlvrs.binfst_inflight) {
		freelock(&binfSlvrs.binfst_lock);
		return;
	}
	binfSlvrs.binfst_inflight = 1;

	ref_array = PSCALLOC(sizeof(struct dynarray));

	/* First, try to gather full crcup's */
	SPLAY_FOREACH(bcrc_ref, crcup_reftree, &binfSlvrs.binfst_tree) {
		
		DEBUG_BCR(PLL_NOTIFY, bcrc_ref, "ref_array sz=%d", 
			  dynarray_len(ref_array));

		if (bcrc_ref->bcr_crcup.nups == MAX_BMAP_INODE_PAIRS) {
			bcrc_ref->bcr_flags |= BCR_SCANNED | BCR_UPDATED;
			dynarray_add(ref_array, bcrc_ref);
		}
		if (dynarray_len(ref_array) >= MAX_BMAP_NCRC_UPDATES)
			break;
	}
	/* Second, try to gather old crcups */
	if (dynarray_len(ref_array) >= MAX_BMAP_NCRC_UPDATES) {
		goto done;
	}
	clock_gettime(CLOCK_REALTIME, &now);
	SPLAY_FOREACH(bcrc_ref, crcup_reftree, &binfSlvrs.binfst_tree) {
		
		DEBUG_BCR(PLL_NOTIFY, bcrc_ref, "ref_array sz=%d now=%lu", 
			  dynarray_len(ref_array), now.tv_sec);

		if (bcrc_ref->bcr_flags & BCR_SCANNED)
			continue;

		if (now.tv_sec < (bcrc_ref->bcr_age.tv_sec + BIOD_CRCUP_MAX_AGE)) {
			bcrc_ref->bcr_flags |= BCR_SCANNED | BCR_UPDATED;
			dynarray_add(ref_array, bcrc_ref);
		}
		if (dynarray_len(ref_array) >= MAX_BMAP_NCRC_UPDATES)
			break;
	}
	if (!dynarray_len(ref_array)) {
		PSCFREE(ref_array);
		freelock(&binfSlvrs.binfst_lock);
		return;
	}
done:
	freelock(&binfSlvrs.binfst_lock);

	rc = slvr_worker_crcup_genrq(ref_array);

	/*
	 * If we fail to send an RPC, we must leave the reference in the tree for future attempt.  
	 * Otherwise, the callback function (i.e., slvr_nbreqset_cb()) should remove them from the tree.
	 */
	if (rc) {
		spinlock(&binfSlvrs.binfst_lock);
		for (i = 0; i < dynarray_len(ref_array); i++) {
			bcrc_ref = dynarray_getpos(ref_array, i);
			bcrc_ref->bcr_flags &= ~(BCR_SCANNED | BCR_UPDATED);
		}
		spinlock(&binfSlvrs.binfst_lock);
	}
}


int
slvr_nbreqset_cb(__unusedx struct pscrpc_request *req,
		 struct pscrpc_async_args *args)
{
	struct dynarray		*a;
	struct slvr_ref		*s;
	int			 i;
	int			 j;
	struct srm_generic_rep	*mp;
	int			 err;
	struct biod_crcup_ref	*bcrc_ref;

	ENTRY;

	err = 0;
	a = args->pointer_arg[0];

	mp = psc_msg_buf(req->rq_repmsg, 0, sizeof(*mp));
	if (req->rq_status || mp->rc)
		err = 1;

	spinlock(&binfSlvrs.binfst_lock);

	psc_assert(a);

	for (i=0; i < dynarray_len(a); i++) {
		bcrc_ref = dynarray_getpos(a, i);

		for (j=0; j < bcrc_ref->bcr_nups; j++) {
			s = bcrc_ref->bcr_slvrs[j];

			SLVR_LOCK(s);

			DEBUG_SLVR(err ? PLL_ERROR : PLL_INFO, s,
				   "crcup %s fid(%"PRId64":%"PRId64")"
{'fraction_non_alphanumeric': 0.13260826182398722, 'fraction_numerical': 0.004589902215126721, 'mean_word_length': 4.669117647058823, 'pattern_counts': {'":': 2, '<': 17, '<?xml version=': 0, '>': 67, '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': '468281', 'n_tokens_mistral': 4868, 'n_tokens_neox': 4542, 'n_words': 806}