<commit_before>/*
 * Copyright (c) 2008 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.
 */

/* Import. */

#include "nmsg_port.h"

#include <assert.h>
#include <pthread.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#define ISC_CHECK_NONE 1
#include <isc/list.h>

#include "private.h"
#include "input.h"
#include "io.h"
#include "output.h"
#include "payload.h"
#include "pbmod.h"
#include "pbmodset.h"
#include "time.h"

/* Data structures. */

struct nmsg_io_pres {
	ISC_LINK(struct nmsg_io_pres)	link;
	FILE				*fp;
	nmsg_pbmod			mod;
	nmsg_pres			pres;
	pthread_mutex_t			lock;
	struct timespec			last;
	void				*clos, *user;
	uint64_t			count_pres_out, count_pres_payload_out;
};

struct nmsg_io_buf {
	ISC_LINK(struct nmsg_io_buf)	link;
	nmsg_buf			buf;
	pthread_mutex_t			lock;
	struct timespec			last;
	void				*user;
	uint64_t			count_nmsg_out, count_nmsg_payload_out;
};

struct nmsg_io {
	ISC_LIST(struct nmsg_io_buf)	r_nmsg;
	ISC_LIST(struct nmsg_io_buf)	w_nmsg;
	ISC_LIST(struct nmsg_io_pres)	r_pres;
	ISC_LIST(struct nmsg_io_pres)	w_pres;
	ISC_LIST(struct nmsg_io_thr)	iothreads;
	bool				quiet, zlibout;
	char				*endline;
	int				debug;
	nmsg_io_closed_fp		closed_fp;
	nmsg_io_output_mode		output_mode;
	nmsg_pbmodset			ms;
	pthread_mutex_t			lock;
	size_t				max;
	unsigned			count, interval;
	unsigned			n_user, user[2];
	volatile bool			stop, stopped;
};

struct nmsg_io_thr {
	ISC_LINK(struct nmsg_io_thr)	link;
	pthread_t			thr;
	nmsg_io				io;
	nmsg_res			res;
	struct timespec			now;
	union {
		struct nmsg_io_buf	*iobuf;
		struct nmsg_io_pres	*iopres;
	};
	uint64_t			count_nmsg_in, count_nmsg_payload_in;
	uint64_t			count_pres_in, count_pres_payload_in;
};

/* Forward. */

static Nmsg__NmsgPayload *make_nmsg_payload(struct nmsg_io_thr *, uint8_t *,
					    size_t);
static nmsg_res write_nmsg(struct nmsg_io_thr *, struct nmsg_io_buf *,
			   Nmsg__Nmsg *);
static nmsg_res write_nmsg_dup(struct nmsg_io_thr *, struct nmsg_io_buf *,
			       const Nmsg__Nmsg *);
static nmsg_res write_nmsg_payload(struct nmsg_io_thr *, struct nmsg_io_buf *,
				   Nmsg__NmsgPayload *);
static nmsg_res write_pres(struct nmsg_io_thr *, struct nmsg_io_pres *,
			   const Nmsg__Nmsg *);
static void *thr_nmsg(void *);
static void *thr_pres(void *);
static void init_timespec_intervals(nmsg_io);

/* Export. */

nmsg_io
nmsg_io_init(nmsg_pbmodset ms, size_t max) {
	struct nmsg_io *io;

	io = calloc(1, sizeof(*io));
	if (io == NULL)
		return (NULL);
	io->ms = ms;
	io->output_mode = nmsg_io_output_mode_stripe;
	pthread_mutex_init(&io->lock, NULL);
	ISC_LIST_INIT(io->iothreads);

	if (max == 0)
		io->max = NMSG_WBUFSZ_ETHER;
	else if (max < NMSG_WBUFSZ_MIN)
		io->max = NMSG_WBUFSZ_MIN;
	else if (max > NMSG_WBUFSZ_MAX)
		io->max = NMSG_WBUFSZ_MAX;
	else
		io->max = max;

	/* reserve space for the outer nmsg layer */
	io->max -= (NMSG_HDRSZ + NMSG_LENHDRSZ_V2 + NMSG_PAYHDRSZ);

	return (io);
}

void
nmsg_io_breakloop(nmsg_io io) {
	struct timespec ts = { .tv_sec = 1, .tv_nsec = 0 };

	io->stop = true;
	nmsg_time_sleep(&ts);
	if (io->stopped != true) {
		struct nmsg_io_thr *iothr;

		for (iothr = ISC_LIST_HEAD(io->iothreads);
		     iothr != NULL;
		     iothr = ISC_LIST_NEXT(iothr, link))
		{
			pthread_cancel(iothr->thr);
		}
	}
}

