<commit_before>/* $Id$ */

#define PSC_SUBSYS PSS_RPC

#include <inttypes.h>

#include "psc_rpc/rpc.h"
#include "psc_rpc/rpclog.h"
#include "psc_util/alloc.h"
#include "psc_util/atomic.h"
#include "psc_util/log.h"
#include "psc_util/thread.h"
#include "psc_util/waitq.h"

struct pscrpc_nbreapthr {
	struct pscrpc_nbreqset	*pnbt_nbset;
};

/**
 * pscrpc_nbreqset_push - send out new requests
 *
 */
static int
pscrpc_nbreqset_push(struct pscrpc_request *req)
{
	return (pscrpc_push_req(req));
}

/**
 * pscrpc_nbreqset_init - make a non-blocking set
 * @nb_interpret:  this must only take into account completed rpcs
 * @nb_callback:   application callback
 */
struct pscrpc_nbreqset *
pscrpc_nbreqset_init(pscrpc_set_interpreterf nb_interpret,
    pscrpc_nbreq_callback nb_callback)
{
	struct pscrpc_nbreqset *nbs;

	nbs = PSCALLOC(sizeof(*nbs));
	psc_waitq_init(&nbs->nb_waitq);
	pscrpc_set_init(&nbs->nb_reqset);
	nbs->nb_reqset.set_interpret = nb_interpret;
	nbs->nb_callback = nb_callback;
	atomic_set(&nbs->nb_outstanding, 0);
	LOCK_INIT(&nbs->nb_lock);
	return nbs;
}

/**
 * pscrpc_nbreqset_add - add a new non-blocking request to the mix
 *
 */
void
pscrpc_nbreqset_add(struct pscrpc_nbreqset *nbs, struct pscrpc_request *req)
{
	req->rq_waitq = &nbs->nb_waitq;
	atomic_inc(&nbs->nb_outstanding);
	pscrpc_set_add_new_req(&nbs->nb_reqset, req);
	if (pscrpc_nbreqset_push(req))
		DEBUG_REQ(PLL_FATAL, req, "Send Failure");
}

/**
 * pscrpc_nbrequest_flush - sync all outstanding requests
 */
int
pscrpc_nbreqset_flush(struct pscrpc_nbreqset *nbs)
{
	return (pscrpc_set_wait(&nbs->nb_reqset));
}

/**
 * pscrpc_nbrequest_reap - remove completed requests from the
 *                  request set and place them into the
 *                  'completed' list.
 * @nbs: the non-blocking set
 * Notes:  Call before pscrpc_check_set(), pscrpc_check_set() manages
 *        'set_remaining'..  If there are problems look at pscrpc_set_wait(),
 *        there you'll find a more sophisticated and proper use of sets.
 *        pscrpc_set_wait() deals with blocking set nevertheless much applies
 *        here as well.
 */
