Factor out explicit frontend API

8 files changed, 517 insertions, 4 deletions

diff --git a/core/backend.c b/core/backend.c
index 83121bd..263fda3 100644
--- a/core/backend.c
+++ b/core/backend.c
@@ -1,9 +1,10 @@
 #include <string.h>
 #ifndef _WIN32
-#define MM_API __attribute__((visibility ("default")))
+	#define MM_API __attribute__((visibility ("default")))
 #else
-#define MM_API __attribute__((dllexport))
+	#define MM_API __attribute__((dllexport))
 #endif
+
 #define BACKEND_NAME "core/be"
 #include "midimonster.h"
 #include "backend.h"
diff --git a/core/backend.h b/core/backend.h
index 46c6c3a..1f85424 100644
--- a/core/backend.h
+++ b/core/backend.h
@@ -10,7 +10,7 @@ int backends_start();
 int backends_stop();
 instance* mm_instance(backend* b);
 
-/* Backend API */
+/* Public backend API */
 MM_API channel* mm_channel(instance* inst, uint64_t ident, uint8_t create);
 MM_API void mm_channel_update(channel* chan, uint64_t ident);
 MM_API instance* mm_instance_find(char* name, uint64_t ident);
diff --git a/core/config.h b/core/config.h
index b96a866..80e8fff 100644
--- a/core/config.h
+++ b/core/config.h
@@ -49,6 +49,9 @@ typedef struct /*_mm_config_override*/ {
 	char* value;
 } config_override;
 
+/* Internal API */
+void config_free();
+
+/* Frontent API */
 int config_read(char* file);
 int config_add_override(override_type type, char* data);
