diff --git a/include/villas/node.h b/include/villas/node.h
index 09ca1a429..6f5503c92 100644
--- a/include/villas/node.h
+++ b/include/villas/node.h
@@ -154,9 +154,9 @@ char * node_name_long(struct node *n);
  */
 int node_reverse(struct node *n);
 
-int node_read(struct node *n, struct sample *smps[], unsigned cnt);
+int node_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
-int node_write(struct node *n, struct sample *smps[], unsigned cnt);
+int node_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 int node_fd(struct node *n);
 
diff --git a/include/villas/node_type.h b/include/villas/node_type.h
index b9026b1d3..dee935d32 100644
--- a/include/villas/node_type.h
+++ b/include/villas/node_type.h
@@ -136,7 +136,7 @@ struct node_type {
 	 * @param cnt	The number of messages which should be received.
 	 * @return	The number of messages actually received.
 	 */
-	int (*read)(struct node *n, struct sample *smps[], unsigned cnt);
+	int (*read)(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 	/** Send multiple messages in a single datagram / packet.
 	 *
@@ -150,7 +150,7 @@ struct node_type {
 	 * @param cnt	The number of messages which should be sent.
 	 * @return	The number of messages actually sent.
 	 */
-	int (*write)(struct node *n, struct sample *smps[], unsigned cnt);
+	int (*write)(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 	/** Reverse source and destination of a node.
 	 *
diff --git a/include/villas/nodes/amqp.h b/include/villas/nodes/amqp.h
index b343190c1..6c46a70d6 100644
--- a/include/villas/nodes/amqp.h
+++ b/include/villas/nodes/amqp.h
@@ -80,10 +80,10 @@ int amqp_start(struct node *n);
 int amqp_stop(struct node *n);
 
 /** @see node_type::read */
-int amqp_read(struct node *n, struct sample *smps[], unsigned cnt);
+int amqp_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::write */
-int amqp_write(struct node *n, struct sample *smps[], unsigned cnt);
+int amqp_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 #ifdef __cplusplus
 }
diff --git a/include/villas/nodes/file.h b/include/villas/nodes/file.h
index 498afb0a4..d8e24e5e1 100644
--- a/include/villas/nodes/file.h
+++ b/include/villas/nodes/file.h
@@ -83,10 +83,10 @@ int file_start(struct node *n);
 int file_stop(struct node *n);
 
 /** @see node_type::read */
-int file_read(struct node *n, struct sample *smps[], unsigned cnt);
+int file_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::write */
-int file_write(struct node *n, struct sample *smps[], unsigned cnt);
+int file_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @} */
 
diff --git a/include/villas/nodes/infiniband.h b/include/villas/nodes/infiniband.h
index cf2cb286d..67289a4dc 100644
--- a/include/villas/nodes/infiniband.h
+++ b/include/villas/nodes/infiniband.h
@@ -34,12 +34,6 @@
 #include <villas/io.h>
 #include <villas/queue_signalled.h>
 #include <rdma/rdma_cma.h>
-
-/* Function pointer typedefs */
-typedef void  (*ib_on_completion) (struct node*, struct ibv_wc*, int*);
-typedef void * (*ib_poll_function) (void*);
-typedef void * (*ib_event_function) (void*);
-
 /* Enums */
 enum poll_mode_e {
 	EVENT,
@@ -67,15 +61,6 @@ struct infiniband {
 	/* Work Completion related */
 	struct poll_s {
 		enum poll_mode_e poll_mode;
-
-		/* On completion function */
-		ib_on_completion on_compl;
-
-		/* Busy poll or Event function */
-		ib_poll_function poll_func;
-
-		/* Poll thread */
-		pthread_t cq_poller_thread;
 	} poll;
 
 	int stopThreads;
@@ -112,33 +97,33 @@ struct infiniband {
 };
 
 /** @see node_type::reverse */
-int infiniband_reverse(struct node *n);
+int ib_reverse(struct node *n);
 
 /** @see node_type::print */
-char * infiniband_print(struct node *n);
+char * ib_print(struct node *n);
 
 /** @see node_type::parse */
-int infiniband_parse(struct node *n, json_t *cfg);
+int ib_parse(struct node *n, json_t *cfg);
 
 /** @see node_type::open */
-int infiniband_start(struct node *n);
+int ib_start(struct node *n);
 
 /** @see node_type::destroy */
-int infiniband_destroy(struct node *n);
+int ib_destroy(struct node *n);
 
 /** @see node_type::close */
-int infiniband_stop(struct node *n);
+int ib_stop(struct node *n);
 
 /** @see node_type::init */
-int infiniband_init(struct super_node *n);
+int ib_init(struct super_node *n);
 
 /** @see node_type::deinit */
-int infiniband_deinit();
+int ib_deinit();
 
 /** @see node_type::read */
-int infiniband_read(struct node *n, struct sample *smps[], unsigned cnt);
+int ib_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::write */
-int infiniband_write(struct node *n, struct sample *smps[], unsigned cnt);
+int ib_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @} */
diff --git a/include/villas/nodes/influxdb.h b/include/villas/nodes/influxdb.h
index 3abbde1f9..2d9167961 100644
--- a/include/villas/nodes/influxdb.h
+++ b/include/villas/nodes/influxdb.h
@@ -65,7 +65,7 @@ int influxdb_open(struct node *n);
 int influxdb_close(struct node *n);
 
 /** @see node_type::write */
-int influxdb_write(struct node *n, struct sample *smps[], unsigned cnt);
+int influxdb_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 #ifdef __cplusplus
 }
diff --git a/include/villas/nodes/loopback.h b/include/villas/nodes/loopback.h
index 4cff73eb6..c80939440 100644
--- a/include/villas/nodes/loopback.h
+++ b/include/villas/nodes/loopback.h
@@ -63,10 +63,10 @@ int loopback_open(struct node *n);
 int loopback_close(struct node *n);
 
 /** @see node_type::read */
-int loopback_read(struct node *n, struct sample *smps[], unsigned cnt);
+int loopback_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::write */
-int loopback_write(struct node *n, struct sample *smps[], unsigned cnt);
+int loopback_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 #ifdef __cplusplus
 }
diff --git a/include/villas/nodes/mqtt.h b/include/villas/nodes/mqtt.h
index 40713c8b2..fb4b96563 100644
--- a/include/villas/nodes/mqtt.h
+++ b/include/villas/nodes/mqtt.h
@@ -95,10 +95,10 @@ int mqtt_init();
 int mqtt_deinit();
 
 /** @see node_type::read */
