hook: fix initialization

include/villas/hook.h

@@ -62,6 +62,7 @@ struct hook {
 };
 
 int hook_init(struct hook *h, struct hook_type *vt, struct path *p, struct node *n);
+int hook_init_builtin_list(struct list *l, bool builtin, int mask, struct path *p, struct node *n);
 
 int hook_parse(struct hook *h, json_t *cfg);
 
@@ -93,7 +94,7 @@ int hook_cmp_priority(const void *a, const void *b);
  *    hooks = [ "print" ]
  * }
  */
-int hook_parse_list(struct list *list, json_t *cfg, struct path *p, struct node *n);
+int hook_parse_list(struct list *list, json_t *cfg, int mask, struct path *p, struct node *n);
 
 #ifdef __cplusplus
 }

include/villas/hook_type.h

@@ -50,7 +50,8 @@ struct sample;
 enum hook_flags {
 	HOOK_BUILTIN	= (1 << 0), /**< Should we add this hook by default to every path?. */
 	HOOK_PATH	= (1 << 1), /**< This hook type is used by paths. */
-	HOOK_NODE	= (1 << 2)  /**< This hook type is used by nodes. */
+	HOOK_NODE_READ	= (1 << 2), /**< This hook type is used by nodes. */
+	HOOK_NODE_WRITE	= (1 << 3)  /**< This hook type is used by nodes. */
 };
 
 struct hook_type {

include/villas/node.h

@@ -81,8 +81,12 @@ struct node
 	json_t *cfg;		/**< A JSON object containing the configuration of the node. */
 };
 
+/** Initialize node with default values */
 int node_init(struct node *n, struct node_type *vt);
 
+/** Do initialization after parsing the configuration */
+int node_init2(struct node *n);
+
 /** Parse settings of a node.
  *
  * @param cfg A JSON object containing the configuration of the node.

lib/hook.c

@@ -42,7 +42,8 @@ int hook_init(struct hook *h, struct hook_type *vt, struct path *p, struct node
 
 	/* Node hooks can only used with nodes,
 	   Path hooks only with paths.. */
-	if ((!(vt->flags & HOOK_NODE) && n) ||
+	if ((!(vt->flags & HOOK_NODE_READ) && n) ||
+	    (!(vt->flags & HOOK_NODE_WRITE) && n) ||
 	    (!(vt->flags & HOOK_PATH) && p))
 		return -1;
 
@@ -182,7 +183,7 @@ int hook_cmp_priority(const void *a, const void *b)
 	return ha->priority - hb->priority;
 }
 
