added event.[ch]

Makefile

@@ -8,7 +8,7 @@ WITH_CURL = yes
 
 # ############################
 
-SAMMLER_SRC := sammler.c configfile.c helper.c logging.c network.c plugins.c
+SAMMLER_SRC := sammler.c configfile.c event.c helper.c logging.c network.c plugins.c
 PLUGIN_SRC := p_ctstat.c p_load.c p_memory.c p_mount.c p_netdev.c p_random.c
 PLUGIN_SRC += p_rtstat.c p_stat.c p_uptime.c p_vmstat.c
 CFLAGS := -O2 -Wall

event.c

@@ -0,0 +1,302 @@
+/***************************************************************************
+ *   Copyright (C) 10/2006 by Olaf Rempel                                  *
+ *   razzor@kopf-tisch.de                                                  *
+ *                                                                         *
+ *   This program is free software; you can redistribute it and/or modify  *
+ *   it under the terms of the GNU General Public License as published by  *
+ *   the Free Software Foundation; either version 2 of the License, or     *
+ *   (at your option) any later version.                                   *
+ *                                                                         *
+ *   This program is distributed in the hope that it will be useful,       *
+ *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
+ *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
+ *   GNU General Public License for more details.                          *
+ *                                                                         *
+ *   You should have received a copy of the GNU General Public License     *
+ *   along with this program; if not, write to the                         *
+ *   Free Software Foundation, Inc.,                                       *
+ *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
+ ***************************************************************************/
+#include <stdlib.h>
+#include <unistd.h>
+#include <string.h>
+
+#include <sys/time.h>
+#include <sys/types.h>
+
+#include "list.h"
+#include "logging.h"
+
+#include "event.h"
+
+static LIST_HEAD(event_fd_list);
+static LIST_HEAD(event_timeout_list);
+
+struct event_fd {
+	struct list_head list;
+	unsigned int flags;
+	int fd;
+	int (*read_cb)(int fd, void *privdata);
+	int (*write_cb)(int fd, void *privdata);
+	void *read_priv;
+	void *write_priv;
+};
+
+struct event_timeout {
+	struct list_head list;
+	unsigned int flags;
+	struct timeval intervall;
+	struct timeval nextrun;
+	int (*callback)(void *privdata);
+	void *privdata;
+};
+
+struct event_fd * event_add_fd(
+			struct event_fd *entry,
+			int fd,
+			int type,
+			int (*callback)(int fd, void *privdata),
+			void *privdata)
+{
+	/* check valid filediskriptor */
+	if (fd < 0 || fd > FD_SETSIZE) {
+		log_print(LOG_ERROR, "event_add_fd(): invalid fd");
+		return NULL;
+	}
+
+	/* check valid type (read/write) */
+	if (!(type & FD_TYPES)) {
+		log_print(LOG_ERROR, "event_add_fd(): invalid type");
+		return NULL;
+	}
+
+	/* create new entry */
+	if (entry == NULL) {
+		entry = malloc(sizeof(struct event_fd));
+		if (entry == NULL) {
+			log_print(LOG_ERROR, "event_add_fd(): out of memory");
+			return NULL;
+		}
+
+		memset(entry, 0, sizeof(struct event_fd));
+		entry->fd = fd;
+
+		/* put it on the list */
+		list_add_tail(&entry->list, &event_fd_list);
+	}
+
+	if (type & FD_READ) {
+		entry->flags = (callback != NULL) ? (entry->flags | FD_READ) : (entry->flags & ~FD_READ);
+		entry->read_cb = callback;
+		entry->read_priv = privdata;
+
+	} else if (type & FD_WRITE) {
+		entry->flags = (callback != NULL) ? (entry->flags | FD_WRITE) : (entry->flags & ~FD_WRITE);
+		entry->write_cb = callback;
+		entry->write_priv = privdata;
+	}
+
+	entry->flags |= EVENT_NEW;
+	return entry;
+}
+
+int event_get_fd(struct event_fd *entry)
+{
+	return (entry != NULL) ? entry->fd: -1;
+}
+
+void event_remove_fd(struct event_fd *entry)
+{
+	/* mark the event as deleted -> remove in select() loop */
+	entry->flags |= EVENT_DELETE;
+}
+
+static void add_timeval(struct timeval *ret, struct timeval *a, struct timeval *b)
+{
+	ret->tv_usec = a->tv_usec + b->tv_usec;
+	ret->tv_sec = a->tv_sec + b->tv_sec;
+
+	if (ret->tv_usec >= 1000000) {
+		ret->tv_usec -= 1000000;
+		ret->tv_sec++;
+	}
+}
+
+static void sub_timeval(struct timeval *ret, struct timeval *a, struct timeval *b)
+{
+	ret->tv_usec = a->tv_usec - b->tv_usec;
+	ret->tv_sec = a->tv_sec - b->tv_sec;
+
+	if (ret->tv_usec < 0) {
+		ret->tv_usec += 1000000;
+		ret->tv_sec--;
+	}
+}
+
+static int cmp_timeval(struct timeval *a, struct timeval *b)
+{
+	if (a->tv_sec > b->tv_sec)
+		return -1;
+
+	if (a->tv_sec < b->tv_sec)
+		return 1;
+
+	if (a->tv_usec > b->tv_usec)
+		return -1;
+
+	if (a->tv_usec < b->tv_usec)
+		return 1;
+
+	return 0;
+}
+
+static void schedule_nextrun(struct event_timeout *entry, struct timeval *now)
+{
+	add_timeval(&entry->nextrun, now, &entry->intervall);
+
+	struct event_timeout *search;
+	list_for_each_entry(search, &event_timeout_list, list) {
+		if (search->nextrun.tv_sec > entry->nextrun.tv_sec) {
+			list_add_tail(&entry->list, &search->list);
+			return;
+
+		} else if (search->nextrun.tv_sec == entry->nextrun.tv_sec &&
+			   search->nextrun.tv_usec > entry->nextrun.tv_usec) {
+				list_add_tail(&entry->list, &search->list);
+				return;
+		}
+	}
+	list_add_tail(&entry->list, &event_timeout_list);
+}
+
+struct event_timeout * event_add_timeout(
+			struct timeval *timeout,
+			int (*callback)(void *privdata),
+			void *privdata)
+{
+	struct event_timeout *entry;
+	entry = malloc(sizeof(struct event_timeout));
+	if (entry == NULL) {
+		log_print(LOG_ERROR, "event_add_timeout(): out of memory");
+		return NULL;
+	}
+
+	entry->flags = 0;
+	memcpy(&entry->intervall, timeout, sizeof(entry->intervall));
+	entry->callback = callback;
+	entry->privdata = privdata;
+
+	struct timeval now;
+	gettimeofday(&now, NULL);
+
+	schedule_nextrun(entry, &now);
+	return entry;
+}
+
+void event_remove_timeout(struct event_timeout *entry)
+{
+	/* mark the event as deleted -> remove in select() loop */
+	entry->flags |= EVENT_DELETE;
+}
+
+int event_loop(void)
+{
+	fd_set *fdsets = malloc(sizeof(fd_set) * 2);
+	if (fdsets == NULL) {
+		log_print(LOG_ERROR, "event_loop(): out of memory");
+		return -1;
+	}
+
+	while (1) {
+		fd_set *readfds = NULL, *writefds = NULL;
+		struct event_fd *entry, *tmp;
+
+		list_for_each_entry_safe(entry, tmp, &event_fd_list, list) {
+			entry->flags &= ~EVENT_NEW;
+
+			if (entry->flags & EVENT_DELETE) {
+				list_del(&entry->list);
+				free(entry);
+
+			} else if (entry->flags & FD_READ) {
+				if (readfds == NULL) {
+					readfds = &fdsets[0];
+					FD_ZERO(readfds);
+				}
+				FD_SET(entry->fd, readfds);
+
+			} else if (entry->flags & FD_WRITE) {
+				if (writefds == NULL) {
+					writefds = &fdsets[1];
+					FD_ZERO(writefds);
+				}
+				FD_SET(entry->fd, writefds);
+			}
+		}
+
+		struct timeval timeout, *timeout_p = NULL;
+		if (!list_empty(&event_timeout_list)) {
+			struct timeval now;
+			gettimeofday(&now, NULL);
+
+			struct event_timeout *entry, *tmp;
+			list_for_each_entry_safe(entry, tmp, &event_timeout_list, list) {
+				if (entry->flags & EVENT_DELETE) {
+					list_del(&entry->list);
+					free(entry);
+					continue;
+				}
+
+				/* timeout not elapsed, exit search (since list is sorted) */
+				if (cmp_timeval(&entry->nextrun, &now) == -1)
+					break;
+
+				/* remove event from list */
+				list_del(&entry->list);
+
+				/* execute callback, when callback returns 0 -> schedule event again */
+				if (entry->callback(entry->privdata)) {
+					free(entry);
+
+				} else {
+					schedule_nextrun(entry, &now);
+				}
+			}
+
+			if (!list_empty(&event_timeout_list)) {
+				entry = list_entry(event_timeout_list.next, typeof(*entry), list);
+
+				/* calc select() timeout */
+				sub_timeval(&timeout, &entry->nextrun, &now);
+				timeout_p = &timeout;
+			}
+		}
+
+		int i = select(FD_SETSIZE, readfds, writefds, NULL, timeout_p);
+		if (i <= 0) {
+			/* On error, -1 is returned, and errno is set
+			 * appropriately; the sets and timeout become
+			 * undefined, so do not rely on their contents
+			 * after an error.
+			 */
+			continue;
+
+		} else {
+			list_for_each_entry(entry, &event_fd_list, list) {
+				if ((entry->flags & EVENT_NEW) != 0) {
+					/* entry has just been added, execute it next round */
+					continue;
+				}
+
+				if ((entry->flags & FD_READ) && FD_ISSET(entry->fd, readfds))
+					if (entry->read_cb(entry->fd, entry->read_priv) != 0)
+						entry->flags |= EVENT_DELETE;
+
+				if ((entry->flags & FD_WRITE) && FD_ISSET(entry->fd, writefds))
+					if (entry->write_cb(entry->fd, entry->write_priv) != 0)
+						entry->flags |= EVENT_DELETE;
+			}
+		}
+	}
+	free(fdsets);
+}

