<commit_before>/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

/**
 * dmconfig library
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include <errno.h>
#include <assert.h>
#include <string.h>
#include <stdio.h>
#include <stdint.h>
#include <inttypes.h>
#include <stdlib.h>
#include <time.h>
#include <arpa/inet.h>

#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <poll.h>
#include <fcntl.h>
#include <ev.h>
#include <syslog.h>
#include <signal.h>
#include <sys/queue.h>

#include "debug.h"

#ifdef HAVE_TALLOC_TALLOC_H
# include <talloc/talloc.h>
#else
# include <talloc.h>
#endif

#include "dmmsg.h"
#include "codes.h"
#include "dmcontext.h"
#include "dmconfig.h"

#include "utils/logx.h"
#include "utils/binary.h"

/* support multiple event loops? */
#if EV_MULTIPLICITY
#define EV_P_UNUSED_ EV_P __attribute__((unused)),
#define SOCK_EV_(sock) (sock)->ev,
#else
#define EV_P_UNUSED_
#define SOCK_EV_(sock)
#endif

static void connectEvent(EV_P_ ev_io *w, int revents);
static void acceptEvent(EV_P_ ev_io *w, int revents);
static void writeEvent(EV_P_UNUSED_ ev_io *w, int revents);
static void readEvent(EV_P_UNUSED_ ev_io *w, int revents);

/* callbacks used by the blocking API automatically */

/* socket helper */
static void dm_stop_events(DMCONTEXT *sock);
static void dm_free_requests(DMCONTEXT *sock, DMCONFIG_EVENT event);

/* process functions */
static void process_request(DMCONTEXT *socket, DM_PACKET *pkt);
static void process_reply(DMCONTEXT *socket, DM_PACKET *pkt);

/* global variables */

static uint32_t hopid = 0;
static uint32_t endid = 0;

#define CALLBACK(WHAT, ...)				\
	do {						\
		if (WHAT)				\
			(WHAT)(__VA_ARGS__);		\
	} while (0)

#define CALLBACK_RC(WHAT, ...)				\
	({						\
		uint32_t _rc = RC_OK;			\
		if (WHAT)				\
			_rc = (WHAT)(__VA_ARGS__);	\
		_rc;					\
	})
/** @private */
static void
connection_error(DMCONTEXT *socket, DMCONFIG_EVENT event)
{
	CALLBACK_RC(socket->connection_cb, event, socket, socket->userdata);
	dm_context_shutdown(socket, event);
}

/** shut down the socket in a socket context
 *
 * @param [in] dmCtx          Pointer to socket context to work on
 *
 * @ingroup API
 */
void
dm_context_shutdown(DMCONTEXT *sock, DMCONFIG_EVENT event)
{
	dm_stop_events(sock);
	dm_free_requests(sock, event);

	shutdown(sock->socket, SHUT_RDWR);
	close(sock->socket);
}


/** @private  put a packet into the send queue
 */