-int hook_parse_list(struct list *list, json_t *cfg, struct path *o, struct node *n)
+int hook_parse_list(struct list *list, json_t *cfg, int mask, struct path *o, struct node *n)
 {
 	if (!json_is_array(cfg))
 		error("Hooks must be configured as a list of objects");
@@ -203,6 +204,9 @@ int hook_parse_list(struct list *list, json_t *cfg, struct path *o, struct node
 		if (!ht)
 			jerror(&err, "Unkown hook type '%s'", type);
 
+		if (!(ht->flags & mask))
+			error("Hook %s not allowed here.", type);
+
 		struct hook *h = (struct hook *) alloc(sizeof(struct hook));
 
 		ret = hook_init(h, ht, o, n);
@@ -218,3 +222,40 @@ int hook_parse_list(struct list *list, json_t *cfg, struct path *o, struct node
 
 	return 0;
 }
+
+int hook_init_builtin_list(struct list *l, bool builtin, int mask, struct path *p, struct node *n)
+{
+	int ret;
+
+	ret = list_init(l);
+	if (ret)
+		return ret;
+
+	if (!builtin)
+		return 0;
+
+	for (size_t i = 0; i < list_length(&plugins); i++) {
+		struct plugin *q = (struct plugin *) list_at(&plugins, i);
+
+		struct hook *h;
+		struct hook_type *vt = &q->hook;
+
+		if (q->type != PLUGIN_TYPE_HOOK)
+			continue;
+
+		if (vt->flags & mask)
+			continue;
+
+		h = (struct hook *) alloc(sizeof(struct hook));
+		if (!h)
+			return -1;
+
+		ret = hook_init(h, vt, p, n);
+		if (ret)
+			return ret;
+
+		list_push(l, h);
+	}
+
+	return 0;
+}

lib/hooks/decimate.c

@@ -82,7 +82,7 @@ static struct plugin p = {
 	.description	= "Downsamping by integer factor",
 	.type		= PLUGIN_TYPE_HOOK,
 	.hook		= {
-		.flags		= HOOK_NODE | HOOK_PATH,
+		.flags		= HOOK_NODE_READ | HOOK_NODE_WRITE | HOOK_PATH,
 		.priority	= 99,
 		.init		= decimate_init,
 		.parse		= decimate_parse,

lib/hooks/drop.c

@@ -119,7 +119,7 @@ static struct plugin p = {
 	.description	= "Drop messages with reordered sequence numbers",
 	.type		= PLUGIN_TYPE_HOOK,
 	.hook		= {
-		.flags		= HOOK_BUILTIN | HOOK_NODE,
+		.flags		= HOOK_BUILTIN | HOOK_NODE_READ,
 		.priority	= 3,
 		.process	= drop_process,
 		.start		= drop_start,

lib/hooks/fix.c

@@ -64,9 +64,9 @@ static struct plugin p = {
 	.description	= "Fix received data by adding missing fields",
 	.type		= PLUGIN_TYPE_HOOK,
 	.hook		= {
-		.flags		= HOOK_BUILTIN | HOOK_NODE,
+		.flags		= HOOK_BUILTIN | HOOK_NODE_READ,
 		.priority	= 1,
-		.process		= fix_process
+		.process	= fix_process
 	}
 };
 

lib/hooks/jitter_calc.c

@@ -130,7 +130,7 @@ static struct plugin p = {
 	.description	= "Calc jitter, mean and variance of GPS vs NTP TS",
 	.type		= PLUGIN_TYPE_HOOK,
 	.hook		= {
-		.flags		= HOOK_NODE,
+		.flags		= HOOK_NODE_READ | HOOK_PATH,
 		.priority	= 0,
 		.init		= jitter_calc_init,
 		.destroy	= jitter_calc_deinit,

lib/hooks/limit_rate.c

@@ -126,7 +126,7 @@ static struct plugin p = {
 	.description	= "Limit sending rate",
 	.type		= PLUGIN_TYPE_HOOK,
 	.hook		= {
-		.flags		= HOOK_NODE | HOOK_PATH,
+		.flags		= HOOK_NODE_READ | HOOK_NODE_WRITE | HOOK_PATH,
 		.priority	= 99,
 		.init		= limit_rate_init,
 		.parse		= limit_rate_parse,

lib/hooks/print.c

@@ -148,7 +148,7 @@ static struct plugin p = {
 	.description	= "Print the message to stdout",
 	.type		= PLUGIN_TYPE_HOOK,
 	.hook		= {
-		.flags		= HOOK_NODE | HOOK_PATH,
+		.flags		= HOOK_NODE_READ | HOOK_NODE_WRITE | HOOK_PATH,
 		.priority	= 99,
 		.init		= print_init,
 		.parse		= print_parse,

lib/hooks/restart.c

@@ -104,9 +104,9 @@ static struct plugin p = {
 	.description	= "Call restart hooks for current node",
 	.type		= PLUGIN_TYPE_HOOK,
 	.hook		= {
-		.flags		= HOOK_NODE | HOOK_BUILTIN,
+		.flags		= HOOK_BUILTIN | HOOK_NODE_READ,
 		.priority 	= 1,
-		.process		= restart_process,
+		.process	= restart_process,
 		.start		= restart_start,
 		.stop		= restart_stop,
 		.size		= sizeof(struct restart)

lib/hooks/shift_seq.c

@@ -63,7 +63,7 @@ static struct plugin p = {
 	.description	= "Shift sequence number of samples",
 	.type		= PLUGIN_TYPE_HOOK,
 	.hook		= {
-		.flags		= HOOK_NODE | HOOK_PATH,
+		.flags		= HOOK_NODE_READ | HOOK_PATH,
 		.priority	= 99,
 		.parse		= shift_seq_parse,
 		.process	= shift_seq_process,

lib/hooks/shift_ts.c

@@ -102,7 +102,7 @@ static struct plugin p = {
 	.description	= "Shift timestamps of samples",
 	.type		= PLUGIN_TYPE_HOOK,
 	.hook		= {
-		.flags		= HOOK_NODE | HOOK_PATH,
+		.flags		= HOOK_NODE_READ | HOOK_PATH,
 		.priority 	= 99,
 		.init		= shift_ts_init,
 		.parse		= shift_ts_parse,

lib/hooks/skip_first.c

@@ -149,7 +149,7 @@ static struct plugin p = {
 	.description	= "Skip the first samples",
 	.type		= PLUGIN_TYPE_HOOK,
 	.hook		= {
-		.flags		= HOOK_NODE | HOOK_PATH,
+		.flags		= HOOK_NODE_READ | HOOK_NODE_WRITE | HOOK_PATH,
 		.priority	= 99,
 		.parse		= skip_first_parse,
 		.start		= skip_first_restart,

lib/hooks/stats.c

@@ -195,7 +195,7 @@ static struct plugin p = {
 	.description	= "Collect statistics for the current path",
 	.type		= PLUGIN_TYPE_HOOK,
 	.hook		= {
-		.flags		= HOOK_NODE,
+		.flags		= HOOK_NODE_READ,
 		.priority	= 2,
 		.init		= stats_collect_init,
 		.destroy	= stats_collect_destroy,

lib/hooks/ts.c

@@ -42,7 +42,7 @@ static struct plugin p = {
 	.description	= "Overwrite origin timestamp of samples with receive timestamp",
 	.type		= PLUGIN_TYPE_HOOK,
 	.hook		= {
-		.flags		= HOOK_NODE,
+		.flags		= HOOK_NODE_READ,
 		.priority	= 99,
 		.process	= ts_process
 	}

lib/node.c

@@ -34,42 +34,29 @@
 #include <villas/signal.h>
 #include <villas/memory.h>
 
+static int node_direction_init2(struct node_direction *nd, struct node *n)
+{
+#ifdef WITH_HOOKS
+	int ret;
+	int m = nd == &n->out
+		? HOOK_NODE_WRITE
+		: HOOK_NODE_READ;
+
+	/* Add internal hooks if they are not already in the list */
+	ret = hook_init_builtin_list(&nd->hooks, nd->builtin, m, NULL, n);
+	if (ret)
+		return ret;
+#endif /* WITH_HOOKS */
+
+	return 0;
+}
+
 static int node_direction_init(struct node_direction *nd, struct node *n)
 {
 	nd->enabled = 0;
 	nd->vectorize = 1;
 	nd->builtin = 1;
 
-	list_init(&nd->signals);
-
-#ifdef WITH_HOOKS
-	/* Add internal hooks if they are not already in the list */
-	list_init(&nd->hooks);
-
-	if (nd->builtin) {
-		int ret;
-		for (size_t i = 0; i < list_length(&plugins); i++) {
-			struct plugin *q = (struct plugin *) list_at(&plugins, i);
-
-			if (q->type != PLUGIN_TYPE_HOOK)
-				continue;
-
-			struct hook_type *vt = &q->hook;
-
-			if (!(vt->flags & HOOK_NODE) || !(vt->flags & HOOK_BUILTIN))
-				continue;
-
-			struct hook *h = (struct hook *) alloc(sizeof(struct hook));
-
-			ret = hook_init(h, vt, NULL, n);
-			if (ret)
-				return ret;
-
-			list_push(&nd->hooks, h);
-		}
-	}
-#endif /* WITH_HOOKS */
-
 	return 0;
 }
 

lib/path.c

@@ -319,34 +319,6 @@ int path_init(struct path *p)
 	p->poll = -1;
 	p->queuelen = DEFAULT_QUEUE_LENGTH;
 
-#ifdef WITH_HOOKS
-	/* Add internal hooks if they are not already in the list */
-	list_init(&p->hooks);
-	if (p->builtin) {
-		int ret;
-
-		for (size_t i = 0; i < list_length(&plugins); i++) {
-			struct plugin *q = (struct plugin *) list_at(&plugins, i);
-
-			if (q->type != PLUGIN_TYPE_HOOK)
-				continue;
-
-			struct hook_type *vt = &q->hook;
-
-			if (!(vt->flags & HOOK_PATH) || !(vt->flags & HOOK_BUILTIN))
-				continue;
-
-			struct hook *h = (struct hook *) alloc(sizeof(struct hook));
-
-			ret = hook_init(h, vt, p, NULL);
-			if (ret)
-				return ret;
-
-			list_push(&p->hooks, h);
-		}
-	}
-#endif /* WITH_HOOKS */
-
 	p->state = STATE_INITIALIZED;
 
 	return 0;
@@ -396,6 +368,11 @@ int path_init2(struct path *p)
 	assert(p->state == STATE_CHECKED);
 
 #ifdef WITH_HOOKS
+	/* Add internal hooks if they are not already in the list */
+	ret = hook_init_builtin_list(&p->hooks, p->builtin, HOOK_PATH, p, NULL);
+	if (ret)
+		return ret;
+
 	/* We sort the hooks according to their priority before starting the path */
 	list_sort(&p->hooks, hook_cmp_priority);
 #endif /* WITH_HOOKS */

lib/super_node.c

@@ -370,6 +370,10 @@ int super_node_start(struct super_node *sn)
 
 		int refs = list_count(&sn->paths, (cmp_cb_t) path_uses_node, n);
 		if (refs > 0) {
+			ret = node_init2(n);
+			if (ret)
+				error("Failed to start node: %s", node_name(n));
+
 			ret = node_start(n);
 			if (ret)
 				error("Failed to start node: %s", node_name(n));

src/villas-pipe.cpp

@@ -374,6 +374,10 @@ check:		if (optarg == endptr)
 	if (ret)
 		error("Invalid node configuration");
 
+	ret = node_init2(node);
+	if (ret)
+		error("Failed to start node %s: reason=%d", node_name(node), ret);
+
 	ret = node_start(node);
 	if (ret)
 		error("Failed to start node %s: reason=%d", node_name(node), ret);

src/villas-signal.cpp

@@ -253,6 +253,10 @@ int main(int argc, char *argv[])
 	if (ret)
 		error("Failed to initialize pool");
 
+	ret = node_init2(&n);
+	if (ret)
+		error("Failed to start node %s: reason=%d", node_name(&n), ret);
+
 	ret = node_start(&n);
 	if (ret)
 		serror("Failed to start node");

src/villas-test-rtt.cpp

@@ -138,8 +138,17 @@ check:		if (optarg == endptr)
 	if (!node)
 		error("There's no node with the name '%s'", nodestr);
 
-	node_type_start(node->_vt, &sn);
-	node_start(node);
+	ret = node_type_start(node->_vt, &sn);
+	if (ret)
+		error("Failed to start node-type %s: reason=%d", plugin_name(node->_vt), ret);
+
+	ret = node_init2(node);
+	if (ret)
+		error("Failed to start node %s: reason=%d", node_name(node), ret);
+
+	ret = node_start(node);
+	if (ret)
+		error("Failed to start node %s: reason=%d", node_name(node), ret);
 
 	test_rtt();