-int mqtt_read(struct node *n, struct sample *smps[], unsigned cnt);
+int mqtt_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::write */
-int mqtt_write(struct node *n, struct sample *smps[], unsigned cnt);
+int mqtt_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 #ifdef __cplusplus
 }
diff --git a/include/villas/nodes/nanomsg.h b/include/villas/nodes/nanomsg.h
index 11b5d2a3e..91d2496b2 100644
--- a/include/villas/nodes/nanomsg.h
+++ b/include/villas/nodes/nanomsg.h
@@ -71,10 +71,10 @@ int nanomsg_start(struct node *n);
 int nanomsg_stop(struct node *n);
 
 /** @see node_type::read */
-int nanomsg_read(struct node *n, struct sample *smps[], unsigned cnt);
+int nanomsg_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::write */
-int nanomsg_write(struct node *n, struct sample *smps[], unsigned cnt);
+int nanomsg_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 #ifdef __cplusplus
 }
diff --git a/include/villas/nodes/ngsi.h b/include/villas/nodes/ngsi.h
index 6160907a5..b417317ca 100644
--- a/include/villas/nodes/ngsi.h
+++ b/include/villas/nodes/ngsi.h
@@ -93,10 +93,10 @@ int ngsi_start(struct node *n);
 int ngsi_stop(struct node *n);
 
 /** @see node_type::read */
-int ngsi_read(struct node *n, struct sample *smps[], unsigned cnt);
+int ngsi_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::write */
-int ngsi_write(struct node *n, struct sample *smps[], unsigned cnt);
+int ngsi_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @} */
 
diff --git a/include/villas/nodes/shmem.h b/include/villas/nodes/shmem.h
index 0426645ce..b2091af9c 100644
--- a/include/villas/nodes/shmem.h
+++ b/include/villas/nodes/shmem.h
@@ -64,10 +64,10 @@ int shmem_start(struct node *n);
 int shmem_stop(struct node *n);
 
 /** @see node_type::read */
-int shmem_read(struct node *n, struct sample *smps[], unsigned cnt);
+int shmem_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::write */
-int shmem_write(struct node *n, struct sample *smps[], unsigned cnt);
+int shmem_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @} */
 
diff --git a/include/villas/nodes/signal_generator.h b/include/villas/nodes/signal_generator.h
index 18d0c0b48..ae227fdc0 100644
--- a/include/villas/nodes/signal_generator.h
+++ b/include/villas/nodes/signal_generator.h
@@ -89,7 +89,7 @@ int signal_generator_start(struct node *n);
 int signal_generator_stop(struct node *n);
 
 /** @see node_type::read */
-int signal_generator_read(struct node *n, struct sample *smps[], unsigned cnt);
+int signal_generator_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 enum signal_generator_type signal_generator_lookup_type(const char *type);
 
diff --git a/include/villas/nodes/socket.h b/include/villas/nodes/socket.h
index ad2ed1372..5db551339 100644
--- a/include/villas/nodes/socket.h
+++ b/include/villas/nodes/socket.h
@@ -119,10 +119,10 @@ int socket_start(struct node *n);
 int socket_stop(struct node *n);
 
 /** @see node_type::write */
-int socket_write(struct node *n, struct sample *smps[], unsigned cnt);
+int socket_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::read */
-int socket_read(struct node *n, struct sample *smps[], unsigned cnt);
+int socket_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::parse */
 int socket_parse(struct node *n, json_t *cfg);
diff --git a/include/villas/nodes/stats.h b/include/villas/nodes/stats.h
index 89a161656..cc50d9148 100644
--- a/include/villas/nodes/stats.h
+++ b/include/villas/nodes/stats.h
@@ -67,7 +67,7 @@ int stats_node_start(struct node *n);
 int stats_node_stop(struct node *n);
 
 /** @see node_type::read */
-int stats_node_read(struct node *n, struct sample *smps[], unsigned cnt);
+int stats_node_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 #ifdef __cplusplus
 }
diff --git a/include/villas/nodes/test_rtt.h b/include/villas/nodes/test_rtt.h
index 20f204410..6f0375232 100644
--- a/include/villas/nodes/test_rtt.h
+++ b/include/villas/nodes/test_rtt.h
@@ -79,10 +79,10 @@ int test_rtt_start(struct node *n);
 int test_rtt_stop(struct node *n);
 
 /** @see node_type::read */
-int test_rtt_read(struct node *n, struct sample *smps[], unsigned cnt);
+int test_rtt_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::write */
-int test_rtt_write(struct node *n, struct sample *smps[], unsigned cnt);
+int test_rtt_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @} */
 
diff --git a/include/villas/nodes/websocket.h b/include/villas/nodes/websocket.h
index 0e5097a17..232bcc04d 100644
--- a/include/villas/nodes/websocket.h
+++ b/include/villas/nodes/websocket.h
@@ -112,10 +112,10 @@ int websocket_stop(struct node *n);
 int websocket_destroy(struct node *n);
 
 /** @see node_type::read */
-int websocket_read(struct node *n, struct sample *smps[], unsigned cnt);
+int websocket_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::write */
-int websocket_write(struct node *n, struct sample *smps[], unsigned cnt);
+int websocket_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @} */
 
diff --git a/include/villas/nodes/zeromq.h b/include/villas/nodes/zeromq.h
index 4f31de46e..c5d62c4b2 100644
--- a/include/villas/nodes/zeromq.h
+++ b/include/villas/nodes/zeromq.h
@@ -102,10 +102,10 @@ int zeromq_start(struct node *n);
 int zeromq_stop(struct node *n);
 
 /** @see node_type::read */
-int zeromq_read(struct node *n, struct sample *smps[], unsigned cnt);
+int zeromq_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @see node_type::write */
-int zeromq_write(struct node *n, struct sample *smps[], unsigned cnt);
+int zeromq_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release);
 
 /** @} */
 
diff --git a/include/villas/stats.h b/include/villas/stats.h
index ac440a5f5..8f628ff10 100644
--- a/include/villas/stats.h
+++ b/include/villas/stats.h
@@ -72,7 +72,7 @@ int stats_destroy(struct stats *s);
 
 void stats_update(struct stats *s, enum stats_id id, double val);
 
-void stats_collect(struct stats *s, struct sample *smps[], size_t cnt);
+void stats_collect(struct stats *s, struct sample *smps[], size_t *cnt);
 
 int stats_commit(struct stats *s);
 
diff --git a/lib/node.c b/lib/node.c
index 267eccff7..b7f11a2ba 100644
--- a/lib/node.c
+++ b/lib/node.c
@@ -412,7 +412,7 @@ int node_destroy(struct node *n)
 	return 0;
 }
 