event.h

@@ -0,0 +1,42 @@
+#ifndef _EVENT_H_
+#define _EVENT_H_
+
+#include <sys/time.h>
+
+#define EVENT_NEW	0x1000
+#define EVENT_DELETE	0x2000
+
+#define FD_READ		0x0001
+#define FD_WRITE	0x0002
+#define FD_TYPES	(FD_READ | FD_WRITE)
+
+#define event_add_readfd(entry, fd, callback, privdata) \
+	event_add_fd(entry, fd, FD_READ, callback, privdata)
+
+#define event_add_writefd(entry, fd, callback, privdata) \
+	event_add_fd(entry, fd, FD_WRITE, callback, privdata)
+
+/* inner details are not visible to external users (TODO: size unknown) */
+struct event_fd;
+struct event_timeout;
+
+struct event_fd * event_add_fd(
+			struct event_fd *entry,
+			int fd,
+			int type,
+			int (*callback)(int fd, void *privdata),
+			void *privdata);
+
+int event_get_fd(struct event_fd *entry);
+void event_remove_fd(struct event_fd *entry);
+
+struct event_timeout * event_add_timeout(
+			struct timeval *timeout,
+			int (*callback)(void *privdata),
+			void *privdata);
+
+void event_remove_timeout(struct event_timeout *entry);
+
+int event_loop(void);
+
+#endif /* _EVENT_H_ */