int
pscrpc_nbreqset_reap(struct pscrpc_nbreqset *nbs)
{
	int nreaped = 0, nchecked = 0, timeout = 1;
	struct pscrpc_request_set *set = &nbs->nb_reqset;
	struct pscrpc_request *req, *next;
	struct l_wait_info lwi;

	spinlock(&nbs->nb_lock);
	if (nbs->nb_flags & NBREQSET_WORK_INPROG) {
		freelock(&nbs->nb_lock);
		RETURN(0);
	} else {
		nbs->nb_flags |= NBREQSET_WORK_INPROG;
		freelock(&nbs->nb_lock);
	}

	lwi = LWI_TIMEOUT(timeout, NULL, NULL);
	pscrpc_cli_wait_event(&set->set_waitq,
		       (nreaped=pscrpc_check_set(set, 0)), &lwi);

	if (!nreaped) {
		spinlock(&nbs->nb_lock);
		nbs->nb_flags &= ~NBREQSET_WORK_INPROG;
		freelock(&nbs->nb_lock);
		return (0);
	}

	pscrpc_set_lock(set);

	psclist_for_each_entry_safe(req, next,
	    &set->set_requests, rq_set_chain_lentry) {
		nchecked++;
		DEBUG_REQ(PLL_INFO, req, "reap if Completed");
		/*
		 * Move sent rpcs to the set_requests list
		 */
		if (req->rq_phase == PSCRPC_RQ_PHASE_COMPLETE) {
			psclist_del(&req->rq_set_chain_lentry);
			atomic_dec(&nbs->nb_outstanding);
			nreaped++;
			/*
			 * paranoia
			 */
			psc_assert(atomic_read(&nbs->nb_outstanding) >= 0);
			/*
			 * This is the caller's last shot at accessing
			 *  this msg..
			 * Let the callback deal with it's own
			 *  error handling, we can't do much from here
			 */
			if (nbs->nb_callback != NULL)
				nbs->nb_callback(req, &req->rq_async_args);
			/*
			 * Be done with it..
			 */
			pscrpc_req_finished(req);
		}
	}
	freelock(&set->set_lock);<commit_msg>if the pscrpc_request async callback handler returns nonzero, do not run req_finished().  if you know of a worse hack, let me know<commit_after>/* $Id$ */

#define PSC_SUBSYS PSS_RPC

#include <inttypes.h>

#include "psc_rpc/rpc.h"
#include "psc_rpc/rpclog.h"
#include "psc_util/alloc.h"
#include "psc_util/atomic.h"
#include "psc_util/log.h"
#include "psc_util/thread.h"
#include "psc_util/waitq.h"

struct pscrpc_nbreapthr {
	struct pscrpc_nbreqset	*pnbt_nbset;
};

/**
 * pscrpc_nbreqset_push - send out new requests
 *
 */
static int
pscrpc_nbreqset_push(struct pscrpc_request *req)
{
	return (pscrpc_push_req(req));
}

/**
 * pscrpc_nbreqset_init - make a non-blocking set
 * @nb_interpret:  this must only take into account completed rpcs
 * @nb_callback:   application callback
 */
struct pscrpc_nbreqset *
pscrpc_nbreqset_init(pscrpc_set_interpreterf nb_interpret,
    pscrpc_nbreq_callback nb_callback)
{
	struct pscrpc_nbreqset *nbs;

	nbs = PSCALLOC(sizeof(*nbs));
	psc_waitq_init(&nbs->nb_waitq);
	pscrpc_set_init(&nbs->nb_reqset);
	nbs->nb_reqset.set_interpret = nb_interpret;
	nbs->nb_callback = nb_callback;
	atomic_set(&nbs->nb_outstanding, 0);
	LOCK_INIT(&nbs->nb_lock);
	return nbs;
}

/**
 * pscrpc_nbreqset_add - add a new non-blocking request to the mix
 *
 */
void
pscrpc_nbreqset_add(struct pscrpc_nbreqset *nbs, struct pscrpc_request *req)
{
	req->rq_waitq = &nbs->nb_waitq;
	atomic_inc(&nbs->nb_outstanding);
	pscrpc_set_add_new_req(&nbs->nb_reqset, req);
	if (pscrpc_nbreqset_push(req))
		DEBUG_REQ(PLL_FATAL, req, "Send Failure");
}

/**
 * pscrpc_nbrequest_flush - sync all outstanding requests
 */
int
pscrpc_nbreqset_flush(struct pscrpc_nbreqset *nbs)
{
	return (pscrpc_set_wait(&nbs->nb_reqset));
}

/**
 * pscrpc_nbrequest_reap - remove completed requests from the
 *                  request set and place them into the
 *                  'completed' list.
 * @nbs: the non-blocking set
 * Notes:  Call before pscrpc_check_set(), pscrpc_check_set() manages
 *        'set_remaining'..  If there are problems look at pscrpc_set_wait(),
 *        there you'll find a more sophisticated and proper use of sets.
 *        pscrpc_set_wait() deals with blocking set nevertheless much applies
 *        here as well.
 */
int
pscrpc_nbreqset_reap(struct pscrpc_nbreqset *nbs)
{
	int rc, nreaped = 0, nchecked = 0, timeout = 1;
	struct pscrpc_request_set *set = &nbs->nb_reqset;
	struct pscrpc_request *req, *next;
	struct l_wait_info lwi;

	spinlock(&nbs->nb_lock);
	if (nbs->nb_flags & NBREQSET_WORK_INPROG) {
		freelock(&nbs->nb_lock);
		RETURN(0);
	} else {
		nbs->nb_flags |= NBREQSET_WORK_INPROG;
		freelock(&nbs->nb_lock);
	}

	lwi = LWI_TIMEOUT(timeout, NULL, NULL);
	pscrpc_cli_wait_event(&set->set_waitq,
		       (nreaped=pscrpc_check_set(set, 0)), &lwi);

	if (!nreaped) {
		spinlock(&nbs->nb_lock);
		nbs->nb_flags &= ~NBREQSET_WORK_INPROG;
		freelock(&nbs->nb_lock);
		return (0);
	}

	pscrpc_set_lock(set);

	psclist_for_each_entry_safe(req, next,
	    &set->set_requests, rq_set_chain_lentry) {
		nchecked++;
		DEBUG_REQ(PLL_INFO, req, "reap if Completed");
		/*
		 * Move sent rpcs to the set_requests list
		 */
		if (req->rq_phase == PSCRPC_RQ_PHASE_COMPLETE) {
			psclist_del(&req->rq_set_chain_lentry);
			atomic_dec(&nbs->nb_outstanding);
			nreaped++;
			/*
			 * paranoia
			 */
			psc_assert(atomic_read(&nbs->nb_outstanding) >= 0);
			/*
			 * This is the caller's last shot at accessing
			 *  this msg..
			 * Let the callback deal with its own
			 *  error handling, we can't do much from here
			 */
			rc = 0;
			if (nbs->nb_callback)
				rc = nbs->nb_callback(req, &req->rq_async_args);
			/*
			 * Be done with it..
			 */
			if (rc == 0)
				pscrpc_req_finished(req);
		}
	}
	freelock(&set->set_lock);
{'fraction_non_alphanumeric': 0.12501694915254238, 'fraction_numerical': 0.0024406779661016948, 'mean_word_length': 4.090407177363699, 'pattern_counts': {'":': 0, '<': 5, '<?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': '6618423', 'n_tokens_mistral': 3176, 'n_tokens_neox': 2976, 'n_words': 588}