-int node_read(struct node *n, struct sample *smps[], unsigned cnt)
+int node_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int readd, nread = 0;
 
@@ -422,7 +422,7 @@ int node_read(struct node *n, struct sample *smps[], unsigned cnt)
 	/* Send in parts if vector not supported */
 	if (n->_vt->vectorize > 0 && n->_vt->vectorize < cnt) {
 		while (cnt - nread > 0) {
-			readd = n->_vt->read(n, &smps[nread], MIN(cnt - nread, n->_vt->vectorize));
+			readd = n->_vt->read(n, &smps[nread], MIN(cnt - nread, n->_vt->vectorize), release);
 			if (readd < 0)
 				return readd;
 
@@ -430,7 +430,7 @@ int node_read(struct node *n, struct sample *smps[], unsigned cnt)
 		}
 	}
 	else {
-		nread = n->_vt->read(n, smps, cnt);
+		nread = n->_vt->read(n, smps, cnt, release);
 		if (nread < 0)
 			return nread;
 	}
@@ -474,7 +474,7 @@ int node_read(struct node *n, struct sample *smps[], unsigned cnt)
 #endif /* WITH_HOOKS */
 }
 
-int node_write(struct node *n, struct sample *smps[], unsigned cnt)
+int node_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int sent, nsent = 0;
 
@@ -491,7 +491,7 @@ int node_write(struct node *n, struct sample *smps[], unsigned cnt)
 	/* Send in parts if vector not supported */
 	if (n->_vt->vectorize > 0 && n->_vt->vectorize < cnt) {
 		while (cnt - nsent > 0) {
-			sent = n->_vt->write(n, &smps[nsent], MIN(cnt - nsent, n->_vt->vectorize));
+			sent = n->_vt->write(n, &smps[nsent], MIN(cnt - nsent, n->_vt->vectorize), release);
 			if (sent < 0)
 				return sent;
 
@@ -500,7 +500,7 @@ int node_write(struct node *n, struct sample *smps[], unsigned cnt)
 		}
 	}
 	else {
-		nsent = n->_vt->write(n, smps, cnt);
+		nsent = n->_vt->write(n, smps, cnt, release);
 		if (nsent < 0)
 			return nsent;
 
diff --git a/lib/nodes/amqp.c b/lib/nodes/amqp.c
index 2e9472c2a..cd512ca50 100644
--- a/lib/nodes/amqp.c
+++ b/lib/nodes/amqp.c
@@ -301,7 +301,7 @@ int amqp_stop(struct node *n)
 	return 0;
 }
 
-int amqp_read(struct node *n, struct sample *smps[], unsigned cnt)
+int amqp_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int ret;
 	struct amqp *a = n->_vd;
@@ -319,7 +319,7 @@ int amqp_read(struct node *n, struct sample *smps[], unsigned cnt)
 	return ret;
 }
 
-int amqp_write(struct node *n, struct sample *smps[], unsigned cnt)
+int amqp_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int ret;
 	struct amqp *a = n->_vd;
diff --git a/lib/nodes/cbuilder.c b/lib/nodes/cbuilder.c
index bf0ecaf2e..a2fb61dfe 100644
--- a/lib/nodes/cbuilder.c
+++ b/lib/nodes/cbuilder.c
@@ -94,7 +94,7 @@ int cbuilder_stop(struct node *n)
 	return 0;
 }
 
-int cbuilder_read(struct node *n, struct sample *smps[], unsigned cnt)
+int cbuilder_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct cbuilder *cb = (struct cbuilder *) n->_vd;
 	struct sample *smp = smps[0];
@@ -127,7 +127,7 @@ int cbuilder_read(struct node *n, struct sample *smps[], unsigned cnt)
 	return 1;
 }
 
-int cbuilder_write(struct node *n, struct sample *smps[], unsigned cnt)
+int cbuilder_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct cbuilder *cb = (struct cbuilder *) n->_vd;
 	struct sample *smp = smps[0];
diff --git a/lib/nodes/file.c b/lib/nodes/file.c
index 674c0bd0b..95a061020 100644
--- a/lib/nodes/file.c
+++ b/lib/nodes/file.c
@@ -278,7 +278,7 @@ int file_destroy(struct node *n)
 	return 0;
 }
 
-int file_read(struct node *n, struct sample *smps[], unsigned cnt)
+int file_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct file *f = (struct file *) n->_vd;
 	int ret;
@@ -345,7 +345,7 @@ retry:	ret = io_scan(&f->io, smps, cnt);
 	return cnt;
 }
 
-int file_write(struct node *n, struct sample *smps[], unsigned cnt)
+int file_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct file *f = (struct file *) n->_vd;
 
diff --git a/lib/nodes/infiniband.c b/lib/nodes/infiniband.c
index 5b103b71e..9cff766e5 100644
--- a/lib/nodes/infiniband.c
+++ b/lib/nodes/infiniband.c
@@ -57,7 +57,7 @@ int ib_disconnect(struct node *n)
 	// Set available receive work requests to zero
 	ib->conn.available_recv_wrs = 0;
 
-	return 0;
+	return ib->stopThreads;
 }
 
 int ib_post_recv_wrs(struct node *n)
@@ -84,108 +84,6 @@ int ib_post_recv_wrs(struct node *n)
 	return ret;
 }
 