-void config_free();
diff --git a/core/core.c b/core/core.c
new file mode 100644
index 0000000..cd097f8
--- /dev/null
+++ b/core/core.c
@@ -0,0 +1,257 @@
+#include <string.h>
+#include <signal.h>
+#include <time.h>
+#include <errno.h>
+#include <unistd.h>
+#ifndef _WIN32
+	#include <sys/select.h>
+	#define MM_API __attribute__((visibility ("default")))
+#else
+	#define MM_API __attribute__((dllexport))
+#endif
+
+#define BACKEND_NAME "core"
+#include "midimonster.h"
+#include "core.h"
+#include "backend.h"
+#include "routing.h"
+#include "plugin.h"
+#include "config.h"
+
+static size_t fds = 0;
+static int max_fd = -1;
+static managed_fd* fd = NULL;
+static managed_fd* signaled_fds = NULL;
+static volatile sig_atomic_t fd_set_dirty = 1;
+static uint64_t global_timestamp = 0;
+
+static fd_set all_fds;
+
+MM_API uint64_t mm_timestamp(){
+	return global_timestamp;
+}
+
+static void core_timestamp(){
+	#ifdef _WIN32
+	global_timestamp = GetTickCount();
+	#else
+	struct timespec current;
+	if(clock_gettime(CLOCK_MONOTONIC_COARSE, &current)){
+		fprintf(stderr, "Failed to update global timestamp, time-based processing for some backends may be impaired: %s\n", strerror(errno));
+		return;
+	}
+
+	global_timestamp = current.tv_sec * 1000 + current.tv_nsec / 1000000;
+	#endif
+}
+
+static fd_set core_collect(int* max_fd){
+	size_t u = 0;
+	fd_set rv_fds;
+
+	if(max_fd){
+		*max_fd = -1;
+	}
+
+	DBGPF("Building selector set from %" PRIsize_t " FDs registered to core", fds);
+	FD_ZERO(&rv_fds);
+	for(u = 0; u < fds; u++){
+		if(fd[u].fd >= 0){
+			FD_SET(fd[u].fd, &rv_fds);
+			if(max_fd){
+				*max_fd = max(*max_fd, fd[u].fd);
+			}
+		}
+	}
+
+	return rv_fds;
+}
+
+MM_API int mm_manage_fd(int new_fd, char* back, int manage, void* impl){
+	backend* b = backend_match(back);
+	size_t u;
+
+	if(!b){
+		fprintf(stderr, "Unknown backend %s registered for managed fd\n", back);
+		return 1;
+	}
+
+	//find exact match
+	for(u = 0; u < fds; u++){
+		if(fd[u].fd == new_fd && fd[u].backend == b){
+			fd[u].impl = impl;
+			if(!manage){
+				fd[u].fd = -1;
+				fd[u].backend = NULL;
+				fd[u].impl = NULL;
+				fd_set_dirty = 1;
+			}
+			return 0;
+		}
+	}
+
+	if(!manage){
+		return 0;
+	}
+
+	//find free slot
+	for(u = 0; u < fds; u++){
+		if(fd[u].fd < 0){
+			break;
+		}
+	}
+	//if necessary expand
+	if(u == fds){
+		fd = realloc(fd, (fds + 1) * sizeof(managed_fd));
+		if(!fd){
+			fprintf(stderr, "Failed to allocate memory\n");
+			return 1;
+		}
+
+		signaled_fds = realloc(signaled_fds, (fds + 1) * sizeof(managed_fd));
+		if(!signaled_fds){
+			fprintf(stderr, "Failed to allocate memory\n");
+			return 1;
+		}
+		fds++;
+	}
+
+	//store new fd
+	fd[u].fd = new_fd;
+	fd[u].backend = b;
+	fd[u].impl = impl;
+	fd_set_dirty = 1;
+	return 0;
+}
+
+int core_initialize(){
+	FD_ZERO(&all_fds);
+
+	//load initial timestamp
+	core_timestamp();
+	
+	#ifdef _WIN32
+	WSADATA wsa;
+	WORD version = MAKEWORD(2, 2);
+	if(WSAStartup(version, &wsa)){
+		return 1;
+	}
+	_fmode = _O_BINARY;
+	#endif
+
+	//attach plugins
+	if(plugins_load(PLUGINS)){
+		LOG("Failed to initialize a backend");
+		return 1;
+	}
+
+	return 0;
+}
+
+int core_start(){
+	if(backends_start()){
+		return 1;
+	}
+
+	routing_stats();
+
+	if(!fds){
+		LOG("No descriptors registered for multiplexing");
+	}
+
+	return 0;
+}
+
+int core_iteration(){
+	fd_set read_fds;
+	struct timeval tv;
+	int error;
+	size_t n, u;
+	#ifdef _WIN32
+	char* error_message = NULL;
+	#else
+	struct timespec ts;
+	#endif
+
+	//rebuild fd set if necessary
+	if(fd_set_dirty || !signaled_fds){
+		all_fds = core_collect(&max_fd);
+		fd_set_dirty = 0;
+	}
+
+	//wait for & translate events
+	read_fds = all_fds;
+	tv = backend_timeout();
+
+	//check whether there are any fds active, windows does not like select() without descriptors
+	if(max_fd >= 0){
+		error = select(max_fd + 1, &read_fds, NULL, NULL, &tv);
+		if(error < 0){
+			#ifndef _WIN32
+			fprintf(stderr, "select failed: %s\n", strerror(errno));
+			#else
+			FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS,
+					NULL, WSAGetLastError(), MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT), (LPTSTR) &error_message, 0, NULL);
+			fprintf(stderr, "select failed: %s\n", error_message);
+			LocalFree(error_message);
+			error_message = NULL;
+			#endif
+			return 1;
+		}
+	}
+	else{
+		DBGPF("No descriptors, sleeping for %zu msec", tv.tv_sec * 1000 + tv.tv_usec / 1000);
+		#ifdef _WIN32
+		Sleep(tv.tv_sec * 1000 + tv.tv_usec / 1000);
+		#else
+		ts.tv_sec = tv.tv_sec;
+		ts.tv_nsec = tv.tv_usec * 1000;
+		nanosleep(&ts, NULL);
+		#endif
+	}
+
+	//update this iteration's timestamp
+	core_timestamp();
+
+	//find all signaled fds
+	n = 0;
+	for(u = 0; u < fds; u++){
+		if(fd[u].fd >= 0 && FD_ISSET(fd[u].fd, &read_fds)){
+			signaled_fds[n] = fd[u];
+			n++;
+		}
+	}
+
+	//run backend processing to collect events
+	DBGPF("%" PRIsize_t " backend FDs signaled", nfds);
+	if(backends_handle(n, signaled_fds)){
+		return 1;
+	}
+
+	//route generated events
+	return routing_iteration();
+}
+
+static void fds_free(){
+	size_t u;
+	for(u = 0; u < fds; u++){
+		if(fd[u].fd >= 0){
+			close(fd[u].fd);
+			fd[u].fd = -1;
+		}
+	}
+	free(fd);
+	fds = 0;
+	fd = NULL;
+}
+
+void core_shutdown(){
+	backends_stop();
+	routing_cleanup();
+	free(signaled_fds);
+	signaled_fds = NULL;
+	fds_free();
+	plugins_close();
+	config_free();
+	fd_set_dirty = 1;
+}
diff --git a/core/core.h b/core/core.h
new file mode 100644
index 0000000..3237a71
--- /dev/null
+++ b/core/core.h
@@ -0,0 +1,43 @@
+/* 
+ * MIDIMonster frontend API
+ * 
+ * These APIs expose the core as a linkable module. Frontends will use these calls
+ * as primary interface to interact with the MIDIMonster core.
+ *
+ * The lifecycle is as follows:
+ *
+ * 	* Initially, only the following API calls are valid:
+ * 			config_add_override()
+ * 			core_initialize()
+ * 		This allows the frontend to configure overrides for any configuration
+ * 		loaded later (e.g. by parsing command line arguments) before initializing
+ * 		the core.
+ * 	* Calling core_initialize() attaches all backend modules to the system and
+ * 		performs platform specific startup operations. From this point on,
+ * 		core_shutdown() must be called before terminating the frontend.
+ * 		All frontend API calls except `core_iteration` are now valid.
+ * 		The core is now in the configuration stage in which the frontend
+ * 		will push any configuration files.
+ * 	* Calling core_start() marks the transition from the configuration phase
+ * 		to the translation phase. The core will activate any configured backends
+ * 		and provide them with the information required to connect to their data
+ * 		sources and sinks. In this stage, only the following API calls are valid:
+ * 			core_iteration()
+ * 			core_shutdown()
+ * 	* The frontend will now repeatedly call core_iteration() to process any incoming
+ * 		events. This API will block execution until either one or more events have
+ * 		been registered or an internal timeout expires.
+ *	* Calling core_shutdown() releases all memory allocated by the core and any
+ *		attached modules or plugins, including all configuration, overrides,
+ *		mappings, statistics, etc. The core is now ready to exit or be
+ *		reinitialized using core_initialize().
+ */
+
+int core_initialize();
+int core_start();
+int core_iteration();
+void core_shutdown();
+
+/* Public backend API */
+MM_API uint64_t mm_timestamp();
+MM_API int mm_manage_fd(int new_fd, char* back, int manage, void* impl);
diff --git a/core/plugin.h b/core/plugin.h
index 64c557f..cdd544c 100644
--- a/core/plugin.h
+++ b/core/plugin.h
@@ -1,3 +1,5 @@
 typedef int (*plugin_init)();
