<commit_before>/*
 * miniRPC - TCP RPC library with asynchronous operations
 *
 * Copyright (C) 2007-2008 Carnegie Mellon University
 *
 * 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 <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <netdb.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <errno.h>
#include <assert.h>
#include <pthread.h>
#define MINIRPC_INTERNAL
#include "internal.h"

static void conn_set_free(struct mrpc_conn_set *set);
static void conn_free(struct mrpc_connection *conn);
static void conn_kill(struct mrpc_connection *conn,
			enum mrpc_disc_reason reason);
static int try_close_fd(struct mrpc_connection *conn);

static int setsockoptval(int fd, int level, int optname, int value)
{
	if (setsockopt(fd, level, optname, &value, sizeof(value)))
		return errno;
	return 0;
}

static void conn_set_start_shutdown(struct mrpc_conn_set *set)
{
	/* We can't call conn_set_free() directly from conn_set_put():
	   if we were called from an event thread, that would deadlock.
	   Have the listener thread clean up the set. */
	selfpipe_set(set->shutdown_pipe);
	selfpipe_set(set->events_notify_pipe);
}

#define REFCOUNT_GET_FUNC(modifier, name, type, member)			\
	modifier void name(type *item)					\
	{								\
		gint old;						\
									\
		if (item == NULL)					\
			return;						\
		old=g_atomic_int_exchange_and_add(&item->member, 1);	\
		assert(old > 0);					\
	}
#define REFCOUNT_PUT_FUNC(modifier, name, type, member, cleanup_action)	\
	modifier void name(type *item)					\
	{								\
		if (item == NULL)					\
			return;						\
		if (g_atomic_int_dec_and_test(&item->member)) {		\
			cleanup_action;					\
		}							\
	}
REFCOUNT_GET_FUNC(static, conn_set_get, struct mrpc_conn_set, refs)
REFCOUNT_PUT_FUNC(static, conn_set_put, struct mrpc_conn_set, refs,
			conn_set_start_shutdown(item))
REFCOUNT_GET_FUNC(exported, mrpc_conn_set_ref, struct mrpc_conn_set,
			user_refs)
REFCOUNT_PUT_FUNC(exported, mrpc_conn_set_unref, struct mrpc_conn_set,
			user_refs, conn_set_put(item))
REFCOUNT_GET_FUNC(, conn_get, struct mrpc_connection, refs)
REFCOUNT_PUT_FUNC(, conn_put, struct mrpc_connection, refs,
			conn_free(item))
REFCOUNT_GET_FUNC(exported, mrpc_conn_ref, struct mrpc_connection, user_refs)
REFCOUNT_PUT_FUNC(exported, mrpc_conn_unref, struct mrpc_connection, user_refs,
			conn_put(item))
#undef REFCOUNT_GET_FUNC
#undef REFCOUNT_PUT_FUNC

/* Returns error code if the connection should be killed */
static mrpc_status_t process_incoming_header(struct mrpc_connection *conn)
{
	mrpc_status_t ret;

	ret=unserialize((xdrproc_t)xdr_mrpc_header, conn->recv_hdr_buf,
				MINIRPC_HEADER_LEN, &conn->recv_msg->hdr,
				sizeof(conn->recv_msg->hdr));
	if (ret) {
		queue_ioerr_event(conn, "Header deserialize failure");<commit_msg>Assert if decrementing refcount takes it negative<commit_after>/*
 * miniRPC - TCP RPC library with asynchronous operations
 *
 * Copyright (C) 2007-2008 Carnegie Mellon University
 *
 * 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 <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <netdb.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <errno.h>
#include <assert.h>
#include <pthread.h>
#define MINIRPC_INTERNAL
#include "internal.h"

static void conn_set_free(struct mrpc_conn_set *set);
static void conn_free(struct mrpc_connection *conn);
static void conn_kill(struct mrpc_connection *conn,
			enum mrpc_disc_reason reason);
static int try_close_fd(struct mrpc_connection *conn);

static int setsockoptval(int fd, int level, int optname, int value)
{
	if (setsockopt(fd, level, optname, &value, sizeof(value)))
		return errno;
	return 0;
}

static void conn_set_start_shutdown(struct mrpc_conn_set *set)
{
	/* We can't call conn_set_free() directly from conn_set_put():
	   if we were called from an event thread, that would deadlock.
	   Have the listener thread clean up the set. */
	selfpipe_set(set->shutdown_pipe);
	selfpipe_set(set->events_notify_pipe);
}

#define REFCOUNT_GET_FUNC(modifier, name, type, member)			\
	modifier void name(type *item)					\
	{								\
		gint old;						\
									\
		if (item == NULL)					\
			return;						\
		old=g_atomic_int_exchange_and_add(&item->member, 1);	\
		assert(old > 0);					\
	}
#define REFCOUNT_PUT_FUNC(modifier, name, type, member, cleanup_action)	\
	modifier void name(type *item)					\
	{								\
		gint old;						\
									\
		if (item == NULL)					\
			return;						\
		old=g_atomic_int_exchange_and_add(&item->member, -1);	\
		assert(old > 0);					\
		if (old == 1) {						\
			cleanup_action;					\
		}							\
	}
REFCOUNT_GET_FUNC(static, conn_set_get, struct mrpc_conn_set, refs)
REFCOUNT_PUT_FUNC(static, conn_set_put, struct mrpc_conn_set, refs,
			conn_set_start_shutdown(item))
REFCOUNT_GET_FUNC(exported, mrpc_conn_set_ref, struct mrpc_conn_set,
			user_refs)
REFCOUNT_PUT_FUNC(exported, mrpc_conn_set_unref, struct mrpc_conn_set,
			user_refs, conn_set_put(item))
REFCOUNT_GET_FUNC(, conn_get, struct mrpc_connection, refs)
REFCOUNT_PUT_FUNC(, conn_put, struct mrpc_connection, refs,
			conn_free(item))
REFCOUNT_GET_FUNC(exported, mrpc_conn_ref, struct mrpc_connection, user_refs)
REFCOUNT_PUT_FUNC(exported, mrpc_conn_unref, struct mrpc_connection, user_refs,
			conn_put(item))
#undef REFCOUNT_GET_FUNC
#undef REFCOUNT_PUT_FUNC

/* Returns error code if the connection should be killed */
static mrpc_status_t process_incoming_header(struct mrpc_connection *conn)
{
	mrpc_status_t ret;

	ret=unserialize((xdrproc_t)xdr_mrpc_header, conn->recv_hdr_buf,
				MINIRPC_HEADER_LEN, &conn->recv_msg->hdr,
				sizeof(conn->recv_msg->hdr));
	if (ret) {
		queue_ioerr_event(conn, "Header deserialize failure");
{'fraction_non_alphanumeric': 0.10213929548013512, 'fraction_numerical': 0.004664629242399871, 'mean_word_length': 4.627149321266969, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 46, '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': '4578023', 'n_tokens_mistral': 2513, 'n_tokens_neox': 2269, 'n_words': 540}