-void ib_completion_target(struct node* n, struct ibv_wc* wc, int* size){}
-
-void ib_completion_source(struct node* n, struct ibv_wc* wc, int* size)
-{
-	for (int i = 0; i < *size; i++)	{
-		if (wc[i].status != IBV_WC_SUCCESS)
-			warn("Work Completion status was not IBV_WC_SUCCES in node %s: %i",
-				node_name(n), wc[i].status);
-
-		sample_put((struct sample *) wc[i].wr_id);
-	}
-}
-
-void * ib_event_thread(void *n)
-{
-	struct infiniband *ib = (struct infiniband *) ((struct node *) n)->_vd;
-	struct ibv_wc wc[ib->cq_size];
-	int size;
-
-	debug(LOG_IB | 1, "Initialized event based poll thread of node %s", node_name(n));
-
-	while (1) {
-		// Function blocks, until an event occurs
-		ibv_get_cq_event(ib->ctx.comp_channel, &ib->ctx.send_cq, NULL);
-
-		// Poll as long as WCs are available
-		while ((size = ibv_poll_cq(ib->ctx.send_cq, ib->cq_size, wc)))
-			ib->poll.on_compl(n, wc, &size);
-
-		// Request a new event in the CQ and acknowledge event
-		ibv_req_notify_cq(ib->ctx.send_cq, 0);
-		ibv_ack_cq_events(ib->ctx.send_cq, 1);
-	}
-}
-
-void * ib_busy_poll_thread(void *n)
-{
-	struct infiniband *ib = (struct infiniband *) ((struct node *) n)->_vd;
-	struct ibv_wc wc[ib->cq_size];
-	int size;
-
-	debug(LOG_IB | 1, "Initialized busy poll thread of node %s", node_name(n));
-
-	while (1) {
-		// Poll as long as WCs are available
-		while ((size = ibv_poll_cq(ib->ctx.send_cq, ib->cq_size, wc)))
-			ib->poll.on_compl(n, wc, &size);
-
-		if (ib->stopThreads)
-			return NULL;
-	}
-}
-
-static void ib_init_wc_poll(struct node *n)
-{
-	int ret;
-	struct infiniband *ib = (struct infiniband *) n->_vd;
-	ib->ctx.comp_channel = NULL;
-
-	debug(LOG_IB | 1, "Starting to initialize completion queues and threads");
-
-	if (ib->poll.poll_mode == EVENT) {
-		// Create completion channel
-		ib->ctx.comp_channel = ibv_create_comp_channel(ib->ctx.id->verbs);
-		if (!ib->ctx.comp_channel)
-			error("Could not create completion channel in node %s", node_name(n));
-
-		debug(LOG_IB | 3, "Created Completion channel");
-	}
-
-	// Create completion queues and bind to channel (or NULL)
-	ib->ctx.recv_cq = ibv_create_cq(ib->ctx.id->verbs, ib->cq_size, NULL, NULL, 0);
-	if (!ib->ctx.recv_cq)
-		error("Could not create receive completion queue in node %s", node_name(n));
-
-	debug(LOG_IB | 3, "Created receive Completion Queue");
-
-	ib->ctx.send_cq = ibv_create_cq(ib->ctx.id->verbs, ib->cq_size, NULL, ib->ctx.comp_channel, 0);
-	if (!ib->ctx.send_cq)
-		error("Could not create send completion queue in node %s", node_name(n));
-
-	debug(LOG_IB | 3, "Created send Completion Queue");
-
-	if (ib->poll.poll_mode == EVENT) {
-		// Request notifications from completion queue
-		ret = ibv_req_notify_cq(ib->ctx.send_cq, 0);
-		if (ret)
-			error("Failed to request notifiy CQ in node %s: %s",
-				node_name(n), gai_strerror(ret));
-
-		debug(LOG_IB | 3, "Called ibv_req_notificy_cq on send Completion Queue");
-	}
-
-	// Initialize polling pthread for source
-	if (ib->is_source) {
-		ret = pthread_create(&ib->poll.cq_poller_thread, NULL, ib->poll.poll_func, n);
-		if (ret)
-			error("Failed to create poll thread of node %s: %s",
-				node_name(n), gai_strerror(ret));
-	}
-}
-
 static void ib_build_ibv(struct node *n)
 {
 	struct infiniband *ib = (struct infiniband *) n->_vd;
@@ -193,8 +91,18 @@ static void ib_build_ibv(struct node *n)
 
 	debug(LOG_IB | 1, "Starting to build IBV components");
 
-	// Initiate poll mode
-	ib_init_wc_poll(n);
+	// Create completion queues. No completion channel!)
+	ib->ctx.recv_cq = ibv_create_cq(ib->ctx.id->verbs, ib->cq_size, NULL, NULL, 0);
+	if (!ib->ctx.recv_cq)
+		error("Could not create receive completion queue in node %s", node_name(n));
+
+	debug(LOG_IB | 3, "Created receive Completion Queue");
+
+	ib->ctx.send_cq = ibv_create_cq(ib->ctx.id->verbs, ib->cq_size, NULL, NULL, 0);
+	if (!ib->ctx.send_cq)
+		error("Could not create send completion queue in node %s", node_name(n));
+
+	debug(LOG_IB | 3, "Created send Completion Queue");
 
 	// Prepare remaining Queue Pair (QP) attributes
 	ib->qp_init.send_cq = ib->ctx.send_cq;
@@ -385,14 +293,10 @@ int ib_parse(struct node *n, json_t *cfg)
 	debug(LOG_IB | 4, "Set  timeout to %i in node %s", timeout, node_name(n));
 
 	// Translate poll mode
-	if (strcmp(poll_mode, "EVENT") == 0) {
+	if (strcmp(poll_mode, "EVENT") == 0)
 		ib->poll.poll_mode = EVENT;
-		ib->poll.poll_func = ib_event_thread;
-	}
-	else if (strcmp(poll_mode, "BUSY") == 0) {
+	else if (strcmp(poll_mode, "BUSY") == 0)
 		ib->poll.poll_mode = BUSY;
-		ib->poll.poll_func = ib_busy_poll_thread;
-	}
 	else
 		error("Failed to translate poll_mode in node %s. %s is not a valid \
 			poll mode!", node_name(n), poll_mode);
@@ -444,17 +348,11 @@ int ib_parse(struct node *n, json_t *cfg)
 				remote, node_name(n), gai_strerror(ret));
 
 		debug(LOG_IB | 4, "Translated %s:%s to a struct addrinfo", ip_adr, port);
-
-		// Set correct Work Completion function
-		ib->poll.on_compl = ib_completion_source;
 	}
 	else {
 		debug(LOG_IB | 3, "Node %s is set up as target", node_name(n));
 
 		ib->is_source = 0;
-
-		// Set correct Work Completion function
-		ib->poll.on_compl = ib_completion_target;
 	}
 
 	return 0;
@@ -508,7 +406,6 @@ void * ib_rdma_cm_event_thread(void *n)
 	struct rdma_cm_event *event;
 	int ret = 0;
 
-
 	debug(LOG_IB | 1, "Started rdma_cm_event thread of node %s", node_name(node));
 
 	// Wait until node is completely started
@@ -586,7 +483,7 @@ void * ib_rdma_cm_event_thread(void *n)
 
 		rdma_ack_cm_event(event);
 
-		if (ret || ib->stopThreads)
+		if (ret)
 			break;
 	}
 
@@ -676,11 +573,10 @@ int ib_stop(struct node *n)
 	// Call RDMA disconnect function
 	// Will flush all outstanding WRs to the Completion Queue and
 	// will call RDMA_CM_EVENT_DISCONNECTED if that is done.
-	if(! ib->is_source && n->state == STATE_CONNECTED)
+	if (n->state == STATE_CONNECTED)
 		ret = rdma_disconnect(ib->ctx.id);
 	else
 		ret = rdma_disconnect(ib->ctx.listen_id);