+
+/* Internal API */
 int plugins_load(char* dir);
 int plugins_close();
diff --git a/core/routing.c b/core/routing.c
new file mode 100644
index 0000000..af93d0f
--- /dev/null
+++ b/core/routing.c
@@ -0,0 +1,198 @@
+#include <string.h>
+#include <signal.h>
+#include <time.h>
+#include <errno.h>
+#include <unistd.h>
+#ifndef _WIN32
+	#include <sys/select.h>
+	#define MM_API __attribute__((visibility ("default")))
+#else
+	#define MM_API __attribute__((dllexport))
+#endif
+
+#define BACKEND_NAME "core/rt"
+#define MM_SWAP_LIMIT 20
+#include "midimonster.h"
+#include "routing.h"
+#include "backend.h"
+
+/* Core-internal structures */
+typedef struct /*_event_collection*/ {
+	size_t alloc;
+	size_t n;
+	channel** channel;
+	channel_value* value;
+} event_collection;
+
+typedef struct /*_mm_channel_mapping*/ {
+	channel* from;
+	size_t destinations;
+	channel** to;
+} channel_mapping;
+
+static struct {
+	//routing_hash is set up for 256 buckets
+	size_t entries[256];
+	channel_mapping* map[256];
+
+	event_collection pool[2];
+	event_collection* events;
+} routing = {
+	.events = routing.pool
+};
+
+static size_t routing_hash(channel* key){
+	uint64_t repr = (uint64_t) key;
+	//return 8bit hash for 256 buckets, not ideal but it works
+	return (repr ^ (repr >> 8) ^ (repr >> 16) ^ (repr >> 24) ^ (repr >> 32)) & 0xFF;
+}
+
+int mm_map_channel(channel* from, channel* to){
+	size_t u, m, bucket = routing_hash(from);
+
+	//find existing source mapping
+	for(u = 0; u < routing.entries[bucket]; u++){
+		if(routing.map[bucket][u].from == from){
+			break;
+		}
+	}
+
+	//create new entry
+	if(u == routing.entries[bucket]){
+		routing.map[bucket] = realloc(routing.map[bucket], (routing.entries[bucket] + 1) * sizeof(channel_mapping));
+		if(!routing.map[bucket]){
+			routing.entries[bucket] = 0;
+			fprintf(stderr, "Failed to allocate memory\n");
+			return 1;
+		}
+
+		memset(routing.map[bucket] + routing.entries[bucket], 0, sizeof(channel_mapping));
+		routing.entries[bucket]++;
+		routing.map[bucket][u].from = from;
+	}
+
+	//check whether the target is already mapped
+	for(m = 0; m < routing.map[bucket][u].destinations; m++){
+		if(routing.map[bucket][u].to[m] == to){
+			return 0;
+		}
+	}
+
+	//add a mapping target
+	routing.map[bucket][u].to = realloc(routing.map[bucket][u].to, (routing.map[bucket][u].destinations + 1) * sizeof(channel*));
+	if(!routing.map[bucket][u].to){
+		fprintf(stderr, "Failed to allocate memory\n");
+		routing.map[bucket][u].destinations = 0;
+		return 1;
+	}
+
+	routing.map[bucket][u].to[routing.map[bucket][u].destinations] = to;
+	routing.map[bucket][u].destinations++;
+	return 0;
+}
+
+MM_API int mm_channel_event(channel* c, channel_value v){
+	size_t u, p, bucket = routing_hash(c);
+
+	//find mapped channels
+	for(u = 0; u < routing.entries[bucket]; u++){
+		if(routing.map[bucket][u].from == c){
+			break;
+		}
+	}
+
+	if(u == routing.entries[bucket]){
+		//target-only channel
+		return 0;
+	}
+
+	//resize event structures to fit additional events
+	if(routing.events->n + routing.map[bucket][u].destinations >= routing.events->alloc){
+		routing.events->channel = realloc(routing.events->channel, (routing.events->alloc + routing.map[bucket][u].destinations) * sizeof(channel*));
+		routing.events->value = realloc(routing.events->value, (routing.events->alloc + routing.map[bucket][u].destinations) * sizeof(channel_value));
+
+		if(!routing.events->channel || !routing.events->value){
+			fprintf(stderr, "Failed to allocate memory\n");
+			routing.events->alloc = 0;
+			routing.events->n = 0;
+			return 1;
+		}
+
+		routing.events->alloc += routing.map[bucket][u].destinations;
+	}
+
+	//enqueue channel events
+	//FIXME this might lead to one channel being mentioned multiple times in an apply call
+	memcpy(routing.events->channel + routing.events->n, routing.map[bucket][u].to, routing.map[bucket][u].destinations * sizeof(channel*));
+	for(p = 0; p < routing.map[bucket][u].destinations; p++){
+		routing.events->value[routing.events->n + p] = v;
+	}
+
+	routing.events->n += routing.map[bucket][u].destinations;
+	return 0;
+}
+
+void routing_stats(){
+	size_t n = 0, u, max = 0;
+
+	//count and report mappings
+	for(u = 0; u < sizeof(routing.map) / sizeof(routing.map[0]); u++){
+		n += routing.entries[u];
+		max = max(max, routing.entries[u]);
+	}
+
+	LOGPF("Routing %" PRIsize_t " sources, largest bucket has %" PRIsize_t " entries",
+			n, max);
+}
+
+int routing_iteration(){
+	event_collection* secondary = NULL;
+	size_t u, swaps = 0;
+
+	//limit number of collector swaps per iteration to prevent complete deadlock
+	while(routing.events->n && swaps < MM_SWAP_LIMIT){
+		//swap primary and secondary event collectors
+		DBGPF("Swapping event collectors, %" PRIsize_t " events in primary", routing.events->n);
+		for(u = 0; u < sizeof(routing.pool) / sizeof(routing.pool[0]); u++){
+			if(routing.events != routing.pool + u){
+				secondary = routing.events;
+				routing.events = routing.pool + u;
+				break;
+			}
+		}
+
+		//push collected events to target backends
+		if(secondary->n && backends_notify(secondary->n, secondary->channel, secondary->value)){
+			fprintf(stderr, "Backends failed to handle output\n");
+			return 1;
+		}
+
+		//reset the event count
+		secondary->n = 0;
+	}
+
+	if(swaps == MM_SWAP_LIMIT){
+		LOG("Iteration swap limit hit, a backend may be configured to route events in an infinite loop");
+	}
+
+	return 0;
+}
+
+void routing_cleanup(){
+	size_t u, n;
+
+	for(u = 0; u < sizeof(routing.map) / sizeof(routing.map[0]); u++){
+		for(n = 0; n < routing.entries[u]; n++){
+			free(routing.map[u][n].to);
+		}
+		free(routing.map[u]);
+		routing.map[u] = NULL;
+		routing.entries[u] = 0;
+	}
+
+	for(u = 0; u < sizeof(routing.pool) / sizeof(routing.pool[0]); u++){
+		free(routing.pool[u].channel);
+		free(routing.pool[u].value);
+		routing.pool[u].alloc = 0;
+	}
+}
diff --git a/core/routing.h b/core/routing.h
new file mode 100644
index 0000000..72dd768
--- /dev/null
+++ b/core/routing.h
@@ -0,0 +1,9 @@
+/* Internal API */
+int mm_map_channel(channel* from, channel* to);
+int routing_iteration();
+void routing_stats();
+void routing_cleanup();
+
+/* Public backend API */
+MM_API int mm_channel_event(channel* c, channel_value v);
+