odissey: add basic backend cancel

src/od_backend.c

@@ -44,6 +44,7 @@
 #include "od_frontend.h"
 #include "od_backend.h"
 #include "od_auth.h"
+#include "od_cancel.h"
 
 void od_backend_close(od_server_t *server)
 {
@@ -435,12 +436,9 @@ int od_backend_reset(od_server_t *server)
 			         "S (reset): not responded, cancel (#%d)",
 			         wait_try_cancel);
 			wait_try_cancel++;
-			/* TODO: */
-			/*
-			rc = od_cancel_of(pooler, route->scheme->server, &server->key);
+			rc = od_cancel(instance, route->scheme->server, &server->key);
 			if (rc < 0)
 				goto error;
-			*/
 			continue;
 		}
 		assert(od_server_is_sync(server));

src/od_cancel.c

@@ -0,0 +1,146 @@
+
+/*
+ * odissey.
+ *
+ * PostgreSQL connection pooler and request router.
+*/
+
+#include <stdlib.h>
+#include <stdarg.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <string.h>
+#include <signal.h>
+
+#include <machinarium.h>
+#include <soprano.h>
+
+#include "od_macro.h"
+#include "od_version.h"
+#include "od_list.h"
+#include "od_pid.h"
+#include "od_syslog.h"
+#include "od_log.h"
+#include "od_daemon.h"
+#include "od_scheme.h"
+#include "od_lex.h"
+#include "od_config.h"
+#include "od_msg.h"
+#include "od_system.h"
+#include "od_instance.h"
+
+#include "od_server.h"
+#include "od_server_pool.h"
+#include "od_client.h"
+#include "od_client_pool.h"
+#include "od_route_id.h"
+#include "od_route.h"
+#include "od_route_pool.h"
+#include "od_io.h"
+
+#include "od_pooler.h"
+#include "od_router.h"
+#include "od_relay.h"
+#include "od_frontend.h"
+#include "od_backend.h"
+#include "od_auth.h"
+#include "od_cancel.h"
+
+int od_cancel(od_instance_t *instance,
+              od_schemeserver_t *server_scheme,
+              so_key_t *key)
+{
+	machine_io_t io = machine_io_create();
+	if (io == NULL)
+		return -1;
+
+	/* resolve server address */
+	char port[16];
+	snprintf(port, sizeof(port), "%d", server_scheme->port);
+	struct addrinfo *ai = NULL;
+	int rc;
+	rc = machine_getaddrinfo(server_scheme->host, port, NULL, &ai, 0);
+	if (rc < 0) {
+		od_error(&instance->log, NULL, "failed to resolve %s:%d",
+		         server_scheme->host,
+		         server_scheme->port);
+		return -1;
+	}
+	assert(ai != NULL);
+
+	machine_set_nodelay(io, instance->scheme.nodelay);
+	if (instance->scheme.keepalive > 0)
+		machine_set_keepalive(io, 1, instance->scheme.keepalive);
+
+	/* connect to server */
+	rc = machine_connect(io, ai->ai_addr, UINT32_MAX);
+	freeaddrinfo(ai);
+	if (rc < 0) {
+		od_error(&instance->log, NULL,
+		         "(cancel) failed to connect to %s:%d",
+		         server_scheme->host,
+		         server_scheme->port);
+		machine_close(io);
+		machine_io_free(io);
+		return -1;
+	}
+	rc = machine_set_readahead(io, instance->scheme.readahead);
+	if (rc == -1) {
+		od_error(&instance->log, NULL, "(cancel) failed to set readahead");
+		return -1;
+	}
+
+	so_stream_t stream;
+	so_stream_init(&stream);
+
+	/* handle tls connection */
+	machine_tls_t tls = NULL;
+	/* TODO */
+#if 0
+	if (server_scheme->tls_verify != OD_TDISABLE) {
+		tls = od_tlsbe(pooler->env, server_scheme);
+		if (tls == NULL) {
+			od_error(&pooler->od->log, NULL,
+			         "(cancel) failed to create tls context",
+			         server_scheme->host,
+			         server_scheme->port);
+			machine_close(io);
+			machine_free_io(io);
+			return -1;
+		}
+		rc = od_tlsbe_connect(pooler->env, io, tls,
+		                      &stream,
+		                      &pooler->od->log, "(cancel)",
+		                      server_scheme);
+		if (rc == -1) {
+			machine_close(io);
+			machine_free_io(io);
+			machine_free_tls(tls);
+			so_stream_free(&stream);
+			return -1;
+		}
+	}
+#endif
+
+	/* send cancel and disconnect */
+	rc = so_fewrite_cancel(&stream, key->key_pid, key->key);
+	if (rc == -1) {
+		machine_close(io);
+		machine_io_free(io);
+		if (tls)
+			machine_tls_free(tls);
+		so_stream_free(&stream);
+		return -1;
+	}
+	rc = od_write(io, &stream);
+	if (rc == -1) {
+		od_error(&instance->log, io, "(cancel): write error: %s",
+		         machine_error(io));
+	}
+	machine_close(io);
+	machine_io_free(io);
+	if (tls)
+		machine_tls_free(tls);
+	so_stream_free(&stream);
+	return 0;
+}

src/od_cancel.h

@@ -0,0 +1,12 @@
+#ifndef OD_CANCEL_H
+#define OD_CANCEL_H
+
+/*
+ * odissey.
+ *
+ * PostgreSQL connection pooler and request router.
+*/
+
+int od_cancel(od_instance_t*, od_schemeserver_t*, so_key_t*);
+
+#endif

src/od_frontend.c

@@ -374,13 +374,14 @@ void od_frontend(void *arg)
 	/* client cancel request */
 	if (client->startup.is_cancel) {
 		od_debug(&instance->log, client->io, "C: cancel request");
+#if 0
+		od_relay_t *relay = client->system->relay;
+		so_key_t key = client->startup.key;
+		od_frontend_close(client);
+		od_cancel(relay, &key);
+#endif
 
 		od_frontend_close(client);
-#if 0
-		so_key_t key = client->startup.key;
-		od_feclose(client);
-		od_cancel(pooler, &key);
-#endif
 		return;
 	}