<commit_before>#include "pgrouter.h"
#include "proto.h"
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>

#define SUBSYS "worker"
#include "locks.inc.c"

#define max(a,b) ((a) > (b) ? (a) : (b))

#define  FSM_START          0
#define  FSM_SHUTDOWN       1
#define  FSM_STARTED        2
#define  FSM_AUTH_REQUIRED  3
#define  FSM_AUTH_SENT      4
#define  FSM_READY          5
#define  FSM_QUERY_SENT     6
#define  FSM_PARSE_SENT     7
#define  FSM_BIND_READY     8
#define  FSM_BIND_SENT      9
#define  FSM_EXEC_READY     10
#define  FSM_EXEC_SENT      11
#define  FSM_WAIT_SYNC      12
#define  FSM_SYNC_SENT      13

#define  ERR_BADPROTO  1
#define  ERR_CONNFAIL  2
#define  ERR_BEFAIL    3

static int connect_to_any_backend(CONTEXT *c, int *i, int *fd)
{
	int rc;

	rc = pgr_pick_any(c);
	if (rc < 0) {
		return 0;
	}
	*i = rc; /* index into backends */

	rc = rdlock(&c->lock, "context", 0);
	if (rc != 0) {
		return 0;
	}
	rc = rdlock(&c->backends[*i].lock, "backend", *i);
	if (rc != 0) {
		unlock(&c->lock, "context", 0);
		return 0;
	}

	char *host = strdup(c->backends[*i].hostname);
	int port = c->backends[*i].port;

	unlock(&c->backends[*i].lock, "backend", *i);
	unlock(&c->lock, "context", 0);

	rc = pgr_connect(host, port, 30); /* FIXME: timeout not honored; pick a better one! */
	if (rc < 0) {
		free(host);
		return 0;
	}
	*fd = rc;
	return 1;
}

static int wait_for_message_from(int fd, PG3_MSG *msg, int type)
{
	int rc;

	switch (type) {
	case 0:
		pgr_debugf("waiting to receive a message from fd %d", fd);
		break;

	case PG3_MSG_STARTUP:
		pgr_debugf("waiting to receive StartupMessage from fd %d", fd);
		break;

	default:
		pgr_debugf("waiting to receive message %d (%c) to fd %d",
				type, isprint(type) ? type : '.', fd);
		break;
	}

	/* FIXME: need a timeout version of pg3_recv */
	rc = pg3_recv(fd, msg, type);
	return rc == 0;
}

static int send_error_to(int fd, int error)
{
	int rc;
	PG3_MSG msg;
	PG3_ERROR err;
	memset(&msg, 0, sizeof(msg));
	memset(&err, 0, sizeof(err));

	switch (error) {
	case ERR_BADPROTO:
		err.severity = "FATAL";
		err.sqlstate = "08P01";
		err.message  = "Protocol violation";
		err.details  = "pgrouter encountered an invalid or inexpected protocol message";
		break;

	case ERR_CONNFAIL:
		err.severity = "FATAL";
		err.sqlstate = "08006";
		err.message  = "Backend connection failed";
		err.details  = "pgrouter was unable to connect to the desired backend database cluster";
		err.hint     = "Check the health of your pgrouter backends";
		break;

	case ERR_BEFAIL:
		err.severity = "FATAL";
		err.sqlstate = "08000";
		err.message  = "Failed to communicate to backend";
		err.details  = "pgrouter was unable to communicate with the connected backend database cluster";
		err.hint     = "Check the health of your pgrouter backends, and consult their logs";
		break;

	default:
		err.severity = "PANIC";
		err.sqlstate = "08000";
		err.message  = "An unrecognized error has occurred";
		err.details  = "pgrouter generated an error condition, but the type of error was not recognized.";
		err.hint     = "Please file a bug report; this is an internal pgrouter error";
		break;
	}

	pgr_debugf("sending error %d to fd %d: %s (%s) %s",
			error, fd, err.severity, err.sqlstate, err.message);

	rc = pg3_error(&msg, &err);
	if (rc != 0) {
		pgr_logf(stderr, LOG_ERR, "unable to send ErrorResponse to fd %d: %s (errno %d)",
				fd, strerror(errno), errno);
		return 0;
	}

	rc = pg3_send(fd, &msg);
	if (rc != 0) {
		pgr_logf(stderr, LOG_ERR, "unable to send ErrorResponse to fd %d: %s (errno %d)",
				fd, strerror(errno), errno);
		return 0;
	}

	return 1;
}

