mirror of
https://git.rwth-aachen.de/acs/public/villas/node/
synced 2025-03-09 00:00:00 +01:00
Merge branch 'node-read-write-interface' into develop
This commit is contained in:
commit
51c26c0bb1
39 changed files with 198 additions and 280 deletions
|
@ -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);
|
||||
|
||||
|
|
|
@ -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.
|
||||
*
|
||||
|
|
|
@ -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
|
||||
}
|
||||
|
|
|
@ -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);
|
||||
|
||||
/** @} */
|
||||
|
||||
|
|
|
@ -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);
|
||||
|
||||
/** @} */
|
||||
|
|
|
@ -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
|
||||
}
|
||||
|
|
|
@ -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
|
||||
}
|
||||
|
|
|
@ -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
|
||||
}
|
||||
|
|
|
@ -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
|
||||
}
|
||||
|
|
|
@ -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);
|
||||
|
||||
/** @} */
|
||||
|
||||
|
|
|
@ -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);
|
||||
|
||||
/** @} */
|
||||
|
||||
|
|
|
@ -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);
|
||||
|
||||
|
|
|
@ -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);
|
||||
|
|
|
@ -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
|
||||
}
|
||||
|
|
|
@ -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);
|
||||
|
||||
/** @} */
|
||||
|
||||
|
|
|
@ -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);
|
||||
|
||||
/** @} */
|
||||
|
||||
|
|
|
@ -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);
|
||||
|
||||
/** @} */
|
||||
|
||||
|
|
|
@ -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);
|
||||
|
||||
|
|
12
lib/node.c
12
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;
|
||||
|
||||
|
|
|
@ -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;
|
||||
|
|
|
@ -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];
|
||||
|
|
|
@ -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;
|
||||
|
||||
|
|
|
@ -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)
|
||||
|
|
|
@ -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;
|
||||
|
||||
|
|
|
@ -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;
|
||||
|
||||
|
|
|
@ -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;
|
||||
|
|
|
@ -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;
|
||||
|
|
|
@ -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;
|
||||
|
|
|
@ -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 */
|
||||
|
|
|
@ -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];
|
||||
|
|
|
@ -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;
|
||||
|
||||
|
|
|
@ -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;
|
||||
|
|
|
@ -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;
|
||||
|
||||
|
|
|
@ -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;
|
||||
|
||||
|
|
|
@ -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;
|
||||
|
|
40
lib/path.c
40
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);
|
||||
}
|
||||
|
|
|
@ -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;
|
||||
|
||||
|
|
|
@ -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);
|
||||
|
|
|
@ -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);
|
||||
|
||||
|
|
Loading…
Add table
Reference in a new issue