<commit_before>//
// Copyright 2019 Staysail Systems, Inc. <<EMAIL>>
// Copyright 2018 Capitar IT Group BV <<EMAIL>>
// Copyright 2018 Devolutions <<EMAIL>>
//
// This software is supplied under the terms of the MIT License, a
// copy of which should be located in the distribution where this
// file was obtained (LICENSE.txt).  A copy of the license may also be
// found online at https://opensource.org/licenses/MIT.
//

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "core/nng_impl.h"

// Inproc transport.  This just transports messages from one
// peer to another.  The inproc transport is only valid within the same
// process.

typedef struct inproc_pair inproc_pair;
typedef struct inproc_pipe inproc_pipe;
typedef struct inproc_ep   inproc_ep;

typedef struct {
	nni_mtx  mx;
	nni_list servers;
} inproc_global;

// inproc_pipe represents one half of a connection.
struct inproc_pipe {
	const char * addr;
	inproc_pair *pair;
	nni_msgq *   rq;
	nni_msgq *   wq;
	nni_pipe *   npipe;
	uint16_t     peer;
	uint16_t     proto;
};

// inproc_pair represents a pair of pipes.  Because we control both
// sides of the pipes, we can allocate and free this in one structure.
struct inproc_pair {
	nni_mtx      mx;
	int          refcnt;
	nni_msgq *   q[2];
	inproc_pipe *pipes[2];
};

struct inproc_ep {
	const char *  addr;
	bool          listener;
	nni_list_node node;
	uint16_t      proto;
	nni_cv        cv;
	nni_list      clients;
	nni_list      aios;
	size_t        rcvmax;
	nni_mtx       mtx;
	nni_dialer *  ndialer;
	nni_listener *nlistener;
};

// nni_inproc is our global state - this contains the list of active endpoints
// which we use for coordinating rendezvous.
static inproc_global nni_inproc;

static int
inproc_init(void)
{
	NNI_LIST_INIT(&nni_inproc.servers, inproc_ep, node);

	nni_mtx_init(&nni_inproc.mx);
	return (0);
}

static void
inproc_fini(void)
{
	nni_mtx_fini(&nni_inproc.mx);
}

static void
inproc_pipe_close(void *arg)
{
	inproc_pipe *pipe = arg;

	if (pipe->rq != NULL) {
		nni_msgq_close(pipe->rq);
	}
	if (pipe->wq != NULL) {
		nni_msgq_close(pipe->wq);
	}
}

// inproc_pair destroy is called when both pipe-ends of the pipe
// have been destroyed.
static void
inproc_pair_destroy(inproc_pair *pair)
{
	nni_msgq_fini(pair->q[0]);
	nni_msgq_fini(pair->q[1]);
	nni_mtx_fini(&pair->mx);
	NNI_FREE_STRUCT(pair);
}

static int
inproc_pipe_alloc(inproc_pipe **pipep, inproc_ep *ep)
{
	inproc_pipe *pipe;

	if ((pipe = NNI_ALLOC_STRUCT(pipe)) == NULL) {
		return (NNG_ENOMEM);
	}

	pipe->proto = ep->proto;
	pipe->addr  = ep->addr;
	*pipep      = pipe;
	return (0);
}

static int
inproc_pipe_init(void *arg, nni_pipe *p)
{
	inproc_pipe *pipe = arg;
	pipe->npipe       = p;

	return (0);
}

static void
inproc_pipe_fini(void *arg)
{
	inproc_pipe *pipe = arg;
	inproc_pair *pair;

	if ((pair = pipe->pair) != NULL) {
		// If we are the last peer, then toss the pair structure.
		nni_mtx_lock(&pair->mx);
		if (pair->pipes[0] == pipe) {
			pair->pipes[0] = NULL;
		} else if (pair->pipes[1] == pipe) {
			pair->pipes[1] = NULL;
		}
		pair->refcnt--;
		if (pair->refcnt == 0) {
			nni_mtx_unlock(&pair->mx);
			inproc_pair_destroy(pair);
		} else {
			nni_mtx_unlock(&pair->mx);
		}
	}

	NNI_FREE_STRUCT(pipe);
}

static void
inproc_pipe_send(void *arg, nni_aio *aio)
{
	inproc_pipe *pipe = arg;
	nni_msg *    msg  = nni_aio_get_msg(aio);
	char *       h;
	size_t       l;
	int          rv;

	// We need to move any header data to the body, because the other
	// side won't know what to do otherwise.
	h = nni_msg_header(msg);
	l = nni_msg_header_len(msg);
	if ((rv = nni_msg_insert(msg, h, l)) != 0) {
		nni_aio_finish(aio, rv, nni_aio_count(aio));
		return;
	}
	nni_msg_header_chop(msg, l);
	nni_msgq_aio_put(pipe->wq, aio);
}

static void
inproc_pipe_recv(void *arg, nni_aio *aio)
{
	inproc_pipe *pipe = arg;

	nni_msgq_aio_get(pipe->rq, aio);
}