uint32_t dm_enqueue(DMCONTEXT *socket, DM2_REQUEST *req, int flags, DMRESULT_CB cb, void *data)
{
	DM2_REQUEST_INFO *rqi;

	if ((flags | REPLY) != REPLY) {
		/* initialize static hopid */
		switch (hopid) {		/* one never knows... */
		case 0:		srand((unsigned int)time(NULL));
		case MAX_INT:	hopid = endid = (float)rand()/RAND_MAX * (MAX_INT-1) + 1;
			break;
		default:	hopid = ++endid;
		}

		req->packet->hop2hop_id = htonl(hopid);
		req->packet->end2end_id = htonl(endid);
	}

	if (!(rqi = talloc_zero(socket, DM2_REQUEST_INFO)))<commit_msg>add workaround for libev timeout issue<commit_after>/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

/**
 * dmconfig library
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include <errno.h>
#include <assert.h>
#include <string.h>
#include <stdio.h>
#include <stdint.h>
#include <inttypes.h>
#include <stdlib.h>
#include <time.h>
#include <arpa/inet.h>

#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <poll.h>
#include <fcntl.h>
#include <ev.h>
#include <syslog.h>
#include <signal.h>
#include <sys/queue.h>

#include "debug.h"

#ifdef HAVE_TALLOC_TALLOC_H
# include <talloc/talloc.h>
#else
# include <talloc.h>
#endif

#include "dmmsg.h"
#include "codes.h"
#include "dmcontext.h"
#include "dmconfig.h"

#include "utils/logx.h"
#include "utils/binary.h"

/* support multiple event loops? */
#if EV_MULTIPLICITY
#define EV_P_UNUSED_ EV_P __attribute__((unused)),
#define SOCK_EV_(sock) (sock)->ev,
#else
#define EV_P_UNUSED_
#define SOCK_EV_(sock)
#endif

static void connectEvent(EV_P_ ev_io *w, int revents);
static void acceptEvent(EV_P_ ev_io *w, int revents);
static void writeEvent(EV_P_UNUSED_ ev_io *w, int revents);
static void readEvent(EV_P_UNUSED_ ev_io *w, int revents);

/* callbacks used by the blocking API automatically */

/* socket helper */
static void dm_stop_events(DMCONTEXT *sock);
static void dm_free_requests(DMCONTEXT *sock, DMCONFIG_EVENT event);

/* process functions */
static void process_request(DMCONTEXT *socket, DM_PACKET *pkt);
static void process_reply(DMCONTEXT *socket, DM_PACKET *pkt);

/* global variables */

static uint32_t hopid = 0;
static uint32_t endid = 0;

#define CALLBACK(WHAT, ...)				\
	do {						\
		if (WHAT)				\
			(WHAT)(__VA_ARGS__);		\
	} while (0)

#define CALLBACK_RC(WHAT, ...)				\
	({						\
		uint32_t _rc = RC_OK;			\
		if (WHAT)				\
			_rc = (WHAT)(__VA_ARGS__);	\
		_rc;					\
	})
/** @private */
static void
connection_error(DMCONTEXT *socket, DMCONFIG_EVENT event)
{
	CALLBACK_RC(socket->connection_cb, event, socket, socket->userdata);
	dm_context_shutdown(socket, event);
}

/** shut down the socket in a socket context
 *
 * @param [in] dmCtx          Pointer to socket context to work on
 *
 * @ingroup API
 */
void
dm_context_shutdown(DMCONTEXT *sock, DMCONFIG_EVENT event)
{
	dm_stop_events(sock);
	dm_free_requests(sock, event);

	shutdown(sock->socket, SHUT_RDWR);
	close(sock->socket);
}


/** @private  put a packet into the send queue
 */
uint32_t dm_enqueue(DMCONTEXT *socket, DM2_REQUEST *req, int flags, DMRESULT_CB cb, void *data)
{
	DM2_REQUEST_INFO *rqi;

	/* workaround: when the consumer is not keeping the event loop alive,
	 * i.e. using another event loop and calling only the synced methods,
	 */
	ev_now_update(socket->ev);

	if ((flags | REPLY) != REPLY) {
		/* initialize static hopid */
		switch (hopid) {		/* one never knows... */
		case 0:		srand((unsigned int)time(NULL));
		case MAX_INT:	hopid = endid = (float)rand()/RAND_MAX * (MAX_INT-1) + 1;
			break;
		default:	hopid = ++endid;
		}

		req->packet->hop2hop_id = htonl(hopid);
		req->packet->end2end_id = htonl(endid);
	}

	if (!(rqi = talloc_zero(socket, DM2_REQUEST_INFO)))
{'fraction_non_alphanumeric': 0.14359051789219388, 'fraction_numerical': 0.006945492979012532, 'mean_word_length': 4.191222570532916, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 70, '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': '6165256', 'n_tokens_mistral': 2648, 'n_tokens_neox': 2538, 'n_words': 659}