diff options
Diffstat (limited to 'midimonster.c')
-rw-r--r-- | midimonster.c | 349 |
1 files changed, 209 insertions, 140 deletions
diff --git a/midimonster.c b/midimonster.c index b8594b4..b418711 100644 --- a/midimonster.c +++ b/midimonster.c @@ -9,6 +9,8 @@ #else #define MM_API __attribute__((dllexport)) #endif +#define BACKEND_NAME "core" +#define MM_SWAP_LIMIT 20 #include "midimonster.h" #include "config.h" #include "backend.h" @@ -28,25 +30,34 @@ typedef struct /*_mm_channel_mapping*/ { channel** to; } channel_mapping; -static size_t mappings = 0; -static channel_mapping* map = NULL; +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 fds = 0; static managed_fd* fd = NULL; static volatile sig_atomic_t fd_set_dirty = 1; static uint64_t global_timestamp = 0; -static event_collection event_pool[2] = { - {0}, - {0} -}; -static event_collection* primary = event_pool; - volatile static sig_atomic_t shutdown_requested = 0; static void signal_handler(int signum){ shutdown_requested = 1; } +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; +} + MM_API uint64_t mm_timestamp(){ return global_timestamp; } @@ -66,53 +77,66 @@ static void update_timestamp(){ } int mm_map_channel(channel* from, channel* to){ - size_t u, m; + size_t u, m, bucket = routing_hash(from); + //find existing source mapping - for(u = 0; u < mappings; u++){ - if(map[u].from == from){ + for(u = 0; u < routing.entries[bucket]; u++){ + if(routing.map[bucket][u].from == from){ break; } } //create new entry - if(u == mappings){ - map = realloc(map, (mappings + 1) * sizeof(channel_mapping)); - if(!map){ + 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(map + mappings, 0, sizeof(channel_mapping)); - mappings++; - map[u].from = from; + + 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 < map[u].destinations; m++){ - if(map[u].to[m] == to){ + for(m = 0; m < routing.map[bucket][u].destinations; m++){ + if(routing.map[bucket][u].to[m] == to){ return 0; } } - map[u].to = realloc(map[u].to, (map[u].destinations + 1) * sizeof(channel*)); - if(!map[u].to){ + //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"); - map[u].destinations = 0; + routing.map[bucket][u].destinations = 0; return 1; } - map[u].to[map[u].destinations] = to; - map[u].destinations++; + routing.map[bucket][u].to[routing.map[bucket][u].destinations] = to; + routing.map[bucket][u].destinations++; return 0; } -static void map_free(){ - size_t u; - for(u = 0; u < mappings; u++){ - free(map[u].to); +static 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; } - free(map); - mappings = 0; - map = NULL; } MM_API int mm_manage_fd(int new_fd, char* back, int manage, void* impl){ @@ -169,7 +193,6 @@ MM_API int mm_manage_fd(int new_fd, char* back, int manage, void* impl){ static void fds_free(){ size_t u; for(u = 0; u < fds; u++){ - //TODO free impl if(fd[u].fd >= 0){ close(fd[u].fd); fd[u].fd = -1; @@ -181,56 +204,46 @@ static void fds_free(){ } MM_API int mm_channel_event(channel* c, channel_value v){ - size_t u, p; + size_t u, p, bucket = routing_hash(c); //find mapped channels - for(u = 0; u < mappings; u++){ - if(map[u].from == c){ + for(u = 0; u < routing.entries[bucket]; u++){ + if(routing.map[bucket][u].from == c){ break; } } - if(u == mappings){ + if(u == routing.entries[bucket]){ //target-only channel return 0; } //resize event structures to fit additional events - if(primary->n + map[u].destinations >= primary->alloc){ - primary->channel = realloc(primary->channel, (primary->alloc + map[u].destinations) * sizeof(channel*)); - primary->value = realloc(primary->value, (primary->alloc + map[u].destinations) * sizeof(channel_value)); + 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(!primary->channel || !primary->value){ + if(!routing.events->channel || !routing.events->value){ fprintf(stderr, "Failed to allocate memory\n"); - primary->alloc = 0; - primary->n = 0; + routing.events->alloc = 0; + routing.events->n = 0; return 1; } - primary->alloc += map[u].destinations; + 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 - for(p = 0; p < map[u].destinations; p++){ - primary->channel[primary->n + p] = map[u].to[p]; - primary->value[primary->n + p] = v; + 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; } - primary->n += map[u].destinations; + routing.events->n += routing.map[bucket][u].destinations; return 0; } -static void event_free(){ - size_t u; - - for(u = 0; u < sizeof(event_pool) / sizeof(event_collection); u++){ - free(event_pool[u].channel); - free(event_pool[u].value); - event_pool[u].alloc = 0; - } -} - static void version(){ printf("MIDIMonster %s\n", MIDIMONSTER_VERSION); } @@ -271,6 +284,9 @@ static int platform_initialize(){ if(WSAStartup(version, &wsa)){ return 1; } + + unsigned error_mode = SetErrorMode(0); + SetErrorMode(error_mode | SEM_FAILCRITICALERRORS | SEM_NOGPFAULTERRORBOX | SEM_NOOPENFILEERRORBOX); #endif return 0; } @@ -325,67 +341,68 @@ static int args_parse(int argc, char** argv, char** cfg_file){ return 0; } -int main(int argc, char** argv){ - fd_set all_fds, read_fds; +static int core_process(size_t nfds, managed_fd* signaled_fds){ event_collection* secondary = NULL; - struct timeval tv; - size_t u, n; - managed_fd* signaled_fds = NULL; - int rv = EXIT_FAILURE, error, maxfd = -1; - char* cfg_file = DEFAULT_CFG; - #ifdef _WIN32 - char* error_message = NULL; - #endif + size_t u, swaps = 0; - //parse commandline arguments - if(args_parse(argc, argv, &cfg_file)){ - return EXIT_FAILURE; + //run backend processing, collect events + DBGPF("%lu backend FDs signaled\n", nfds); + if(backends_handle(nfds, signaled_fds)){ + return 1; } - if(platform_initialize()){ - fprintf(stderr, "Failed to perform platform-specific initialization\n"); - return EXIT_FAILURE; - } + //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, %lu events in primary\n", 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; + } + } - FD_ZERO(&all_fds); - //initialize backends - if(plugins_load(PLUGINS)){ - fprintf(stderr, "Failed to initialize a backend\n"); - goto bail; + //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; } - //read config - if(config_read(cfg_file)){ - fprintf(stderr, "Failed to read configuration file %s\n", cfg_file); - backends_stop(); - channels_free(); - instances_free(); - map_free(); - fds_free(); - plugins_close(); - config_free(); - return (usage(argv[0]) | platform_shutdown()); + if(swaps == MM_SWAP_LIMIT){ + LOG("Iteration swap limit hit, a backend may be configured to route events in an infinite loop"); } - //load an initial timestamp - update_timestamp(); + return 0; +} - //start backends - if(backends_start()){ - goto bail; - } +static int core_loop(){ + fd_set all_fds, read_fds; + managed_fd* signaled_fds = NULL; + struct timeval tv; + int error, maxfd = -1; + size_t n, u; + #ifdef _WIN32 + char* error_message = NULL; + #else + struct timespec ts; + #endif - signal(SIGINT, signal_handler); + FD_ZERO(&all_fds); //process events while(!shutdown_requested){ //rebuild fd set if necessary - if(fd_set_dirty){ + if(fd_set_dirty || !signaled_fds){ all_fds = fds_collect(&maxfd); signaled_fds = realloc(signaled_fds, fds * sizeof(managed_fd)); if(!signaled_fds){ fprintf(stderr, "Failed to allocate memory\n"); - goto bail; + return 1; } fd_set_dirty = 0; } @@ -393,20 +410,38 @@ int main(int argc, char** argv){ //wait for & translate events read_fds = all_fds; tv = backend_timeout(); - error = select(maxfd + 1, &read_fds, NULL, NULL, &tv); - if(error < 0){ - #ifndef _WIN32 - fprintf(stderr, "select failed: %s\n", strerror(errno)); + + //check whether there are any fds active, windows does not like select() without descriptors + if(maxfd >= 0){ + error = select(maxfd + 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 + free(signaled_fds); + 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 - 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; + ts.tv_sec = tv.tv_sec; + ts.tv_nsec = tv.tv_usec * 1000; + nanosleep(&ts, NULL); #endif - break; } + //update this iteration's timestamp + update_timestamp(); + //find all signaled fds n = 0; for(u = 0; u < fds; u++){ @@ -416,47 +451,81 @@ int main(int argc, char** argv){ } } - //update this iteration's timestamp - update_timestamp(); - - //run backend processing, collect events - DBGPF("%lu backend FDs signaled\n", n); - if(backends_handle(n, signaled_fds)){ - goto bail; + //fetch and process events + if(core_process(n, signaled_fds)){ + free(signaled_fds); + return 1; } + } - while(primary->n){ - //swap primary and secondary event collectors - DBGPF("Swapping event collectors, %lu events in primary\n", primary->n); - for(u = 0; u < sizeof(event_pool) / sizeof(event_collection); u++){ - if(primary != event_pool + u){ - secondary = primary; - primary = event_pool + u; - break; - } - } + free(signaled_fds); + return 0; +} - //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"); - goto bail; - } +int main(int argc, char** argv){ + int rv = EXIT_FAILURE; + char* cfg_file = DEFAULT_CFG; + size_t u, n = 0, max = 0; - //reset the event count - secondary->n = 0; - } + //parse commandline arguments + if(args_parse(argc, argv, &cfg_file)){ + return EXIT_FAILURE; + } + + if(platform_initialize()){ + fprintf(stderr, "Failed to perform platform-specific initialization\n"); + return EXIT_FAILURE; + } + + //initialize backends + if(plugins_load(PLUGINS)){ + fprintf(stderr, "Failed to initialize a backend\n"); + goto bail; + } + + //read config + if(config_read(cfg_file)){ + fprintf(stderr, "Failed to parse master configuration file %s\n", cfg_file); + backends_stop(); + routing_cleanup(); + fds_free(); + plugins_close(); + config_free(); + return (usage(argv[0]) | platform_shutdown()); + } + + //load an initial timestamp + update_timestamp(); + + //start backends + if(backends_start()){ + goto bail; + } + + signal(SIGINT, signal_handler); + + //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); + + if(!fds){ + fprintf(stderr, "No descriptors registered for multiplexing\n"); + } + + //run the core loop + if(!core_loop()){ + rv = EXIT_SUCCESS; } - rv = EXIT_SUCCESS; bail: //free all data - free(signaled_fds); backends_stop(); - channels_free(); - instances_free(); - map_free(); + routing_cleanup(); fds_free(); - event_free(); plugins_close(); config_free(); platform_shutdown(); |