-
 	if (ret)
 		error("Error while calling rdma_disconnect in node %s: %s",
 			node_name(n), gai_strerror(ret));
@@ -695,14 +591,6 @@ int ib_stop(struct node *n)
 
 	debug(LOG_IB | 3, "Joined rdma_cm_event_thread");
 
-	// Wait for polling thread to join
-	if (ib->is_source) {
-		ret = pthread_join(ib->poll.cq_poller_thread, NULL);
-		if (ret)
-			error("Error while joining cq_poller_thread in node %s: %i", node_name(n), ret);
-	}
-
-
 	// Destroy RDMA CM ID
 	rdma_destroy_id(ib->ctx.id);
 	debug(LOG_IB | 3, "Destroyed rdma_cm_id");
@@ -730,7 +618,7 @@ int ib_deinit()
 	return 0;
 }
 
-int ib_read(struct node *n, struct sample *smps[], unsigned cnt)
+int ib_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct infiniband *ib = (struct infiniband *) n->_vd;
 	struct ibv_wc wc[n->in.vectorize];
@@ -738,19 +626,19 @@ int ib_read(struct node *n, struct sample *smps[], unsigned cnt)
     	struct ibv_sge sge[cnt];
 	struct ibv_mr *mr;
 	int ret = 0;
+	int i = 0; //Used for first loop: post receive Work Requests
+	int j = 0; //Used for second loop: get values from Completion Queue
+	int k = 0; //Used for third loop: reorder list
 
 	debug(LOG_IB | 15, "ib_read is called");
 
 	if (n->state == STATE_CONNECTED) {
 
-		if (ib->conn.available_recv_wrs < ib->qp_init.cap.max_recv_wr && cnt==n->in.vectorize)	{
+		if (ib->conn.available_recv_wrs < ib->qp_init.cap.max_recv_wr && cnt)	{
 			// Get Memory Region
 			mr = memory_ib_get_mr(smps[0]);
 
-			for (int i = 0; i < cnt; i++) {
-				// Increase refcnt of sample
-				sample_get(smps[i]);
-
+			for (i = 0; i < cnt; i++) {
 				// Prepare receive Scatter/Gather element
     				sge[i].addr = (uint64_t) &smps[i]->data;
     				sge[i].length = SAMPLE_DATA_LEN(DEFAULT_SAMPLELEN);
@@ -768,6 +656,8 @@ int ib_read(struct node *n, struct sample *smps[], unsigned cnt)
 					debug(LOG_IB | 10, "Prepared %i new receive Work Requests", (i+1));
 
 					wr[i].next = NULL;
+					i++;
+
 					break;
 				}
 			}
@@ -779,7 +669,6 @@ int ib_read(struct node *n, struct sample *smps[], unsigned cnt)
 				    node_name(n), ret, bad_wr->wr_id);
 
 			debug(LOG_IB | 10, "Succesfully posted receive Work Requests");
-
 		}
 
 		// Poll Completion Queue
@@ -790,47 +679,62 @@ int ib_read(struct node *n, struct sample *smps[], unsigned cnt)
 
 			ib->conn.available_recv_wrs -= ret;
 
-			for (int i = 0; i < ret; i++) {
-				if (wc[i].status == IBV_WC_WR_FLUSH_ERR) {
+			for (j = 0; j < ret; j++) {
+				if (wc[j].status == IBV_WC_WR_FLUSH_ERR) {
 					debug(LOG_IB | 5, "Received IBV_WC_WR_FLUSH_ERR (ib_read). Ignore it.");
 
 					ret = 0;
 				}
-				else if (wc[i].status != IBV_WC_SUCCESS) {
+				else if (wc[j].status != IBV_WC_SUCCESS) {
 					warn("Work Completion status was not IBV_WC_SUCCES in node %s: %i",
-						node_name(n), wc[i].status);
+						node_name(n), wc[j].status);
 					ret = 0;
 				}
-				else if (wc[i].opcode & IBV_WC_RECV) {
-					smps[i] = (struct sample*)(wc[i].wr_id);
-					smps[i]->length = wc[i].byte_len/sizeof(double);
+				else if (wc[j].opcode & IBV_WC_RECV) {
+					smps[j] = (struct sample *) (wc[j].wr_id);
+					smps[j]->length = wc[i].byte_len / sizeof(double);
 				}
 				else
 					ret = 0;
 
-				//Release sample
-				sample_put((struct sample *) (wc[i].wr_id));
-				debug(LOG_IB | 10, "Releasing sample %p", (struct sample *) (wc[i].wr_id));
 			}
 		}
-	}
 
+		// Reorder list
+		// The first part of the list is fine. It's filled with received values. Now append
+		// unused values.
+		//	* j ==> Values from completion queue
+		//	* i ==> Values posted to receive queue
+		//	* (cnt)  ==> Available values to post to receive queue
+		//
+		// Thus:
+		//	* (cnt - i) ==> number of unused values
+		int l;
+		for (k = j, l = 0; k < j + (cnt - i); k++, l++) {
+			smps[k] = smps[i + l];
+		}
+
+
+		*release = k;
+	}
 	return ret;
 }
 