static uint16_t
inproc_pipe_peer(void *arg)
{
	inproc_pipe *pipe = arg;

	return (pipe->peer);
}
<commit_msg>inproc can use msg_header_clear to save a branch.<commit_after>//
// Copyright 2020 Staysail Systems, Inc. <<EMAIL>>
// Copyright 2018 Capitar IT Group BV <<EMAIL>>
// Copyright 2018 Devolutions <<EMAIL>>
//
// This software is supplied under the terms of the MIT License, a
// copy of which should be located in the distribution where this
// file was obtained (LICENSE.txt).  A copy of the license may also be
// found online at https://opensource.org/licenses/MIT.
//

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "core/nng_impl.h"

// Inproc transport.  This just transports messages from one
// peer to another.  The inproc transport is only valid within the same
// process.

typedef struct inproc_pair inproc_pair;
typedef struct inproc_pipe inproc_pipe;
typedef struct inproc_ep   inproc_ep;

typedef struct {
	nni_mtx  mx;
	nni_list servers;
} inproc_global;

// inproc_pipe represents one half of a connection.
struct inproc_pipe {
	const char * addr;
	inproc_pair *pair;
	nni_msgq *   rq;
	nni_msgq *   wq;
	nni_pipe *   npipe;
	uint16_t     peer;
	uint16_t     proto;
};

// inproc_pair represents a pair of pipes.  Because we control both
// sides of the pipes, we can allocate and free this in one structure.
struct inproc_pair {
	nni_mtx      mx;
	int          refcnt;
	nni_msgq *   q[2];
	inproc_pipe *pipes[2];
};

struct inproc_ep {
	const char *  addr;
	bool          listener;
	nni_list_node node;
	uint16_t      proto;
	nni_cv        cv;
	nni_list      clients;
	nni_list      aios;
	size_t        rcvmax;
	nni_mtx       mtx;
	nni_dialer *  ndialer;
	nni_listener *nlistener;
};

// nni_inproc is our global state - this contains the list of active endpoints
// which we use for coordinating rendezvous.
static inproc_global nni_inproc;

static int
inproc_init(void)
{
	NNI_LIST_INIT(&nni_inproc.servers, inproc_ep, node);

	nni_mtx_init(&nni_inproc.mx);
	return (0);
}

static void
inproc_fini(void)
{
	nni_mtx_fini(&nni_inproc.mx);
}

static void
inproc_pipe_close(void *arg)
{
	inproc_pipe *pipe = arg;

	if (pipe->rq != NULL) {
		nni_msgq_close(pipe->rq);
	}
	if (pipe->wq != NULL) {
		nni_msgq_close(pipe->wq);
	}
}

// inproc_pair destroy is called when both pipe-ends of the pipe
// have been destroyed.
static void
inproc_pair_destroy(inproc_pair *pair)
{
	nni_msgq_fini(pair->q[0]);
	nni_msgq_fini(pair->q[1]);
	nni_mtx_fini(&pair->mx);
	NNI_FREE_STRUCT(pair);
}

static int
inproc_pipe_alloc(inproc_pipe **pipep, inproc_ep *ep)
{
	inproc_pipe *pipe;

	if ((pipe = NNI_ALLOC_STRUCT(pipe)) == NULL) {
		return (NNG_ENOMEM);
	}

	pipe->proto = ep->proto;
	pipe->addr  = ep->addr;
	*pipep      = pipe;
	return (0);
}

static int
inproc_pipe_init(void *arg, nni_pipe *p)
{
	inproc_pipe *pipe = arg;
	pipe->npipe       = p;

	return (0);
}

static void
inproc_pipe_fini(void *arg)
{
	inproc_pipe *pipe = arg;
	inproc_pair *pair;

	if ((pair = pipe->pair) != NULL) {
		// If we are the last peer, then toss the pair structure.
		nni_mtx_lock(&pair->mx);
		if (pair->pipes[0] == pipe) {
			pair->pipes[0] = NULL;
		} else if (pair->pipes[1] == pipe) {
			pair->pipes[1] = NULL;
		}
		pair->refcnt--;
		if (pair->refcnt == 0) {
			nni_mtx_unlock(&pair->mx);
			inproc_pair_destroy(pair);
		} else {
			nni_mtx_unlock(&pair->mx);
		}
	}

	NNI_FREE_STRUCT(pipe);
}

static void
inproc_pipe_send(void *arg, nni_aio *aio)
{
	inproc_pipe *pipe = arg;
	nni_msg *    msg  = nni_aio_get_msg(aio);
	char *       h;
	size_t       l;
	int          rv;

	// We need to move any header data to the body, because the other
	// side won't know what to do otherwise.
	h = nni_msg_header(msg);
	l = nni_msg_header_len(msg);
	if ((rv = nni_msg_insert(msg, h, l)) != 0) {
		nni_aio_finish(aio, rv, nni_aio_count(aio));
		return;
	}
	nni_msg_header_clear(msg);
	nni_msgq_aio_put(pipe->wq, aio);
}

static void
inproc_pipe_recv(void *arg, nni_aio *aio)
{
	inproc_pipe *pipe = arg;

	nni_msgq_aio_get(pipe->rq, aio);
}

static uint16_t
inproc_pipe_peer(void *arg)
{
	inproc_pipe *pipe = arg;

	return (pipe->peer);
}

{'fraction_non_alphanumeric': 0.12176938369781312, 'fraction_numerical': 0.008200795228628231, 'mean_word_length': 3.552601809954751, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 71, 'https://': 2, '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': '494571', 'n_tokens_mistral': 3366, 'n_tokens_neox': 3241, 'n_words': 832}