<commit_before>/*
 *  This file is part of carbon-c-relay.
 *
 *  carbon-c-relay is free software: you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation, either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  carbon-c-relay is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with carbon-c-relay.  If not, see <http://www.gnu.org/licenses/>.
 */


#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <errno.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <sys/resource.h>

#include "relay.h"
#include "queue.h"
#include "dispatcher.h"
#include "collector.h"

enum servertype {
	ORIGIN,
	BACKUP,
	INTERNAL
};

typedef struct _server {
	const char *ip;
	unsigned short port;
	struct sockaddr_in serv_addr;
	int fd;
	queue *queue;
	enum servertype type;
	pthread_t tid;
	struct _server **secondaries;
	size_t secondariescnt;
	char failure:1;
	char running:1;
	char keep_running:1;
	size_t metrics;
	size_t dropped;
	size_t ticks;
	struct _server *next;
} server;

#define BATCH_SIZE   2500
#define QUEUE_SIZE  25000

static server *servers = NULL;

/**
 * Reads from the queue and sends items to the remote server.  This
 * function is designed to be a thread.  Data sending is attempted to be
 * batched, but sent one by one to reduce loss on sending failure.
 * A connection with the server is maintained for as long as there is
 * data to be written.  As soon as there is none, the connection is
 * dropped.
 */
static void *
server_queuereader(void *d)
{
	server *self = (server *)d;
	size_t qlen;
	size_t len;
	ssize_t slen;
	const char *metrics[BATCH_SIZE + 1];
	const char **metric = metrics;
	struct timeval start, stop;
	struct timeval timeout;
	char nowbuf[24];
	queue *queue;

	*metric = NULL;
	self->metrics = 0;
	self->ticks = 0;

	timeout.tv_sec = 0;
	timeout.tv_usec = 650 * 1000;  /* less than 1s (dispatcher stall timeout) */

	self->running = 1;
	while (1) {
		if (self->failure) {
			if (!self->keep_running) {
				qlen = self->type == ORIGIN ? queue_len(self->queue) : 0;
				if (qlen > 0)
					fprintf(stderr, "dropping %zd metrics for %s:%u\n",
							qlen, self->ip, self->port);
				break;
			}
			/* if there was a failure last round, wait a bit for the server
			 * to recover */
			usleep(250 * 1000);  /* 250ms */
		}

		/* if we have work to do, try to do it */
		queue = self->queue;
		if ((qlen = queue_len(queue)) == 0) {
			/* if there are secondaries, check if they are in a
			 * condition where they could use some help */
			queue = NULL;
			for (len = 0; len < self->secondariescnt; len++) {
				if (self->secondaries[len] == self)
					continue;
				/* need help condition:
				 * - server is in failure mode, or
				 * - the queue is filled for > 3/4th (remember, our
				 *   queue is empty) */
				if (self->secondaries[len]->failure | /* ensure qlen */
						queue_free(self->secondaries[len]->queue) <
						(qlen = queue_len(self->secondaries[len]->queue)) / 4)
				{
					queue = self->secondaries[len]->queue;
					break;
				}
			}
			if (qlen == 0) {
				/* if we're idling, close the connection, this allows us
				 * to reduce connections, while keeping the connection alive
				 * if we're writing a lot */
				if (self->fd >= 0) {
					close(self->fd);
					self->fd = -1;
				}
				if (!self->keep_running)
					break;
				usleep(250 * 1000);  /* 250ms */
				/* it makes no sense to try and do something, so skip */
				continue;
			}
		}

		/* at this point we've got work to do, if we're instructed to
		 * shut down, however, try to get everything out of the door<commit_msg>server_queuereader: add parenthesis to pacify compiler<commit_after>/*
 *  This file is part of carbon-c-relay.
 *
 *  carbon-c-relay is free software: you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation, either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  carbon-c-relay is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with carbon-c-relay.  If not, see <http://www.gnu.org/licenses/>.
 */


#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <errno.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <sys/resource.h>

#include "relay.h"
#include "queue.h"
#include "dispatcher.h"
#include "collector.h"

enum servertype {
	ORIGIN,
	BACKUP,
	INTERNAL
};

typedef struct _server {
	const char *ip;
	unsigned short port;
	struct sockaddr_in serv_addr;
	int fd;
	queue *queue;
	enum servertype type;
	pthread_t tid;
	struct _server **secondaries;
	size_t secondariescnt;
	char failure:1;
	char running:1;
	char keep_running:1;
	size_t metrics;
	size_t dropped;
	size_t ticks;
	struct _server *next;
} server;

#define BATCH_SIZE   2500
#define QUEUE_SIZE  25000

static server *servers = NULL;

/**
 * Reads from the queue and sends items to the remote server.  This
 * function is designed to be a thread.  Data sending is attempted to be
 * batched, but sent one by one to reduce loss on sending failure.
 * A connection with the server is maintained for as long as there is
 * data to be written.  As soon as there is none, the connection is
 * dropped.
 */
static void *
server_queuereader(void *d)
{
	server *self = (server *)d;
	size_t qlen;
	size_t len;
	ssize_t slen;
	const char *metrics[BATCH_SIZE + 1];
	const char **metric = metrics;
	struct timeval start, stop;
	struct timeval timeout;
	char nowbuf[24];
	queue *queue;

	*metric = NULL;
	self->metrics = 0;
	self->ticks = 0;

	timeout.tv_sec = 0;
	timeout.tv_usec = 650 * 1000;  /* less than 1s (dispatcher stall timeout) */

	self->running = 1;
	while (1) {
		if (self->failure) {
			if (!self->keep_running) {
				qlen = self->type == ORIGIN ? queue_len(self->queue) : 0;
				if (qlen > 0)
					fprintf(stderr, "dropping %zd metrics for %s:%u\n",
							qlen, self->ip, self->port);
				break;
			}
			/* if there was a failure last round, wait a bit for the server
			 * to recover */
			usleep(250 * 1000);  /* 250ms */
		}

		/* if we have work to do, try to do it */
		queue = self->queue;
		if ((qlen = queue_len(queue)) == 0) {
			/* if there are secondaries, check if they are in a
			 * condition where they could use some help */
			queue = NULL;
			for (len = 0; len < self->secondariescnt; len++) {
				if (self->secondaries[len] == self)
					continue;
				/* need help condition:
				 * - server is in failure mode, or
				 * - the queue is filled for > 3/4th (remember, our
				 *   queue is empty) */
				if (self->secondaries[len]->failure | /* ensure qlen */
						(queue_free(self->secondaries[len]->queue) <
						 (qlen = queue_len(self->secondaries[len]->queue)) / 4))
				{
					queue = self->secondaries[len]->queue;
					break;
				}
			}
			if (qlen == 0) {
				/* if we're idling, close the connection, this allows us
				 * to reduce connections, while keeping the connection alive
				 * if we're writing a lot */
				if (self->fd >= 0) {
					close(self->fd);
					self->fd = -1;
				}
				if (!self->keep_running)
					break;
				usleep(250 * 1000);  /* 250ms */
				/* it makes no sense to try and do something, so skip */
				continue;
			}
		}

		/* at this point we've got work to do, if we're instructed to
		 * shut down, however, try to get everything out of the door
{'fraction_non_alphanumeric': 0.11584305934939161, 'fraction_numerical': 0.014651105040973429, 'mean_word_length': 3.041645760160562, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6374131', 'n_tokens_mistral': 3065, 'n_tokens_neox': 2643, 'n_words': 992}