<commit_before>/*
 * Copyright (c) 2011 by Internet Systems Consortium, Inc. ("ISC")
 *
 * Permission to use, copy, modify, and/or distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
 * OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */

static void
free_seqsrcs(nmsg_input_t input) {
	struct nmsg_seqsrc *seqsrc, *seqsrc_next;

	seqsrc = ISC_LIST_HEAD(input->stream->seqsrcs);
	while (seqsrc != NULL) {
		if (_nmsg_global_debug >= 5) {
			fprintf(stderr, "%s: source %s/%hu: "
				"count=%" PRIu64 " dropped=%" PRIu64 " (%.4f)\n",
				__func__,
				seqsrc->addr_str, ntohs(seqsrc->key.port),
				seqsrc->count, seqsrc->count_dropped,
				(seqsrc->count_dropped) /
					(seqsrc->count_dropped + seqsrc->count + 1.0)
			);
		}
		input->stream->count_recv += seqsrc->count;
		input->stream->count_drop += seqsrc->count_dropped;
		seqsrc_next = ISC_LIST_NEXT(seqsrc, link);
		free(seqsrc);
		seqsrc = seqsrc_next;
	}

	if (_nmsg_global_debug >= 4 && input->stream->count_recv > 0) {
		double frac = (input->stream->count_drop + 0.0) /
			(input->stream->count_recv + input->stream->count_drop + 0.0);
		fprintf(stderr,
			"%s: input=%p count_recv=%" PRIu64 " count_drop=%" PRIu64 " (%.4f)\n",
			__func__,
			input,
			input->stream->count_recv,
			input->stream->count_drop,
			frac
		);
	}
}

static void
reset_seqsrc(struct nmsg_seqsrc *seqsrc, const char *why) {
	if (_nmsg_global_debug >= 5) {
		fprintf(stderr,
			"%s: resetting source %s/%hu: %s: "
			"count= %" PRIu64 " count_dropped= %" PRIu64 "\n",
			__func__,
			seqsrc->addr_str,
			ntohs(seqsrc->key.port),
			why,
			seqsrc->count,
			seqsrc->count_dropped
		);
	}
	seqsrc->sequence = 0;
	seqsrc->count = 0;
	seqsrc->count_dropped = 0;
}

static void
input_update_seqsrc(nmsg_input_t input, Nmsg__Nmsg *nmsg, struct nmsg_seqsrc *seqsrc) {
	if (!(input->type == nmsg_input_type_stream &&
	      input->stream->type == nmsg_stream_type_sock &&
	      nmsg != NULL && nmsg->has_sequence))
	{
		return;
	}

	seqsrc->count += 1;

	if (seqsrc->sequence > 0 &&
	    nmsg->has_seq_state &&
	    nmsg->seq_state == NMSG__SEQ_STATE__INIT)
	{
		reset_seqsrc(seqsrc, "reinitialized");
	} else if (seqsrc->sequence != nmsg->sequence) {
		int64_t delta = ((int64_t)(nmsg->sequence)) -
				((int64_t)(seqsrc->sequence));
		delta %= 4294967296;
		if (delta < 0)
			delta += 4294967296;
		if (seqsrc->init)
			seqsrc->init = false;
		else
			seqsrc->count_dropped += delta;

		if (_nmsg_global_debug >= 5) {
			fprintf(stderr,
				"%s: source %s/%hu: expected sequence (%u) != wire sequence (%u), "
				"delta %" PRIu64 ", drop fraction %.4f\n",
				__func__,
				seqsrc->addr_str, ntohs(seqsrc->key.port),
				seqsrc->sequence,
				nmsg->sequence,
				delta,
				(seqsrc->count_dropped) /
					(seqsrc->count_dropped + seqsrc->count + 1.0)
			);
		}
	}
	seqsrc->sequence = nmsg->sequence + 1;
}