-int ib_write(struct node *n, struct sample *smps[], unsigned cnt)
+int ib_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *ready)
 {
 	struct infiniband *ib = (struct infiniband *) n->_vd;
 	struct ibv_send_wr wr[cnt], *bad_wr = NULL;
 	struct ibv_sge sge[cnt];
+	struct ibv_wc wc[ib->cq_size];
 	struct ibv_mr *mr;
 	int ret;
+	int i = 0; //Used for first loop: prepare work requests to post to send queue
+	int j = 0; //Used for second loop: get values from Completion Queue
 
 	debug(LOG_IB | 10, "ib_write is called");
 
 	if (n->state == STATE_CONNECTED) {
-		memset(&wr, 0, sizeof(wr));
-
+		// First, write
 		//ToDo: Place this into configuration and create checks if settings are valid
 		int send_inline = 1;
 
@@ -839,17 +743,14 @@ int ib_write(struct node *n, struct sample *smps[], unsigned cnt)
 		// Get Memory Region
 		mr = memory_ib_get_mr(smps[0]);
 
-		for (int i = 0; i < cnt; i++) {
-			// Increase refcnt of sample
-			sample_get(smps[i]);
-
+		for (i = 0; i < cnt; i++) {
 			//Set Scatter/Gather element to data of sample
-			sge[i].addr = (uint64_t)&smps[i]->data;
+			sge[i].addr = (uint64_t) &smps[i]->data;
 			sge[i].length = smps[i]->length*sizeof(double);
 			sge[i].lkey = mr->lkey;
 
 			// Set Send Work Request
-			wr[i].wr_id = (uintptr_t)smps[i]; //This way the sample can be release in WC
+			wr[i].wr_id = (uintptr_t) smps[i]; //This way the sample can be release in WC
 			wr[i].sg_list = &sge[i];
 			wr[i].num_sge = 1;
 
@@ -875,9 +776,23 @@ int ib_write(struct node *n, struct sample *smps[], unsigned cnt)
 		}
 
 		debug(LOG_IB | 4, "Succesfully posted receive Work Requests");
+
+
+		// Subsequently, check if something is available in completion queue
+		ret = ibv_poll_cq(ib->ctx.send_cq, ib->cq_size, wc);
+
+		for (j = 0; j < ret; j++) {
+			if (wc[j].status != IBV_WC_SUCCESS && wc[j].status != IBV_WC_WR_FLUSH_ERR)
+				warn("Work Completion status was not IBV_WC_SUCCES in node %s: %i",
+					node_name(n), wc[j].status);
+
+			smps[j] = (struct sample *) (wc[j].wr_id);
+		}
+
+		*ready = j;
 	}
 
-	return cnt;
+	return i;
 }
 
 int ib_fd(struct node *n)
diff --git a/lib/nodes/influxdb.c b/lib/nodes/influxdb.c
index b6fae2062..416469616 100644
--- a/lib/nodes/influxdb.c
+++ b/lib/nodes/influxdb.c
@@ -114,7 +114,7 @@ int influxdb_close(struct node *n)
 	return 0;
 }
 
-int influxdb_write(struct node *n, struct sample *smps[], unsigned cnt)
+int influxdb_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct influxdb *i = (struct influxdb *) n->_vd;
 
diff --git a/lib/nodes/loopback.c b/lib/nodes/loopback.c
index 62b11519c..cf200e1d4 100644
--- a/lib/nodes/loopback.c
+++ b/lib/nodes/loopback.c
@@ -69,7 +69,7 @@ int loopback_close(struct node *n)
 	return queue_signalled_destroy(&l->queue);
 }
 
-int loopback_read(struct node *n, struct sample *smps[], unsigned cnt)
+int loopback_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int avail;
 
@@ -86,7 +86,7 @@ int loopback_read(struct node *n, struct sample *smps[], unsigned cnt)
 	return avail;
 }
 
-int loopback_write(struct node *n, struct sample *smps[], unsigned cnt)
+int loopback_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int copied;
 
diff --git a/lib/nodes/mqtt.c b/lib/nodes/mqtt.c
index 893d10d8d..296b61985 100644
--- a/lib/nodes/mqtt.c
+++ b/lib/nodes/mqtt.c
@@ -358,7 +358,7 @@ int mqtt_deinit()
 	return 0;
 }
 
-int mqtt_read(struct node *n, struct sample *smps[], unsigned cnt)
+int mqtt_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int pulled;
 	struct mqtt *m = (struct mqtt *) n->_vd;
@@ -372,7 +372,7 @@ int mqtt_read(struct node *n, struct sample *smps[], unsigned cnt)
 	return pulled;
 }
 
-int mqtt_write(struct node *n, struct sample *smps[], unsigned cnt)
+int mqtt_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int ret;
 	struct mqtt *m = (struct mqtt *) n->_vd;
diff --git a/lib/nodes/nanomsg.c b/lib/nodes/nanomsg.c
index 432734803..0ce59c31c 100644
--- a/lib/nodes/nanomsg.c
+++ b/lib/nodes/nanomsg.c
@@ -228,7 +228,7 @@ int nanomsg_deinit()
 	return 0;
 }
 
-int nanomsg_read(struct node *n, struct sample *smps[], unsigned cnt)
+int nanomsg_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct nanomsg *m = (struct nanomsg *) n->_vd;
 	int bytes;
@@ -242,7 +242,7 @@ int nanomsg_read(struct node *n, struct sample *smps[], unsigned cnt)
 	return io_sscan(&m->io, data, bytes, NULL, smps, cnt);
 }
 
-int nanomsg_write(struct node *n, struct sample *smps[], unsigned cnt)
+int nanomsg_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int ret;
 	struct nanomsg *m = (struct nanomsg *) n->_vd;
diff --git a/lib/nodes/ngsi.c b/lib/nodes/ngsi.c
index 8e5618e43..59a8588e9 100644
--- a/lib/nodes/ngsi.c
+++ b/lib/nodes/ngsi.c
@@ -1,4 +1,4 @@
-/** Node type: OMA Next Generation Services Interface 10 (NGSI) (FIWARE context broker)
+/** Node type: OMA Next Generation Services Interface 9 (NGSI) (FIWARE context broker)
  *
  * @author Steffen Vogel <stvogel@eonerc.rwth-aachen.de>
  * @copyright 2017, Institute for Automation of Complex Power Systems, EONERC
@@ -533,7 +533,7 @@ int ngsi_stop(struct node *n)
 	return ret;
 }
 
-int ngsi_read(struct node *n, struct sample *smps[], unsigned cnt)
+int ngsi_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct ngsi *i = (struct ngsi *) n->_vd;
 	int ret;
@@ -558,7 +558,7 @@ out:	json_decref(entity);
 	return ret;
 }
 
-int ngsi_write(struct node *n, struct sample *smps[], unsigned cnt)
+int ngsi_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct ngsi *i = (struct ngsi *) n->_vd;
 	int ret;
diff --git a/lib/nodes/shmem.c b/lib/nodes/shmem.c
index 177b548be..38bfbc249 100644
--- a/lib/nodes/shmem.c
+++ b/lib/nodes/shmem.c
@@ -110,7 +110,7 @@ int shmem_stop(struct node *n)
 	return shmem_int_close(&shm->intf);
 }
 
-int shmem_read(struct node *n, struct sample *smps[], unsigned cnt)
+int shmem_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct shmem *shm = (struct shmem *) n->_vd;
 	int recv;
@@ -134,7 +134,7 @@ int shmem_read(struct node *n, struct sample *smps[], unsigned cnt)
 	return recv;
 }
 
-int shmem_write(struct node *n, struct sample *smps[], unsigned cnt)
+int shmem_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct shmem *shm = (struct shmem *) n->_vd;
 	struct sample *shared_smps[cnt]; /* Samples need to be copied to the shared pool first */
diff --git a/lib/nodes/signal_generator.c b/lib/nodes/signal_generator.c
index 13a26dba6..8d733fdc8 100644
--- a/lib/nodes/signal_generator.c
+++ b/lib/nodes/signal_generator.c
@@ -212,7 +212,7 @@ int signal_generator_stop(struct node *n)
 	return 0;
 }
 