static int relay_msg_to(int fd, PG3_MSG *msg)
{
	int rc;

	rc = pg3_send(fd, msg);
	if (rc != 0) {
		pgr_logf(stderr, LOG_ERR, "unable to relay message to fd %d: %s (errno %d)",
				fd, strerror(errno), errno);
		return 0;
	}
	return 1;
}

static void handle_client(CONTEXT *c, int fefd)
{
	int state = FSM_START;
	int rc, backend, befd, port;
	char *host;
	PG3_MSG msg;

	for (;;) {
		switch (state) {
		case FSM_START:
			if (!wait_for_message_from(fefd, &msg, PG3_MSG_STARTUP)) {
				state = FSM_SHUTDOWN;
				break;
			}

			if (!connect_to_any_backend(c, &backend, &befd)) {
				send_error_to(fefd, ERR_CONNFAIL);
				state = FSM_SHUTDOWN;
				break;
			}

			state = FSM_STARTED;
			break;

		case FSM_SHUTDOWN:
			close(befd);
			close(fefd);
			return;
<commit_msg>Better debugging in worker<commit_after>#include "pgrouter.h"
#include "proto.h"
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>

#define SUBSYS "worker"
#include "locks.inc.c"

#define max(a,b) ((a) > (b) ? (a) : (b))

#define  FSM_START          0
#define  FSM_SHUTDOWN       1
#define  FSM_STARTED        2
#define  FSM_AUTH_REQUIRED  3
#define  FSM_AUTH_SENT      4
#define  FSM_READY          5
#define  FSM_QUERY_SENT     6
#define  FSM_PARSE_SENT     7
#define  FSM_BIND_READY     8
#define  FSM_BIND_SENT      9
#define  FSM_EXEC_READY     10
#define  FSM_EXEC_SENT      11
#define  FSM_WAIT_SYNC      12
#define  FSM_SYNC_SENT      13

static char* fsm_name(int n)
{
	switch (n) {
	case FSM_START:         return "START";
	case FSM_SHUTDOWN:      return "SHUTDOWN";
	case FSM_STARTED:       return "STARTED";
	case FSM_AUTH_REQUIRED: return "AUTH_REQUIRED";
	case FSM_AUTH_SENT:     return "AUTH_SENT";
	case FSM_READY:         return "READY";
	case FSM_QUERY_SENT:    return "QUERY_SENT";
	case FSM_PARSE_SENT:    return "PARSE_SENT";
	case FSM_BIND_READY:    return "BIND_READY";
	case FSM_BIND_SENT:     return "BIND_SENT";
	case FSM_EXEC_READY:    return "EXEC_READY";
	case FSM_EXEC_SENT:     return "EXEC_SENT";
	case FSM_WAIT_SYNC:     return "WAIT_SYNC";
	case FSM_SYNC_SENT:     return "SYNC_SENT";
	default:                return "(unknow)";
	}
}

#define  ERR_BADPROTO  1
#define  ERR_CONNFAIL  2
#define  ERR_BEFAIL    3

static int connect_to_any_backend(CONTEXT *c, int *i, int *fd)
{
	int rc;

	rc = pgr_pick_any(c);
	if (rc < 0) {
		return 0;
	}
	*i = rc; /* index into backends */

	rc = rdlock(&c->lock, "context", 0);
	if (rc != 0) {
		return 0;
	}
	rc = rdlock(&c->backends[*i].lock, "backend", *i);
	if (rc != 0) {
		unlock(&c->lock, "context", 0);
		return 0;
	}

	char *host = strdup(c->backends[*i].hostname);
	int port = c->backends[*i].port;

	unlock(&c->backends[*i].lock, "backend", *i);
	unlock(&c->lock, "context", 0);

	rc = pgr_connect(host, port, 30); /* FIXME: timeout not honored; pick a better one! */
	if (rc < 0) {
		free(host);
		return 0;
	}
	free(host);
	*fd = rc;
	return 1;
}

static int wait_for_message_from(int fd, PG3_MSG *msg, int type)
{
	int rc;

	switch (type) {
	case 0:
		pgr_debugf("waiting to receive a message from fd %d", fd);
		break;

	case PG3_MSG_STARTUP:
		pgr_debugf("waiting to receive StartupMessage from fd %d", fd);
		break;

	default:
		pgr_debugf("waiting to receive message %d (%c) to fd %d",
				type, isprint(type) ? type : '.', fd);
		break;
	}

	/* FIXME: need a timeout version of pg3_recv */
	rc = pg3_recv(fd, msg, type);
	return rc == 0;
}

static int send_error_to(int fd, int error)
{
	int rc;
	PG3_MSG msg;
	PG3_ERROR err;
	memset(&msg, 0, sizeof(msg));
	memset(&err, 0, sizeof(err));

	switch (error) {
	case ERR_BADPROTO:
		err.severity = "FATAL";
		err.sqlstate = "08P01";
		err.message  = "Protocol violation";
		err.details  = "pgrouter encountered an invalid or inexpected protocol message";
		break;

	case ERR_CONNFAIL:
		err.severity = "FATAL";
		err.sqlstate = "08006";
		err.message  = "Backend connection failed";
		err.details  = "pgrouter was unable to connect to the desired backend database cluster";
		err.hint     = "Check the health of your pgrouter backends";
		break;

	case ERR_BEFAIL:
		err.severity = "FATAL";
		err.sqlstate = "08000";
		err.message  = "Failed to communicate to backend";
		err.details  = "pgrouter was unable to communicate with the connected backend database cluster";
		err.hint     = "Check the health of your pgrouter backends, and consult their logs";
		break;

	default:
		err.severity = "PANIC";
		err.sqlstate = "08000";
		err.message  = "An unrecognized error has occurred";
		err.details  = "pgrouter generated an error condition, but the type of error was not recognized.";
		err.hint     = "Please file a bug report; this is an internal pgrouter error";
		break;
	}

	pgr_debugf("sending error %d to fd %d: %s (%s) %s",
			error, fd, err.severity, err.sqlstate, err.message);

	rc = pg3_error(&msg, &err);
	if (rc != 0) {
		pgr_logf(stderr, LOG_ERR, "unable to send ErrorResponse to fd %d: %s (errno %d)",
				fd, strerror(errno), errno);
		return 0;
	}

	rc = pg3_send(fd, &msg);
	if (rc != 0) {
		pgr_logf(stderr, LOG_ERR, "unable to send ErrorResponse to fd %d: %s (errno %d)",
				fd, strerror(errno), errno);
		return 0;
	}

	return 1;
}

static int relay_msg_to(int fd, PG3_MSG *msg)
{
	int rc;

	rc = pg3_send(fd, msg);
	if (rc != 0) {
		pgr_logf(stderr, LOG_ERR, "unable to relay message to fd %d: %s (errno %d)",
				fd, strerror(errno), errno);
		return 0;
	}
	return 1;
}

static void handle_client(CONTEXT *c, int fefd)
{
	int state = FSM_START;
	int rc, backend, befd, port;
	char *host;
	PG3_MSG msg;

	befd = backend = -1;
	pgr_debugf("not connected to a backend; fefd = %d, befd = %d",
			backend, fefd, befd);

	for (;;) {
		pgr_debugf("FSM[%s] backend = %d, fefd = %d, befd = %d",
				fsm_name(FSM_START), backend, fefd, befd);
		switch (state) {
		case FSM_START:
			if (!wait_for_message_from(fefd, &msg, PG3_MSG_STARTUP)) {
				state = FSM_SHUTDOWN;
				break;
			}

			if (!connect_to_any_backend(c, &backend, &befd)) {
				send_error_to(fefd, ERR_CONNFAIL);
				state = FSM_SHUTDOWN;
				break;
			}
			pgr_debugf("connected to backend/%d; fefd = %d, befd = %d",
					backend, fefd, befd);

			if (!relay_msg_to(befd, &msg)) {
				state = FSM_SHUTDOWN;
				break;
			}
			state = FSM_STARTED;
			break;

		case FSM_SHUTDOWN:
			if (befd >= 0) {
				close(befd);
			}
			close(fefd);
			return;

{'fraction_non_alphanumeric': 0.13564731697586277, 'fraction_numerical': 0.01575902653101935, 'mean_word_length': 3.4249779346866727, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 28, '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': '1943338', 'n_tokens_mistral': 4346, 'n_tokens_neox': 4009, 'n_words': 1118}