static void
get_seqsrc(nmsg_input_t input, struct nmsg_seqsrc **ss, struct sockaddr_storage *addr_ss) {
	struct nmsg_seqsrc *seqsrc, *seqsrc_next;
	struct sockaddr_in *sai;
	struct sockaddr_in6 *sai6;
<commit_msg>input_seqsrc: input_update_seqsrc(): reset the seqsrc on an implausibly large delta<commit_after>/*
 * Copyright (c) 2011 by Internet Systems Consortium, Inc. ("ISC")
 *
 * Permission to use, copy, modify, and/or distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
 * OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */

static void
free_seqsrcs(nmsg_input_t input) {
	struct nmsg_seqsrc *seqsrc, *seqsrc_next;

	seqsrc = ISC_LIST_HEAD(input->stream->seqsrcs);
	while (seqsrc != NULL) {
		if (_nmsg_global_debug >= 5) {
			fprintf(stderr, "%s: source %s/%hu: "
				"count=%" PRIu64 " dropped=%" PRIu64 " (%.4f)\n",
				__func__,
				seqsrc->addr_str, ntohs(seqsrc->key.port),
				seqsrc->count, seqsrc->count_dropped,
				(seqsrc->count_dropped) /
					(seqsrc->count_dropped + seqsrc->count + 1.0)
			);
		}
		input->stream->count_recv += seqsrc->count;
		input->stream->count_drop += seqsrc->count_dropped;
		seqsrc_next = ISC_LIST_NEXT(seqsrc, link);
		free(seqsrc);
		seqsrc = seqsrc_next;
	}

	if (_nmsg_global_debug >= 4 && input->stream->count_recv > 0) {
		double frac = (input->stream->count_drop + 0.0) /
			(input->stream->count_recv + input->stream->count_drop + 0.0);
		fprintf(stderr,
			"%s: input=%p count_recv=%" PRIu64 " count_drop=%" PRIu64 " (%.4f)\n",
			__func__,
			input,
			input->stream->count_recv,
			input->stream->count_drop,
			frac
		);
	}
}

static void
reset_seqsrc(struct nmsg_seqsrc *seqsrc, const char *why) {
	if (_nmsg_global_debug >= 5) {
		fprintf(stderr,
			"%s: resetting source %s/%hu: %s: "
			"count= %" PRIu64 " count_dropped= %" PRIu64 "\n",
			__func__,
			seqsrc->addr_str,
			ntohs(seqsrc->key.port),
			why,
			seqsrc->count,
			seqsrc->count_dropped
		);
	}
	seqsrc->sequence = 0;
	seqsrc->count = 0;
	seqsrc->count_dropped = 0;
}

static void
input_update_seqsrc(nmsg_input_t input, Nmsg__Nmsg *nmsg, struct nmsg_seqsrc *seqsrc) {
	if (!(input->type == nmsg_input_type_stream &&
	      input->stream->type == nmsg_stream_type_sock &&
	      nmsg != NULL && nmsg->has_sequence))
	{
		return;
	}

	seqsrc->count += 1;

	if (seqsrc->sequence > 0 &&
	    nmsg->has_seq_state &&
	    nmsg->seq_state == NMSG__SEQ_STATE__INIT)
	{
		reset_seqsrc(seqsrc, "reinitialized");
	} else if (seqsrc->sequence != nmsg->sequence) {
		int64_t delta = ((int64_t)(nmsg->sequence)) -
				((int64_t)(seqsrc->sequence));
		delta %= 4294967296;
		if (delta < 0)
			delta += 4294967296;

		if (seqsrc->init) {
			/* don't count the delta as a drop, since the seqsrc
			 * has just been initialized */
			seqsrc->init = false;
			goto out;
		}

		if (delta > 1048576) {
			/* don't count the delta as a drop, since the delta
			 * is implausibly large */
			reset_seqsrc(seqsrc, "implausibly large delta");
			goto out;
		}

		/* count the delta as a drop */
		seqsrc->count_dropped += delta;

		if (_nmsg_global_debug >= 5) {
			fprintf(stderr,
				"%s: source %s/%hu: expected sequence (%u) != wire sequence (%u), "
				"delta %" PRIu64 ", drop fraction %.4f\n",
				__func__,
				seqsrc->addr_str, ntohs(seqsrc->key.port),
				seqsrc->sequence,
				nmsg->sequence,
				delta,
				(seqsrc->count_dropped) /
					(seqsrc->count_dropped + seqsrc->count + 1.0)
			);
		}
	}
out:
	seqsrc->sequence = nmsg->sequence + 1;
}

static void
get_seqsrc(nmsg_input_t input, struct nmsg_seqsrc **ss, struct sockaddr_storage *addr_ss) {
	struct nmsg_seqsrc *seqsrc, *seqsrc_next;
	struct sockaddr_in *sai;
	struct sockaddr_in6 *sai6;

{'fraction_non_alphanumeric': 0.13704720805544657, 'fraction_numerical': 0.01896168432064862, 'mean_word_length': 3.738537794299876, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 132, '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': '1261729', 'n_tokens_mistral': 3148, 'n_tokens_neox': 2753, 'n_words': 699}