-int signal_generator_read(struct node *n, struct sample *smps[], unsigned cnt)
+int signal_generator_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct signal_generator *s = (struct signal_generator *) n->_vd;
 	struct sample *t = smps[0];
diff --git a/lib/nodes/socket.c b/lib/nodes/socket.c
index 61c7fa2a1..ec8d58bac 100644
--- a/lib/nodes/socket.c
+++ b/lib/nodes/socket.c
@@ -336,7 +336,7 @@ int socket_destroy(struct node *n)
 	return 0;
 }
 
-int socket_read(struct node *n, struct sample *smps[], unsigned cnt)
+int socket_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int ret;
 	struct socket *s = (struct socket *) n->_vd;
@@ -401,7 +401,7 @@ out:	free(buf);
 	return ret;
 }
 
-int socket_write(struct node *n, struct sample *smps[], unsigned cnt)
+int socket_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct socket *s = (struct socket *) n->_vd;
 
diff --git a/lib/nodes/stats.c b/lib/nodes/stats.c
index 4a5e0abd9..57dcfa2ab 100644
--- a/lib/nodes/stats.c
+++ b/lib/nodes/stats.c
@@ -119,7 +119,7 @@ int stats_node_destroy(struct node *n)
 	return 0;
 }
 
-int stats_node_read(struct node *n, struct sample *smps[], unsigned cnt)
+int stats_node_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct stats_node *sn = (struct stats_node *) n->_vd;
 	struct stats *s = sn->node->stats;
diff --git a/lib/nodes/test_rtt.c b/lib/nodes/test_rtt.c
index 2de5d887f..8240c695d 100644
--- a/lib/nodes/test_rtt.c
+++ b/lib/nodes/test_rtt.c
@@ -277,7 +277,7 @@ int test_rtt_stop(struct node *n)
 	return 0;
 }
 
-int test_rtt_read(struct node *n, struct sample *smps[], unsigned cnt)
+int test_rtt_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int i, ret, values;
 	uint64_t steps;
@@ -348,7 +348,7 @@ int test_rtt_read(struct node *n, struct sample *smps[], unsigned cnt)
 	return i;
 }
 
-int test_rtt_write(struct node *n, struct sample *smps[], unsigned cnt)
+int test_rtt_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	struct test_rtt *t = (struct test_rtt *) n->_vd;
 
diff --git a/lib/nodes/websocket.c b/lib/nodes/websocket.c
index a81b8b2e8..008dbfe11 100644
--- a/lib/nodes/websocket.c
+++ b/lib/nodes/websocket.c
@@ -467,7 +467,7 @@ int websocket_destroy(struct node *n)
 	return 0;
 }
 
-int websocket_read(struct node *n, struct sample *smps[], unsigned cnt)
+int websocket_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int avail;
 
@@ -484,7 +484,7 @@ int websocket_read(struct node *n, struct sample *smps[], unsigned cnt)
 	return avail;
 }
 
-int websocket_write(struct node *n, struct sample *smps[], unsigned cnt)
+int websocket_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int avail;
 
diff --git a/lib/nodes/zeromq.c b/lib/nodes/zeromq.c
index 930e45bf9..db27d6f6b 100644
--- a/lib/nodes/zeromq.c
+++ b/lib/nodes/zeromq.c
@@ -419,7 +419,7 @@ int zeromq_destroy(struct node *n)
 	return 0;
 }
 
-int zeromq_read(struct node *n, struct sample *smps[], unsigned cnt)
+int zeromq_read(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int recv, ret;
 	struct zeromq *z = (struct zeromq *) n->_vd;
@@ -455,7 +455,7 @@ int zeromq_read(struct node *n, struct sample *smps[], unsigned cnt)
 	return recv;
 }
 
-int zeromq_write(struct node *n, struct sample *smps[], unsigned cnt)
+int zeromq_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *release)
 {
 	int ret;
 	struct zeromq *z = (struct zeromq *) n->_vd;
diff --git a/lib/path.c b/lib/path.c
index 48048d7e8..2878b5100 100644
--- a/lib/path.c
+++ b/lib/path.c
@@ -70,7 +70,8 @@ static int path_source_destroy(struct path_source *ps)
 
 static void path_source_read(struct path_source *ps, struct path *p, int i)
 {
-	int recv, tomux, ready, cnt;
+	int recv, tomux, allocated, cnt;
+	unsigned release;
 
 	cnt = ps->node->in.vectorize;
 
@@ -79,18 +80,20 @@ static void path_source_read(struct path_source *ps, struct path *p, int i)
 	struct sample **tomux_smps;
 
 	/* Fill smps[] free sample blocks from the pool */
-	ready = sample_alloc_many(&ps->pool, read_smps, cnt);
-	if (ready != cnt)
+	allocated = sample_alloc_many(&ps->pool, read_smps, cnt);
+	if (allocated != cnt)
 		warn("Pool underrun for path source %s", node_name(ps->node));
 
 	/* Read ready samples and store them to blocks pointed by smps[] */
-	recv = node_read(ps->node, read_smps, ready);
+	release = allocated;
+
+	recv = node_read(ps->node, read_smps, allocated, &release);
 	if (recv == 0)
 		goto out2;
 	else if (recv < 0)
 		error("Failed to read samples from node %s", node_name(ps->node));
-	else if (recv < ready)
-		warn("Partial read for path %s: read=%u, expected=%u", path_name(p), recv, ready);
+	else if (recv < allocated)
+		warn("Partial read for path %s: read=%u, expected=%u", path_name(p), recv, allocated);
 
 	bitset_set(&p->received, i);
 
@@ -141,7 +144,7 @@ static void path_source_read(struct path_source *ps, struct path *p, int i)
 	}
 
 	sample_put_many(muxed_smps, tomux);
-out2:	sample_put_many(read_smps, ready);
+out2:	sample_put_many(read_smps, release);
 }
 
 static int path_destination_init(struct path_destination *pd, int queuelen)
