odissey: implement relay pool

odissey.conf

@@ -23,7 +23,7 @@ odissey {
 #		tls_key_file ""
 #		tls_ca_file ""
 #		tls_protocols ""
-		workers 8
+		workers 1
 		client_max 100
 	}
 

src/CMakeLists.txt

@@ -16,6 +16,7 @@ set(od_src
 	od_pooler.c
 	od_router.c
 	od_relay.c
+	od_relay_pool.c
     od_frontend.c
     od_backend.c
     od_auth.c

src/od_instance.c

@@ -41,8 +41,9 @@
 
 #include "od_pooler.h"
 #include "od_router.h"
-#include "od_frontend.h"
 #include "od_relay.h"
+#include "od_relay_pool.h"
+#include "od_frontend.h"
 
 void od_instance_init(od_instance_t *instance)
 {
@@ -143,13 +144,13 @@ int od_instance_main(od_instance_t *instance, int argc, char **argv)
 	/* run system services */
 	od_pooler_t pooler;
 	od_router_t router;
-	od_relay_t  relay;
+	od_relaypool_t relay_pool;
 
 	od_system_t system = {
-		.pooler   = &pooler,
-		.router   = &router,
-		.relay    = &relay,
-		.instance = instance
+		.pooler     = &pooler,
+		.router     = &router,
+		.relay_pool = &relay_pool,
+		.instance   = instance
 	};
 	system.task_queue = machine_queue_create();
 	if (system.task_queue == NULL) {
@@ -167,8 +168,10 @@ int od_instance_main(od_instance_t *instance, int argc, char **argv)
 	if (rc == -1)
 		return 1;
 
-	od_relay_init(&relay, &system);
-	rc = od_relay_start(&relay);
+	rc = od_relaypool_init(&relay_pool, &system, instance->scheme.workers);
+	if (rc == -1)
+		return 1;
+	rc = od_relaypool_start(&relay_pool);
 	if (rc == -1)
 		return 1;
 

src/od_relay.c

@@ -47,7 +47,7 @@ od_relay(void *arg)
 	od_relay_t *relay = arg;
 	od_instance_t *instance = relay->system->instance;
 
-	od_log(&instance->log, "relay: started");
+	od_log(&instance->log, "relay %d: started", relay->id);
 
 	for (;;)
 	{
@@ -87,16 +87,19 @@ od_relay(void *arg)
 	od_log(&instance->log, "relay: stopped");
 }
 
-void od_relay_init(od_relay_t *relay, od_system_t *system)
+void od_relay_init(od_relay_t *relay, od_system_t *system, int id)
 {
 	relay->machine = -1;
+	relay->id = id;
 	relay->system = system;
 }
 
 int od_relay_start(od_relay_t *relay)
 {
 	od_instance_t *instance = relay->system->instance;
-	relay->machine = machine_create("relay", od_relay, relay);
+	char name[32];
+	snprintf(name, sizeof(name), "relay: %d", relay->id);
+	relay->machine = machine_create(name, od_relay, relay);
 	if (relay->machine == -1) {
 		od_error(&instance->log, "failed to start relay");
 		return 1;

src/od_relay.h

@@ -12,10 +12,11 @@ typedef struct od_relay od_relay_t;
 struct od_relay
 {
 	int64_t      machine;
+	int          id;
 	od_system_t *system;
 };
 
-void od_relay_init(od_relay_t*, od_system_t*);
+void od_relay_init(od_relay_t*, od_system_t*, int);
 int  od_relay_start(od_relay_t*);
 
 #endif /* OD_RELAY_H */

src/od_relay_pool.c

@@ -0,0 +1,70 @@
+
+/*
+ * odissey.
+ *
+ * PostgreSQL connection pooler and request router.
+*/
+
+#include <stdlib.h>
+#include <stdarg.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <string.h>
+#include <inttypes.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_pooler.h"
+#include "od_relay.h"
+#include "od_relay_pool.h"
+#include "od_frontend.h"
+
+int od_relaypool_init(od_relaypool_t *pool, od_system_t *system, int count)
+{
+	pool->count = count;
+	pool->pool = malloc(sizeof(od_relay_t) * count);
+	if (pool->pool == NULL)
+		return -1;
+	int i;
+	for (i = 0; i < count; i++) {
+		od_relay_t *relay = &pool->pool[i];
+		od_relay_init(relay, system, i);
+	}
+	return 0;
+}
+
+int od_relaypool_start(od_relaypool_t *pool)
+{
+	int i;
+	for (i = 0; i < pool->count; i++) {
+		od_relay_t *relay = &pool->pool[i];
+		int rc;
+		rc = od_relay_start(relay);
+		if (rc == -1)
+			return -1;
+	}
+	return 0;
+}

src/od_relay_pool.h

@@ -0,0 +1,21 @@
+#ifndef OD_RELAY_POOL_H
+#define OD_RELAY_POOL_H
+
+/*
+ * odissey.
+ *
+ * PostgreSQL connection pooler and request router.
+*/
+
+typedef struct od_relaypool od_relaypool_t;
+
+struct od_relaypool
+{
+	od_relay_t *pool;
+	int count;
+};
+
+int od_relaypool_init(od_relaypool_t*, od_system_t*, int);
+int od_relaypool_start(od_relaypool_t*);
+
+#endif /* OD_RELAY_POOL_H */

src/od_scheme.c

@@ -194,6 +194,12 @@ int od_scheme_validate(od_scheme_t *scheme, od_log_t *log)
 		return -1;
 	}
 
+	/* workers */
+	if (scheme->workers == 0) {
+		od_error(log, "bad workers number");
+		return -1;
+	}
+
 	/* routing mode */
 	if (scheme->routing == NULL) {
 		od_error(log, "routing mode is not set");

src/od_system.h

@@ -13,8 +13,8 @@ struct od_system
 {
 	void *instance;
 	void *pooler;
-	void *relay;
 	void *router;
+	void *relay_pool;
 	machine_queue_t task_queue;
 };