network.c

@@ -8,6 +8,7 @@
 #include <arpa/inet.h>
 
 #include "configfile.h"
+#include "event.h"
 #include "helper.h"
 #include "list.h"
 #include "logging.h"
@@ -17,8 +18,13 @@
 
 #define BUFSIZE 8192
 
-static struct sockaddr_in fwd_sa;
+struct fwd_addr {
+	struct list_head list;
+	struct sockaddr_in addr;
+};
+
 static int fwd_sock;
+static LIST_HEAD(fwd_list);
 
 // todo: never freed..
 static char *tx_buf, *rx_buf;
@@ -26,6 +32,9 @@ static int tx_pos;
 
 int net_submit(const char *hostname, const char *plugin, const char *filename, int ds_id, const char *data)
 {
+	if (list_empty(&fwd_list))
+		return 0;
+
 	int size = snprintf(tx_buf + tx_pos, BUFSIZE - tx_pos, "%s:%s:%s:%d %s\n",
 				hostname, plugin, filename, ds_id, data);
 
@@ -39,16 +48,19 @@ int net_submit(const char *hostname, const char *plugin, const char *filename, i
 	return 0;
 }
 
-int net_submit_flush(void)
+void net_submit_flush(void)
 {
-	if (tx_pos != 0)
+	if (tx_pos == 0)
-		sendto(fwd_sock, tx_buf, tx_pos, 0, (struct sockaddr *)&fwd_sa, sizeof(fwd_sa));
+		return;
+
+	struct fwd_addr *entry;
+	list_for_each_entry(entry, &fwd_list, list)
+		sendto(fwd_sock, tx_buf, tx_pos, 0, (struct sockaddr *)&entry->addr, sizeof(entry->addr));
 
 	tx_pos = 0;
-	return 0;
 }
 
-int net_receive(int socket)
+static int net_receive(int socket, void *privdata)
 {
 	int size = recv(socket, rx_buf, BUFSIZE, 0);
 	int rx_pos = 0;
@@ -76,75 +88,110 @@ int net_receive(int socket)
 	return 0;
 }
 
-static int net_config_get_saddr(const char *section, const char *option, struct sockaddr_in *sa)
+static int parse_saddr(const char *addr, struct sockaddr_in *sa)
 {
-	char *part[2];
+	char *addr_cpy = strdup(addr);
-
+	if (addr_cpy == NULL) {
-	const char *ret = config_get_string(section, option, NULL);
+		log_print(LOG_WARN, "parse_saddr(): out of memory");
-	if (ret == NULL)
 		return -1;
+	}
 
-	if (strsplit(ret, ":", part, 2) != 2)
+	char *tmp;
+	char *ip = strtok_r(addr_cpy, ":", &tmp);
+	if (ip == NULL) {
+		free(addr_cpy);
 		return -1;
+	}
+
+	char *port = strtok_r(NULL, ":", &tmp);
+	if (port == NULL) {
+		free(addr_cpy);
+		return -1;
+	}
 
 	sa->sin_family = AF_INET;
-	sa->sin_port = htons(atoi(part[1]));
+	sa->sin_port = htons(atoi(port));
-	inet_aton(part[0], &sa->sin_addr);
+	int ret = inet_aton(ip, &sa->sin_addr);
+
+	free(addr_cpy);
+	return (ret != 0) ? 0 : -1;
+}
+
+static int net_init_srv_cb(const char *value, void *privdata)
+{
+	struct sockaddr_in addr;
+	if (parse_saddr(value, &addr) == -1) {
+		log_print(LOG_WARN, "invalid listen addr: '%s'", value);
+		return -1;
+	}
+
+	int sock = socket(PF_INET, SOCK_DGRAM, 0);
+	if (sock < 0) {
+		log_print(LOG_ERROR, "net_init_srv_cb(): socket()");
+		return -1;
+	}
+
+	if (bind(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
+		log_print(LOG_ERROR, "net_init_srv_cb(): bind()");
+		close(sock);
+		return -1;
+	}
+
+	event_add_readfd(NULL, sock, net_receive, NULL);
+	log_print(LOG_INFO, "listen on %s:%d",
+		inet_ntoa(addr.sin_addr),
+		ntohs(addr.sin_port));
 
 	return 0;
 }
 
-int net_init_cli(void)
+static int net_init_cli_cb(const char *value, void *privdata)
 {
-	int ret = net_config_get_saddr("global", "forward", &fwd_sa);
+	struct fwd_addr *entry = malloc(sizeof(struct fwd_addr));
-	if (ret < 0)
+	if (entry == NULL) {
-		return -1;
+		log_print(LOG_ERROR, "net_init_cli_cb(): out of memory");
-
-	tx_buf = malloc(BUFSIZE);
-	if (tx_buf == NULL) {
-		log_print(LOG_ERROR, "net_init_cli(): out of memory");
 		return -1;
 	}
 
-	tx_pos = 0;
+	if (parse_saddr(value, &entry->addr) == -1) {
+		log_print(LOG_WARN, "invalid listen addr: '%s'", value);
+		free(entry);
+		return -1;
+	}
+
+	list_add(&entry->list, &fwd_list);
+	log_print(LOG_INFO, "forwarding to %s:%d",
+		inet_ntoa(entry->addr.sin_addr),
+		ntohs(entry->addr.sin_port));
+
+	return 0;
+}
+
+int net_init(void)
+{
+	int srv_cnt = config_get_strings("global", "listen", net_init_srv_cb, NULL);
+	if (srv_cnt > 0) {
+		rx_buf = malloc(BUFSIZE);
+		if (rx_buf == NULL) {
+			log_print(LOG_ERROR, "net_init(): out of memory");
+			return -1;
+		}
+	}
+
+	int cli_cnt = config_get_strings("global", "forward", net_init_cli_cb, NULL);
+	if (cli_cnt > 0) {
+		tx_buf = malloc(BUFSIZE);
+		if (tx_buf == NULL) {
+			log_print(LOG_ERROR, "net_init(): out of memory");
+			return -1;
+		}
 
 		fwd_sock = socket(PF_INET, SOCK_DGRAM, 0);
 		if (fwd_sock < 0) {
-		log_print(LOG_ERROR, "net_init_cli(): socket()");
+			log_print(LOG_ERROR, "net_init(): socket()");
 			return -1;
 		}
+	}
 
-	log_print(LOG_INFO, "forwarding to %s:%d", inet_ntoa(fwd_sa.sin_addr), ntohs(fwd_sa.sin_port));
 	return 0;
 }
-
-int net_init_srv(void)
-{
-	struct sockaddr_in sa_srv;
-
-	int ret = net_config_get_saddr("global", "listen", &sa_srv);
-	if (ret < 0)
-		return -1;
-
-	int srv_sock = socket(PF_INET, SOCK_DGRAM, 0);
-	if (srv_sock < 0) {
-		log_print(LOG_ERROR, "net_init_src(): socket()");
-		return -1;
-	}
-
-	if (bind(srv_sock, (struct sockaddr *)&sa_srv, sizeof(sa_srv)) < 0) {
-		log_print(LOG_ERROR, "net_init_src(): bind()");
-		close(srv_sock);
-		return -1;
-	}
-
-	rx_buf = malloc(BUFSIZE);
-	if (rx_buf == NULL) {
-		log_print(LOG_ERROR, "net_init_srv(): out of memory");
-		close(srv_sock);
-		return -1;
-	}
-
-	log_print(LOG_INFO, "listen on %s:%d", inet_ntoa(sa_srv.sin_addr), ntohs(sa_srv.sin_port));
-	return srv_sock;
-}

network.h

@@ -3,11 +3,9 @@
 
 #include "plugins.h"
 
-int net_init_srv(void);
+int net_init(void);
-int net_init_cli(void);
-int net_receive(int sock);
 
 int net_submit(const char *hostname, const char *plugin, const char *filename, int ds_id, const char *data);
-int net_submit_flush();
+void net_submit_flush(void);
 
 #endif /* _NETWORK_H_ */

plugins.c

@@ -30,38 +30,36 @@
 
 #include "plugins.h"
 #include "configfile.h"
+#include "event.h"
 #include "logging.h"
 #include "network.h"
 #include "rrdtool.h"
 
 #define BUFSIZE 512
 
+#define SUBMIT_NET_ONLY	0x01
+
 static LIST_HEAD(plugin_list);
 
-static int plugin_flags;
+static int submit_flags;
 
 static char *scratchpad;
 
 static int plugin_init_cb(const char *filename, void *privdata)
 {
 	int *bufsize = (int *)privdata;
-	struct sammler_plugin *plugin = NULL;
+
 	static const char *plugin_dir;
-
-	char *buffer;
-	void *tmp;
-	int len;
-
 	if (plugin_dir == NULL)
 		plugin_dir = config_get_string("global", "plugin_dir", ".");
 
-	buffer = malloc(PATH_MAX);
+	char *buffer = malloc(PATH_MAX);
 	if (buffer == NULL) {
 		log_print(LOG_ERROR, "plugin_load: out of memory");
 		return -1;
 	}
 
-	len = snprintf(buffer, PATH_MAX, "%s/%s", plugin_dir, filename);
+	int len = snprintf(buffer, PATH_MAX, "%s/%s", plugin_dir, filename);
 	if (len < 0 || len >= PATH_MAX) {
 		log_print(LOG_ERROR, "plugin_load: file name too long: %s/%s", plugin_dir, filename);
 		free(buffer);
@@ -69,7 +67,7 @@ static int plugin_init_cb(const char *filename, void *privdata)
 	}
 
 	dlerror();
-	tmp = dlopen(buffer, RTLD_NOW);
+	void *tmp = dlopen(buffer, RTLD_NOW);
 	if (tmp == NULL) {
 		log_print(LOG_ERROR, "plugin_load: dlopen: %s", dlerror());
 		free(buffer);
@@ -78,7 +76,7 @@ static int plugin_init_cb(const char *filename, void *privdata)
 
 	free(buffer);
 
-	plugin = dlsym(tmp, "plugin");
+	struct sammler_plugin *plugin = dlsym(tmp, "plugin");
 	if (plugin == NULL) {
 		log_print(LOG_ERROR, "plugin_load: failed to load '%s'", filename);
 		dlclose(tmp);
@@ -100,33 +98,12 @@ static int plugin_init_cb(const char *filename, void *privdata)
 	return 0;
 }
 
-int plugin_init(int flags)
+static int plugins_probe(void *privdata)
 {
-	struct sammler_plugin *plugin;
-	int bufsize = 0;
-
-	config_get_strings("global", "plugin", plugin_init_cb, &bufsize);
-	plugin_flags = flags;
-
-	scratchpad = malloc(bufsize);
-	if (scratchpad == NULL) {
-		log_print(LOG_ERROR, "plugin_init: out of memory");
-		return -1;
-	}
-
-	list_for_each_entry(plugin, &plugin_list, list)
-		plugin->buffer = scratchpad;
-
-	return 0;
-}
-
-void plugins_probe(void)
-{
-	struct sammler_plugin *plugin;
 	time_t now;
-
 	time(&now);
 
+	struct sammler_plugin *plugin;
 	list_for_each_entry(plugin, &plugin_list, list) {
 		if (plugin->lastprobe + plugin->interval <= now) {
 			plugin->probe();
@@ -134,8 +111,35 @@ void plugins_probe(void)
 		}
 	}
 
-	if (plugin_flags & PLUGIN_NET)
 	net_submit_flush();
+	return 0;
+}
+
+int plugin_init(void)
+{
+	int bufsize = 0;
+	config_get_strings("global", "plugin", plugin_init_cb, &bufsize);
+
+	scratchpad = malloc(bufsize);
+	if (scratchpad == NULL) {
+		log_print(LOG_ERROR, "plugin_init: out of memory");
+		return -1;
+	}
+
+	struct sammler_plugin *plugin;
+	list_for_each_entry(plugin, &plugin_list, list)
+		plugin->buffer = scratchpad;
+
+	const char *fwd_only = config_get_string("global", "forward_only", NULL);
+	if (fwd_only == NULL || strncmp(fwd_only, "true", 4))
+		submit_flags |= SUBMIT_NET_ONLY;
+
+	struct timeval tv;
+	tv.tv_sec = 1;
+	tv.tv_usec = 0;
+	event_add_timeout(&tv, plugins_probe, NULL);
+
+	return 0;
 }
 
 struct sammler_plugin * plugin_lookup(const char *name)
@@ -145,29 +149,24 @@ struct sammler_plugin * plugin_lookup(const char *name)
 		if (!strcmp(plugin->name, name))
 			return plugin;
 	}
-
 	return NULL;
 }
 
 int probe_submit(struct sammler_plugin *plugin, const char *filename, int ds_id, const char *fmt, ... )
 {
 	static const char *hostname = NULL;
-
-	va_list az;
-	char *buffer;
-	int len;
-
 	if (hostname == NULL)
 		hostname = config_get_string("global", "hostname", "localhost");
 
-	buffer = malloc(BUFSIZE);
+	char *buffer = malloc(BUFSIZE);
 	if (buffer == NULL) {
 		log_print(LOG_ERROR, "probe_submit: out of memory");
 		return -1;
 	}
 
+	va_list az;
 	va_start(az, fmt);
-	len = vsnprintf(buffer, BUFSIZE, fmt, az);
+	int len = vsnprintf(buffer, BUFSIZE, fmt, az);
 	va_end(az);
 
 	if (len < 0 || len >= BUFSIZE) {
@@ -176,12 +175,11 @@ int probe_submit(struct sammler_plugin *plugin, const char *filename, int ds_id,
 		return -1;
 	}
 
-	if (plugin_flags & PLUGIN_RRD)
-		rrd_submit(hostname, plugin->name, filename, ds_id, buffer);
-
-	if (plugin_flags & PLUGIN_NET)
 	net_submit(hostname, plugin->name, filename, ds_id, buffer);
 
+	if (!(submit_flags & SUBMIT_NET_ONLY))
+		rrd_submit(hostname, plugin->name, filename, ds_id, buffer);
+
 	free(buffer);
 	return 0;
 }

plugins.h

@@ -6,9 +6,6 @@
 #include "list.h"
 #include "logging.h"
 
-#define PLUGIN_RRD	0x01
-#define PLUGIN_NET	0x02
-
 struct sammler_plugin {
 	struct list_head list;
 	const char *name;
@@ -27,9 +24,7 @@ struct sammler_plugin {
 	const char * (*get_ds) (int ds_id);
 };
 
-int plugin_init(int flags);
+int plugin_init(void);
-
-void plugins_probe(void);
 
 struct sammler_plugin * plugin_lookup(const char *name);
 

sammler.c

@@ -28,6 +28,7 @@
 #include <signal.h>
 
 #include "configfile.h"
+#include "event.h"
 #include "logging.h"
 #include "network.h"
 #include "plugins.h"
@@ -96,44 +97,11 @@ int main(int argc, char *argv[])
 	const char *hostname = config_get_string("global", "hostname", "localhost");
 	log_print(LOG_EVERYTIME, "sammler (pid:%d) started on host '%s'", getpid(), hostname);
 
-	fd_set fdsel, fdcpy;
+	if (net_init())
-	FD_ZERO(&fdsel);
+		exit(-1);
 
-	int srv_sock = net_init_srv();
+	plugin_init();
-	if (srv_sock != -1)
-		FD_SET(srv_sock, &fdsel);
-
-	int probe_flags = 0;
-	if (net_init_cli() != -1)
-		probe_flags |= PLUGIN_NET;
-
-	const char *fwd_only = config_get_string("global", "forward_only", NULL);
-	if (fwd_only == NULL || strncmp(fwd_only, "true", 4))
-		probe_flags |= PLUGIN_RRD;
-
-	plugin_init(probe_flags);
-
-	struct timeval tv;
-	tv.tv_sec = 0;
-	tv.tv_usec = 0;
-
-	while (1) {
-		memcpy(&fdcpy, &fdsel, sizeof(fdsel));
-
-		int i = select(FD_SETSIZE, &fdcpy, NULL, NULL, &tv);
-		if (i == -1) {
-			log_print(LOG_ERROR, "select()");
-
-		/* timeout */
-		} else if (i == 0) {
-			plugins_probe();
-			tv.tv_sec = 1;
-			tv.tv_usec = 0;
-
-		} else if (FD_ISSET(srv_sock, &fdsel)) {
-			net_receive(srv_sock);
-		}
-	}
 
+	event_loop();
 	return 0;
 }