@@ -196,28 +199,31 @@ static void path_destination_write(struct path_destination *pd, struct path *p)
 {
 	int cnt = pd->node->out.vectorize;
 	int sent;
-	int available;
 	int released;
+	int allocated;
+	unsigned release;
 
 	struct sample *smps[cnt];
 
 	/* As long as there are still samples in the queue */
 	while (1) {
-		available = queue_pull_many(&pd->queue, (void **) smps, cnt);
-		if (available == 0)
+		allocated = queue_pull_many(&pd->queue, (void **) smps, cnt);
+		if (allocated == 0)
 			break;
-		else if (available < cnt)
-			debug(LOG_PATH | 5, "Queue underrun for path %s: available=%u expected=%u", path_name(p), available, cnt);
+		else if (allocated < cnt)
+			debug(LOG_PATH | 5, "Queue underrun for path %s: allocated=%u expected=%u", path_name(p), allocated, cnt);
 
-		debug(LOG_PATH | 15, "Dequeued %u samples from queue of node %s which is part of path %s", available, node_name(pd->node), path_name(p));
+		debug(LOG_PATH | 15, "Dequeued %u samples from queue of node %s which is part of path %s", allocated, node_name(pd->node), path_name(p));
 
-		sent = node_write(pd->node, smps, available);
+		release = allocated;
+
+		sent = node_write(pd->node, smps, allocated, &release);
 		if (sent < 0)
 			error("Failed to sent %u samples to node %s", cnt, node_name(pd->node));
-		else if (sent < available)
-			warn("Partial write to node %s: written=%d, expected=%d", node_name(pd->node), sent, available);
+		else if (sent < allocated)
+			warn("Partial write to node %s: written=%d, expected=%d", node_name(pd->node), sent, allocated);
 
-		released = sample_put_many(smps, sent);
+		released = sample_put_many(smps, release);
 
 		debug(LOG_PATH | 15, "Released %d samples back to memory pool", released);
 	}
diff --git a/src/villas-pipe.cpp b/src/villas-pipe.cpp
index 0133454e2..8b09fcfe7 100644
--- a/src/villas-pipe.cpp
+++ b/src/villas-pipe.cpp
@@ -128,8 +128,8 @@ static void usage()
 
 static void * send_loop(void *ctx)
 {
-	unsigned last_sequenceno = 0;
-	int ret, scanned, sent, ready, cnt = 0;
+	unsigned last_sequenceno = 0, release;
+	int ret, scanned, sent, allocated, cnt = 0;
 	struct sample *smps[node->out.vectorize];
 
 	/* Initialize memory */
@@ -139,13 +139,13 @@ static void * send_loop(void *ctx)
 		error("Failed to allocate memory for receive pool.");
 
 	while (!io_eof(&io)) {
-		ready = sample_alloc_many(&sendd.pool, smps, node->out.vectorize);
+		allocated = sample_alloc_many(&sendd.pool, smps, node->out.vectorize);
 		if (ret < 0)
 			error("Failed to get %u samples out of send pool (%d).", node->out.vectorize, ret);
-		else if (ready < node->out.vectorize)
+		else if (allocated < node->out.vectorize)
 			warn("Send pool underrun");
 
-		scanned = io_scan(&io, smps, ready);
+		scanned = io_scan(&io, smps, allocated);
 		if (scanned < 0) {
 			continue;
 			warn("Failed to read samples from stdin");
@@ -161,13 +161,15 @@ static void * send_loop(void *ctx)
 				smps[i]->sequence = last_sequenceno++;
 		}
 
-		sent = node_write(node, smps, scanned);
+		release = allocated;
+
+		sent = node_write(node, smps, scanned, &release);
 		if (sent < 0)
 			warn("Failed to sent samples to node %s: reason=%d", node_name(node), sent);
 		else if (sent < scanned)
 			warn("Failed to sent %d out of %d samples to node %s", scanned-sent, scanned, node_name(node));
 
-		sample_put_many(smps, ready);
+		sample_put_many(smps, release);
 
 		cnt += sent;
 		if (sendd.limit > 0 && cnt >= sendd.limit)
@@ -194,7 +196,8 @@ leave:	if (io_eof(&io)) {
 
 static void * recv_loop(void *ctx)
 {
-	int recv, ret, cnt = 0, ready = 0;
+	int recv, ret, cnt = 0, allocated = 0;
+	unsigned release;
 	struct sample *smps[node->in.vectorize];
 
 	/* Initialize memory */
@@ -204,19 +207,21 @@ static void * recv_loop(void *ctx)
 		error("Failed to allocate memory for receive pool.");
 
 	for (;;) {
-		ready = sample_alloc_many(&recvv.pool, smps, node->in.vectorize);
-		if (ready < 0)
+		allocated = sample_alloc_many(&recvv.pool, smps, node->in.vectorize);
+		if (allocated < 0)
 			error("Failed to allocate %u samples from receive pool.", node->in.vectorize);
-		else if (ready < node->in.vectorize)
+		else if (allocated < node->in.vectorize)
 			warn("Receive pool underrun");
 
-		recv = node_read(node, smps, ready);
+		release = allocated;
+
+		recv = node_read(node, smps, allocated, &release);
 		if (recv < 0)
 			warn("Failed to receive samples from node %s: reason=%d", node_name(node), recv);
 
 		io_print(&io, smps, recv);
 
-		sample_put_many(smps, ready);
+		sample_put_many(smps, release);
 
 		cnt += recv;
 		if (recvv.limit > 0 && cnt >= recvv.limit)
@@ -239,7 +244,7 @@ int main(int argc, char *argv[])
 
 	sendd.enabled = true;
 	sendd.limit = -1;
-	
+
 	recvv.enabled = true;
 	recvv.limit = -1;
 
diff --git a/src/villas-signal.cpp b/src/villas-signal.cpp
index 2330d83e3..508d6c641 100644
--- a/src/villas-signal.cpp
+++ b/src/villas-signal.cpp
@@ -170,7 +170,9 @@ int main(int argc, char *argv[])
 	for (;;) {
 		t = sample_alloc(&q);
 
-		node_read(&n, &t, 1);
+		unsigned release = 1; // release = allocated
+
+		node_read(&n, &t, 1, &release);
 		io_print(&io, &t, 1);
 
 		sample_put(t);
diff --git a/src/villas-test-rtt.cpp b/src/villas-test-rtt.cpp
index c591c853e..a91f33461 100644
--- a/src/villas-test-rtt.cpp
+++ b/src/villas-test-rtt.cpp
@@ -166,8 +166,13 @@ void test_rtt() {
 	while (running && (count < 0 || count--)) {
 		clock_gettime(CLOCK_ID, &send);
 
-		node_write(node, &smp_send, 1); /* Ping */
-		node_read(node,  &smp_recv, 1); /* Pong */
+		unsigned release;
+
+		release = 1; // release = allocated
+		node_write(node, &smp_send, 1, &release); /* Ping */
+
+		release = 1; // release = allocated
+		node_read(node,  &smp_recv, 1, &release); /* Pong */
 
 		clock_gettime(CLOCK_ID, &recv);