nmsg_res
nmsg_io_loop(nmsg_io io) {
	nmsg_res res;
	struct nmsg_io_buf *iobuf;
	struct nmsg_io_pres *iopres;
	struct nmsg_io_thr *iothr, *iothr_next;

	res = nmsg_res_success;

	if (io->interval > 0)
		init_timespec_intervals(io);

	if (io->endline == NULL)
		io->endline = strdup("\n");
<commit_msg>io: enable zlibout on outputs if requested<commit_after>/*
 * Copyright (c) 2008 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.
 */

/* Import. */

#include "nmsg_port.h"

#include <assert.h>
#include <pthread.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#define ISC_CHECK_NONE 1
#include <isc/list.h>

#include "private.h"
#include "input.h"
#include "io.h"
#include "output.h"
#include "payload.h"
#include "pbmod.h"
#include "pbmodset.h"
#include "time.h"

/* Data structures. */

struct nmsg_io_pres {
	ISC_LINK(struct nmsg_io_pres)	link;
	FILE				*fp;
	nmsg_pbmod			mod;
	nmsg_pres			pres;
	pthread_mutex_t			lock;
	struct timespec			last;
	void				*clos, *user;
	uint64_t			count_pres_out, count_pres_payload_out;
};

struct nmsg_io_buf {
	ISC_LINK(struct nmsg_io_buf)	link;
	nmsg_buf			buf;
	pthread_mutex_t			lock;
	struct timespec			last;
	void				*user;
	uint64_t			count_nmsg_out, count_nmsg_payload_out;
};

struct nmsg_io {
	ISC_LIST(struct nmsg_io_buf)	r_nmsg;
	ISC_LIST(struct nmsg_io_buf)	w_nmsg;
	ISC_LIST(struct nmsg_io_pres)	r_pres;
	ISC_LIST(struct nmsg_io_pres)	w_pres;
	ISC_LIST(struct nmsg_io_thr)	iothreads;
	bool				quiet, zlibout;
	char				*endline;
	int				debug;
	nmsg_io_closed_fp		closed_fp;
	nmsg_io_output_mode		output_mode;
	nmsg_pbmodset			ms;
	pthread_mutex_t			lock;
	size_t				max;
	unsigned			count, interval;
	unsigned			n_user, user[2];
	volatile bool			stop, stopped;
};

struct nmsg_io_thr {
	ISC_LINK(struct nmsg_io_thr)	link;
	pthread_t			thr;
	nmsg_io				io;
	nmsg_res			res;
	struct timespec			now;
	union {
		struct nmsg_io_buf	*iobuf;
		struct nmsg_io_pres	*iopres;
	};
	uint64_t			count_nmsg_in, count_nmsg_payload_in;
	uint64_t			count_pres_in, count_pres_payload_in;
};

/* Forward. */

static Nmsg__NmsgPayload *make_nmsg_payload(struct nmsg_io_thr *, uint8_t *,
					    size_t);
static nmsg_res write_nmsg(struct nmsg_io_thr *, struct nmsg_io_buf *,
			   Nmsg__Nmsg *);
static nmsg_res write_nmsg_dup(struct nmsg_io_thr *, struct nmsg_io_buf *,
			       const Nmsg__Nmsg *);
static nmsg_res write_nmsg_payload(struct nmsg_io_thr *, struct nmsg_io_buf *,
				   Nmsg__NmsgPayload *);
static nmsg_res write_pres(struct nmsg_io_thr *, struct nmsg_io_pres *,
			   const Nmsg__Nmsg *);
static void *thr_nmsg(void *);
static void *thr_pres(void *);
static void init_timespec_intervals(nmsg_io);

/* Export. */

nmsg_io
nmsg_io_init(nmsg_pbmodset ms, size_t max) {
	struct nmsg_io *io;

	io = calloc(1, sizeof(*io));
	if (io == NULL)
		return (NULL);
	io->ms = ms;
	io->output_mode = nmsg_io_output_mode_stripe;
	pthread_mutex_init(&io->lock, NULL);
	ISC_LIST_INIT(io->iothreads);

	if (max == 0)
		io->max = NMSG_WBUFSZ_ETHER;
	else if (max < NMSG_WBUFSZ_MIN)
		io->max = NMSG_WBUFSZ_MIN;
	else if (max > NMSG_WBUFSZ_MAX)
		io->max = NMSG_WBUFSZ_MAX;
	else
		io->max = max;

	/* reserve space for the outer nmsg layer */
	io->max -= (NMSG_HDRSZ + NMSG_LENHDRSZ_V2 + NMSG_PAYHDRSZ);

	return (io);
}

void
nmsg_io_breakloop(nmsg_io io) {
	struct timespec ts = { .tv_sec = 1, .tv_nsec = 0 };

	io->stop = true;
	nmsg_time_sleep(&ts);
	if (io->stopped != true) {
		struct nmsg_io_thr *iothr;

		for (iothr = ISC_LIST_HEAD(io->iothreads);
		     iothr != NULL;
		     iothr = ISC_LIST_NEXT(iothr, link))
		{
			pthread_cancel(iothr->thr);
		}
	}
}

nmsg_res
nmsg_io_loop(nmsg_io io) {
	nmsg_res res;
	struct nmsg_io_buf *iobuf;
	struct nmsg_io_pres *iopres;
	struct nmsg_io_thr *iothr, *iothr_next;

	res = nmsg_res_success;

	if (io->interval > 0)
		init_timespec_intervals(io);

	if (io->endline == NULL)
		io->endline = strdup("\n");

	if (io->zlibout == true) {
		for (iobuf = ISC_LIST_HEAD(io->w_nmsg);
		     iobuf != NULL;
		     iobuf = ISC_LIST_NEXT(iobuf, link))
		{
			nmsg_output_set_zlibout(iobuf->buf, true);
		}
	}

{'fraction_non_alphanumeric': 0.1054047992823503, 'fraction_numerical': 0.004709576138147566, 'mean_word_length': 4.224956063268893, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 60, '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': '279312', 'n_tokens_mistral': 3952, 'n_tokens_neox': 3646, 'n_words': 875}