diff --git a/etc/loopback.conf b/etc/loopback.conf index 9939254ff..803c0b432 100644 --- a/etc/loopback.conf +++ b/etc/loopback.conf @@ -32,9 +32,11 @@ nodes = { node1 = { type = "socket", layer = "udp", - local = "*:12000", # Local ip:port, use '*' for random port + local = "*:12000", # Local ip:port, use '*' for random port remote = "127.0.0.1:12001", combine = 5, + header = "villas", # app_hdr can be gtskt or default. If not provided, default header will be used + vectorize = 1, # number of samples to fetch per iteration from the socket netem = { enabled = false, delay = 1000000, @@ -45,9 +47,10 @@ nodes = { node2 = { type = "socket", layer = "udp", - local = "*:12002", # Local ip:port, use '*' for random port - remote = "127.0.0.1:12003" - combine = 30 + local = "*:12001", # Local ip:port, use '*' for random port + remote = "127.0.0.1:12002", + vectorize = 30, + header = "villas" } }; @@ -55,6 +58,6 @@ paths = ( { in = "node1", # Name of the node we listen to (see above) out = "node2", # And we loop back to the origin - hook = ["decimate", "print"] + hook = ["decimate:2", "print"] } ); diff --git a/include/villas/msg_format.h b/include/villas/msg_format.h index cb6863d49..5e9e2bccc 100644 --- a/include/villas/msg_format.h +++ b/include/villas/msg_format.h @@ -13,7 +13,7 @@ #include #ifdef __linux__ - #define _BSD_SOURCE 1 + #define _BSD_SOURCE 1 #include #elif defined(__PPC__) /* Xilinx toolchain */ #include diff --git a/include/villas/nodes/socket.h b/include/villas/nodes/socket.h index 4af49ed20..64c1dd82e 100644 --- a/include/villas/nodes/socket.h +++ b/include/villas/nodes/socket.h @@ -23,9 +23,14 @@ #include "node.h" enum socket_layer { - LAYER_ETH, - LAYER_IP, - LAYER_UDP + SOCKET_LAYER_ETH, + SOCKET_LAYER_IP, + SOCKET_LAYER_UDP +}; + +enum socket_header { + SOCKET_HEADER_DEFAULT, /**> Default header in the payload, (see msg_format.h) */ + SOCKET_HEADER_GTNET_SKT /**> No header in the payload, same as HDR_NONE*/ }; union sockaddr_union { @@ -44,6 +49,9 @@ struct socket { /** The OSI / IP layer which should be used for this socket */ enum socket_layer layer; + /** Payload header type */ + enum socket_header header; + /** Local address of the socket */ union sockaddr_union local; /** Remote address of the socket */ diff --git a/include/villas/sample.h b/include/villas/sample.h index e15c5a41d..56306ee83 100644 --- a/include/villas/sample.h +++ b/include/villas/sample.h @@ -4,7 +4,7 @@ * @author Steffen Vogel * @copyright 2014-2016, Institute for Automation of Complex Power Systems, EONERC * This file is part of VILLASnode. All Rights Reserved. Proprietary and confidential. - * Unauthorized copying of this file, via any medium is strictly prohibited. + * Unauthorized copying of this file, via any medium is strictly prohibited. */ #ifndef _SAMPLE_H_ @@ -16,6 +16,9 @@ #include #include +/* Forward declarations */ +struct pool; + /** The length of a sample datastructure with \p values values in bytes. */ #define SAMPLE_LEN(len) (sizeof(struct sample) + SAMPLE_DATA_LEN(len)) @@ -31,12 +34,13 @@ enum sample_flags { SAMPLE_OFFSET = 2, SAMPLE_SEQUENCE = 4, SAMPLE_VALUES = 8, - SAMPLE_ALL = 16-1 + SAMPLE_ALL = 16-1 }; struct sample { - int length; /**< The number of values in sample::values. */ int sequence; /**< The sequence number of this sample. */ + int length; /**< The number of values in sample::values which are valid. */ + int capacity; /**< The number of values in sample::values for which memory is reserved. */ /** All timestamps are seconds / nano seconds after 1.1.1970 UTC */ struct { @@ -44,7 +48,7 @@ struct sample { struct timespec received; /**< The point in time when this data was received. */ struct timespec sent; /**< The point in time this data was send for the last time. */ } ts; - + /** The values. */ union { float f; /**< Floating point values (note msg::endian) */ @@ -52,6 +56,9 @@ struct sample { } data[]; }; +/** Request \p cnt samples from memory pool \p p and initialize them. */ +int sample_get_many(struct pool *p, struct sample *smps[], int cnt); + /** Print a sample in human readable form to a file stream. * * @param buf A character buffer of len bytes. diff --git a/lib/cfg.c b/lib/cfg.c index c4f3b8b95..468902f27 100644 --- a/lib/cfg.c +++ b/lib/cfg.c @@ -303,7 +303,7 @@ int cfg_parse_node(config_setting_t *cfg, struct list *nodes, struct settings *s config_setting_t *cfg_vectorize = config_setting_lookup(cfg, "vectorize"); if (n->vectorize <= 0) - cerror(cfg_vectorize, "Invalid value for `vectorize`. Must be natural number!"); + cerror(cfg_vectorize, "Invalid value for `vectorize` %d. Must be natural number!", n->vectorize); if (vt->vectorize && vt->vectorize < n->vectorize) cerror(cfg_vectorize, "Invalid value for `vectorize`. Node type %s requires a number smaller than %d!", node_name_type(n), vt->vectorize); diff --git a/lib/hooks/hooks-internal.c b/lib/hooks/hooks-internal.c index 0d3d2bb7a..26958703c 100644 --- a/lib/hooks/hooks-internal.c +++ b/lib/hooks/hooks-internal.c @@ -2,7 +2,7 @@ * * @author Steffen Vogel * @copyright 2014-2016, Institute for Automation of Complex Power Systems, EONERC - * This file is part of S2SS. All Rights Reserved. Proprietary and confidential. + * This file is part of VILLASnode. All Rights Reserved. Proprietary and confidential. * Unauthorized copying of this file, via any medium is strictly prohibited. *********************************************************************************/ diff --git a/lib/nodes/opal.c b/lib/nodes/opal.c index 42849065f..86b891faf 100644 --- a/lib/nodes/opal.c +++ b/lib/nodes/opal.c @@ -88,7 +88,7 @@ int opal_deinit() if (err != EOK) error("Failed to close shared memory area (%d)", err); - debug(4, "Closing OPAL shared memory mapping"); + debug(DBG_OPAL | 4, "Closing OPAL shared memory mapping"); err = OpalSystemCtrl_UnRegister(print_shmem_name); if (err != EOK) @@ -104,7 +104,7 @@ int opal_deinit() int opal_print_global() { - debug(2, "Controller ID: %u", params.controllerID); + debug(DBG_OPAL | 2, "Controller ID: %u", params.controllerID); char *sbuf = alloc(send_icons * 5); char *rbuf = alloc(recv_icons * 5); @@ -114,17 +114,17 @@ int opal_print_global() for (int i = 0; i < recv_icons; i++) strcatf(&rbuf, "%u ", recv_ids[i]); - debug(2, "Send Blocks: %s", sbuf); - debug(2, "Receive Blocks: %s", rbuf); + debug(DBG_OPAL | 2, "Send Blocks: %s", sbuf); + debug(DBG_OPAL | 2, "Receive Blocks: %s", rbuf); free(sbuf); free(rbuf); - debug(2, "Control Block Parameters:"); + debug(DBG_OPAL | 2, "Control Block Parameters:"); for (int i = 0; i < GENASYNC_NB_FLOAT_PARAM; i++) - debug(2, "FloatParam[]%u] = %f", i, params.FloatParam[i]); + debug(DBG_OPAL | 2, "FloatParam[]%u] = %f", i, params.FloatParam[i]); for (int i = 0; i < GENASYNC_NB_STRING_PARAM; i++) - debug(2, "StringParam[%u] = %s", i, params.StringParam[i]); + debug(DBG_OPAL | 2, "StringParam[%u] = %s", i, params.StringParam[i]); return 0; } diff --git a/lib/nodes/socket.c b/lib/nodes/socket.c index a578b1c48..257da3f67 100644 --- a/lib/nodes/socket.c +++ b/lib/nodes/socket.c @@ -20,13 +20,12 @@ #include #include "nodes/socket.h" +#include "config.h" +#include "utils.h" #include "kernel/if.h" #include "kernel/nl.h" #include "kernel/tc.h" - -#include "config.h" -#include "utils.h" #include "msg.h" #include "sample.h" #include "queue.h" @@ -41,7 +40,7 @@ int socket_init(int argc, char * argv[], config_setting_t *cfg) { if (getuid() != 0) error("The 'socket' node-type requires superuser privileges!"); - + nl_init(); /* Fill link cache */ list_init(&interfaces); @@ -63,14 +62,14 @@ int socket_init(int argc, char * argv[], config_setting_t *cfg) if (rtnl_link_get_ifindex(i->nl_link) == rtnl_link_get_ifindex(link)) goto found; } - + /* If not found, create a new interface */ i = if_create(link); list_push(&interfaces, i); found: list_push(&i->sockets, s); } - + /** @todo Improve mapping of NIC IRQs per path */ int affinity; if (!config_setting_lookup_int(cfg, "affinity", &affinity)) @@ -95,22 +94,27 @@ int socket_deinit() char * socket_print(struct node *n) { struct socket *s = n->_vd; - char *layer = NULL, *buf; - + char *layer = NULL, *header = NULL, *buf; + switch (s->layer) { - case LAYER_UDP: layer = "udp"; break; - case LAYER_IP: layer = "ip"; break; - case LAYER_ETH: layer = "eth"; break; + case SOCKET_LAYER_UDP: layer = "udp"; break; + case SOCKET_LAYER_IP: layer = "ip"; break; + case SOCKET_LAYER_ETH: layer = "eth"; break; + } + + switch (s->header) { + case SOCKET_HEADER_GTNET_SKT: header = "gtnet-skt"; break; + case SOCKET_HEADER_DEFAULT: header = "villas"; break; } char *local = socket_print_addr((struct sockaddr *) &s->local); char *remote = socket_print_addr((struct sockaddr *) &s->remote); - buf = strf("layer=%s, local=%s, remote=%s", layer, local, remote); - + buf = strf("layer=%s, header=%s, local=%s, remote=%s", layer, header, local, remote); + free(local); free(remote); - + return buf; } @@ -123,9 +127,9 @@ int socket_open(struct node *n) /* Create socket */ switch (s->layer) { - case LAYER_UDP: s->sd = socket(sin->sin_family, SOCK_DGRAM, IPPROTO_UDP); break; - case LAYER_IP: s->sd = socket(sin->sin_family, SOCK_RAW, ntohs(sin->sin_port)); break; - case LAYER_ETH: s->sd = socket(sll->sll_family, SOCK_DGRAM, sll->sll_protocol); break; + case SOCKET_LAYER_UDP: s->sd = socket(sin->sin_family, SOCK_DGRAM, IPPROTO_UDP); break; + case SOCKET_LAYER_IP: s->sd = socket(sin->sin_family, SOCK_RAW, ntohs(sin->sin_port)); break; + case SOCKET_LAYER_ETH: s->sd = socket(sll->sll_family, SOCK_DGRAM, sll->sll_protocol); break; default: error("Invalid socket type!"); } @@ -150,8 +154,8 @@ int socket_open(struct node *n) /* Set socket priority, QoS or TOS IP options */ int prio; switch (s->layer) { - case LAYER_UDP: - case LAYER_IP: + case SOCKET_LAYER_UDP: + case SOCKET_LAYER_IP: prio = IPTOS_LOWDELAY; if (setsockopt(s->sd, IPPROTO_IP, IP_TOS, &prio, sizeof(prio))) serror("Failed to set type of service (QoS)"); @@ -175,11 +179,11 @@ int socket_reverse(struct node *n) { struct socket *s = n->_vd; union sockaddr_union tmp; - + tmp = s->local; s->local = s->remote; s->remote = tmp; - + return 0; } @@ -196,90 +200,131 @@ int socket_close(struct node *n) int socket_destroy(struct node *n) { struct socket *s = n->_vd; - + rtnl_qdisc_put(s->tc_qdisc); rtnl_cls_put(s->tc_classifier); - + return 0; } int socket_read(struct node *n, struct sample *smps[], unsigned cnt) { struct socket *s = n->_vd; - - int samples, ret, received; + + int samples, ret, received, length; ssize_t bytes; - struct msg msgs[cnt]; - struct msg hdr; + if (s->header == SOCKET_HEADER_GTNET_SKT) { + if (cnt < 1) + return 0; - struct iovec iov[2*cnt]; - struct msghdr mhdr = { - .msg_iov = iov - }; - - /* Peak into message header of the first sample and to get total packet size. */ - bytes = recv(s->sd, &hdr, sizeof(struct msg), MSG_PEEK | MSG_TRUNC); - if (bytes < sizeof(struct msg) || bytes % 4 != 0) { - warn("Packet size is invalid"); - return -1; + /* The GTNETv2-SKT protocol send every sample in a single packet. + * socket_read() receives a single packet. */ + struct sample *smp = smps[0]; + + /* Receive next sample */ + bytes = recv(s->sd, &smp->data[0], SAMPLE_DATA_LEN(smp->capacity), MSG_TRUNC); + if (bytes == 0) + error("Remote node %s closed the connection", node_name(n)); /** @todo Should we really hard fail here? */ + else if (bytes < 0) + serror("Failed recv from node %s", node_name(n)); + else if (bytes % 4 != 0) { + warn("Packet size is invalid: %zd Must be multiple of 4 bytes.", bytes); + recv(s->sd, NULL, 0, 0); /* empty receive buffer */ + return -1; + } + + debug(3, "Received %zd bytes", bytes); + + length = bytes / SAMPLE_DATA_LEN(1); + if (length > smp->capacity) { + warn("Node %s received more values than supported. Dropping %u values", node_name(n), length - smp->capacity); + length = smp->capacity; + } + + /** @todo Should we generate sequence no here manually? + * Or maybe optinally use the first data value as a sequence? + * However this would require the RTDS model to be changed. */ + smp->sequence = 0; + smp->length = length; + + received = 1; /* GTNET-SKT sends every sample in a single packet */ } + else { + struct msg msgs[cnt]; + struct msg hdr; + struct iovec iov[2*cnt]; + struct msghdr mhdr = { + .msg_iov = iov + }; - ret = msg_verify(&hdr); - if (ret) { - warn("Invalid message received: reason=%d, bytes=%zd", ret, bytes); - return -1; - } - - /* Convert message to host endianess */ - if (hdr.endian != MSG_ENDIAN_HOST) - msg_swap(&hdr); - - samples = bytes / MSG_LEN(hdr.length); - - if (samples > cnt) { - warn("Received more samples than supported. Dropping %u samples", samples - cnt); - samples = cnt; - } + /* Peak into message header of the first sample and to get total packet size. */ + bytes = recv(s->sd, &hdr, sizeof(struct msg), MSG_PEEK | MSG_TRUNC); + if (bytes < sizeof(struct msg) || bytes % 4 != 0) { + warn("Packet size is invalid: %zd Must be multiple of 4 bytes.", bytes); + recv(s->sd, NULL, 0, 0); /* empty receive buffer */ + return -1; + } - /* We expect that all received samples have the same amount of values! */ - for (int i = 0; i < samples; i++) { - iov[2*i+0].iov_base = &msgs[i]; - iov[2*i+0].iov_len = MSG_LEN(0); - - iov[2*i+1].iov_base = SAMPLE_DATA_OFFSET(smps[i]); - iov[2*i+1].iov_len = SAMPLE_DATA_LEN(hdr.length); - - mhdr.msg_iovlen += 2; - } - - /* Receive message from socket */ - bytes = recvmsg(s->sd, &mhdr, 0); - if (bytes == 0) - error("Remote node %s closed the connection", node_name(n)); - else if (bytes < 0) - serror("Failed recv from node %s", node_name(n)); - - for (received = 0; received < samples; received++) { - struct msg *m = &msgs[received]; - struct sample *s = smps[received]; - - ret = msg_verify(m); - if (ret) - break; - - if (m->length != hdr.length) - break; + ret = msg_verify(&hdr); + if (ret) { + warn("Invalid message received: reason=%d, bytes=%zd", ret, bytes); + recv(s->sd, NULL, 0, 0); /* empty receive buffer */ + return -1; + } /* Convert message to host endianess */ - if (m->endian != MSG_ENDIAN_HOST) - msg_swap(m); + if (hdr.endian != MSG_ENDIAN_HOST) + msg_swap(&hdr); - s->length = m->length; - s->sequence = m->sequence; - s->ts.origin = MSG_TS(m); + samples = bytes / MSG_LEN(hdr.length); + if (samples > cnt) { + warn("Node %s received more samples than supported. Dropping %u samples", node_name(n), samples - cnt); + samples = cnt; + } + + /* We expect that all received samples have the same amount of values! */ + for (int i = 0; i < samples; i++) { + iov[2*i+0].iov_base = &msgs[i]; + iov[2*i+0].iov_len = MSG_LEN(0); + + iov[2*i+1].iov_base = SAMPLE_DATA_OFFSET(smps[i]); + iov[2*i+1].iov_len = SAMPLE_DATA_LEN(hdr.length); + + mhdr.msg_iovlen += 2; + + if (hdr.length > smps[i]->capacity) + error("Node %s received more values than supported. Dropping %d values.", node_name(n), hdr.length - smps[i]->capacity); + } + + /* Receive message from socket */ + bytes = recvmsg(s->sd, &mhdr, 0); //--? samples - cnt samples dropped + if (bytes == 0) + error("Remote node %s closed the connection", node_name(n)); + else if (bytes < 0) + serror("Failed recv from node %s", node_name(n)); + + for (received = 0; received < samples; received++) { + struct msg *m = &msgs[received]; + struct sample *smp = smps[received]; + + ret = msg_verify(m); + if (ret) + break; + + if (m->length != hdr.length) + break; + + /* Convert message to host endianess */ + if (m->endian != MSG_ENDIAN_HOST) + msg_swap(m); + + smp->length = m->length; + smp->sequence = m->sequence; + smp->ts.origin = MSG_TS(m); + } } - + debug(DBG_SOCKET | 17, "Received message of %zd bytes: %u samples", bytes, received); return received; @@ -289,73 +334,99 @@ int socket_write(struct node *n, struct sample *smps[], unsigned cnt) { struct socket *s = n->_vd; ssize_t bytes; - - struct msg msgs[cnt]; - struct iovec iov[2*cnt]; - struct msghdr mhdr = { - .msg_iov = iov, - .msg_iovlen = ARRAY_LEN(iov) - }; + int sent = 0; /* Construct iovecs */ - for (int i = 0; i < cnt; i++) { - msgs[i] = MSG_INIT(smps[i]->length, smps[i]->sequence); - - msgs[i].ts.sec = smps[i]->ts.origin.tv_sec; - msgs[i].ts.nsec = smps[i]->ts.origin.tv_nsec; - - iov[i*2+0].iov_base = &msgs[i]; - iov[i*2+0].iov_len = MSG_LEN(0); + if (s->header == SOCKET_HEADER_GTNET_SKT) { + if (cnt < 1) + return 0; - iov[i*2+1].iov_base = SAMPLE_DATA_OFFSET(smps[i]); - iov[i*2+1].iov_len = SAMPLE_DATA_LEN(smps[i]->length); + for (int i = 0; i < cnt; i++) { + bytes = sendto(s->sd, &smps[i]->data, SAMPLE_DATA_LEN(smps[i]->length), 0, (struct sockaddr *) &s->remote, sizeof(s->remote)); + if (bytes < 0) + serror("Failed send to node %s", node_name(n)); + + sent++; + + debug(DBG_SOCKET | 17, "Sent packet of %zd bytes with 1 sample", bytes); + } + } + else { + struct msg msgs[cnt]; + struct iovec iov[2*cnt]; + struct msghdr mhdr = { + .msg_iov = iov, + .msg_iovlen = ARRAY_LEN(iov), + .msg_name = (struct sockaddr *) &s->remote, + .msg_namelen = sizeof(s->remote) + }; + + for (int i = 0; i < cnt; i++) { + + msgs[i] = MSG_INIT(smps[i]->length, smps[i]->sequence); + + msgs[i].ts.sec = smps[i]->ts.origin.tv_sec; + msgs[i].ts.nsec = smps[i]->ts.origin.tv_nsec; + + iov[i*2+0].iov_base = &msgs[i]; + iov[i*2+0].iov_len = MSG_LEN(0); + + iov[i*2+1].iov_base = SAMPLE_DATA_OFFSET(smps[i]); + iov[i*2+1].iov_len = SAMPLE_DATA_LEN(smps[i]->length); + } + + /* Send message */ + bytes = sendmsg(s->sd, &mhdr, 0); + if (bytes < 0) + serror("Failed send to node %s", node_name(n)); + + sent = cnt; /** @todo Find better way to determine how many values we actually sent */ + + debug(DBG_SOCKET | 17, "Sent packet of %zd bytes with %u samples", bytes, cnt); } - /* Specify destination address for connection-less procotols */ - switch (s->layer) { - case LAYER_UDP: - case LAYER_IP: - case LAYER_ETH: - mhdr.msg_name = (struct sockaddr *) &s->remote; - mhdr.msg_namelen = sizeof(s->remote); - break; - } - - /* Send message */ - bytes = sendmsg(s->sd, &mhdr, 0); - if (bytes < 0) - serror("Failed send to node %s", node_name(n)); - - debug(DBG_SOCKET | 17, "Sent packet of %zd bytes with %u samples", bytes, cnt); - - return cnt; + return sent; } int socket_parse(struct node *n, config_setting_t *cfg) { - const char *local, *remote, *layer; + const char *local, *remote, *layer, *hdr; int ret; struct socket *s = n->_vd; + /* IP layer */ if (!config_setting_lookup_string(cfg, "layer", &layer)) - cerror(cfg, "Missing layer for node %s", node_name(n)); + s->layer = SOCKET_LAYER_UDP; + else { + if (!strcmp(layer, "eth")) + s->layer = SOCKET_LAYER_ETH; + else if (!strcmp(layer, "ip")) + s->layer = SOCKET_LAYER_IP; + else if (!strcmp(layer, "udp")) + s->layer = SOCKET_LAYER_UDP; + else + cerror(cfg, "Invalid layer '%s' for node %s", layer, node_name(n)); + } - if (!strcmp(layer, "eth")) - s->layer = LAYER_ETH; - else if (!strcmp(layer, "ip")) - s->layer = LAYER_IP; - else if (!strcmp(layer, "udp")) - s->layer = LAYER_UDP; - else - cerror(cfg, "Invalid layer '%s' for node %s", layer, node_name(n)); + /* Application header */ + if (!config_setting_lookup_string(cfg, "header", &hdr)) + s->header = SOCKET_HEADER_DEFAULT; + else { + if (!strcmp(hdr, "gtnet-skt")) + s->header = SOCKET_HEADER_GTNET_SKT; + else if (!strcmp(hdr, "default")) + s->header = SOCKET_HEADER_DEFAULT; + else + cerror(cfg, "Invalid application header type '%s' for node %s", hdr, node_name(n)); + } if (!config_setting_lookup_string(cfg, "remote", &remote)) cerror(cfg, "Missing remote address for node %s", node_name(n)); if (!config_setting_lookup_string(cfg, "local", &local)) cerror(cfg, "Missing local address for node %s", node_name(n)); - + ret = socket_parse_addr(local, (struct sockaddr *) &s->local, s->layer, AI_PASSIVE); if (ret) { cerror(cfg, "Failed to resolve local address '%s' of node %s: %s", @@ -382,7 +453,7 @@ char * socket_print_addr(struct sockaddr *saddr) { union sockaddr_union *sa = (union sockaddr_union *) saddr; char *buf = alloc(64); - + /* Address */ switch (sa->sa.sa_family) { case AF_INET6: @@ -392,7 +463,7 @@ char * socket_print_addr(struct sockaddr *saddr) case AF_INET: inet_ntop(AF_INET, &sa->sin.sin_addr, buf, 64); break; - + case AF_PACKET: strcatf(&buf, "%02x", sa->sll.sll_addr[0]); for (int i = 1; i < sa->sll.sll_halen; i++) @@ -402,7 +473,7 @@ char * socket_print_addr(struct sockaddr *saddr) default: error("Unknown address family: '%u'", sa->sa.sa_family); } - + /* Port / Interface */ switch (sa->sa.sa_family) { case AF_INET6: @@ -415,7 +486,7 @@ char * socket_print_addr(struct sockaddr *saddr) struct rtnl_link *link = rtnl_link_get(cache, sa->sll.sll_ifindex); if (!link) error("Failed to get interface for index: %u", sa->sll.sll_ifindex); - + strcatf(&buf, "%%%s", rtnl_link_get_name(link)); strcatf(&buf, ":%hu", ntohs(sa->sll.sll_protocol)); break; @@ -433,7 +504,7 @@ int socket_parse_addr(const char *addr, struct sockaddr *saddr, enum socket_laye char *copy = strdup(addr); int ret; - if (layer == LAYER_ETH) { /* Format: "ab:cd:ef:12:34:56%ifname:protocol" */ + if (layer == SOCKET_LAYER_ETH) { /* Format: "ab:cd:ef:12:34:56%ifname:protocol" */ /* Split string */ char *node = strtok(copy, "%"); char *ifname = strtok(NULL, ":"); @@ -445,7 +516,7 @@ int socket_parse_addr(const char *addr, struct sockaddr *saddr, enum socket_laye error("Failed to parse MAC address: %s", node); memcpy(&sa->sll.sll_addr, &mac->ether_addr_octet, 6); - + /* Get interface index from name */ struct nl_cache *cache = nl_cache_mngt_require("route/link"); struct rtnl_link *link = rtnl_link_get_by_name(cache, ifname); @@ -476,13 +547,13 @@ int socket_parse_addr(const char *addr, struct sockaddr *saddr, enum socket_laye service = NULL; switch (layer) { - case LAYER_IP: + case SOCKET_LAYER_IP: hint.ai_socktype = SOCK_RAW; hint.ai_protocol = (service) ? strtol(service, NULL, 0) : IPPROTO_VILLAS; hint.ai_flags |= AI_NUMERICSERV; break; - case LAYER_UDP: + case SOCKET_LAYER_UDP: hint.ai_socktype = SOCK_DGRAM; hint.ai_protocol = IPPROTO_UDP; break; @@ -493,17 +564,15 @@ int socket_parse_addr(const char *addr, struct sockaddr *saddr, enum socket_laye /* Lookup address */ struct addrinfo *result; - ret = getaddrinfo(node, (layer == LAYER_IP) ? NULL : service, &hint, &result); + ret = getaddrinfo(node, (layer == SOCKET_LAYER_IP) ? NULL : service, &hint, &result); if (!ret) { - - if (layer == LAYER_IP) { + if (layer == SOCKET_LAYER_IP) { /* We mis-use the sin_port field to store the IP protocol number on RAW sockets */ struct sockaddr_in *sin = (struct sockaddr_in *) result->ai_addr; sin->sin_port = htons(result->ai_protocol); } memcpy(sa, result->ai_addr, result->ai_addrlen); - freeaddrinfo(result); } } @@ -516,7 +585,7 @@ int socket_parse_addr(const char *addr, struct sockaddr *saddr, enum socket_laye static struct node_type vt = { .name = "socket", .description = "BSD network sockets", - .vectorize = 0, /* unlimited */ + .vectorize = 0, .size = sizeof(struct socket), .destroy = socket_destroy, .reverse = socket_reverse, diff --git a/lib/path.c b/lib/path.c index e43a7d217..c032e41fa 100644 --- a/lib/path.c +++ b/lib/path.c @@ -28,7 +28,7 @@ static void path_write(struct path *p, bool resend) /* The first message in the chunk which we want to send */ if (resend) - base = p->in->received - cnt; /* we simply resent the last vector of samples */ + base = p->in->received - cnt; /* we simply resend the last vector of samples */ else { base = n->sent; } diff --git a/lib/sample.c b/lib/sample.c index bce894e16..b2d5a505b 100644 --- a/lib/sample.c +++ b/lib/sample.c @@ -8,9 +8,23 @@ #include +#include "pool.h" #include "sample.h" #include "timing.h" +int sample_get_many(struct pool *p, struct sample *smps[], int cnt) { + int ret; + + ret = pool_get_many(p, (void **) smps, cnt); + if (ret < 0) + return ret; + + for (int i = 0; i < ret; i++) + smps[i]->capacity = (p->blocksz - sizeof(**smps)) / sizeof(smps[0]->data[0]); + + return ret; +} + int sample_print(char *buf, size_t len, struct sample *s, int flags) { size_t off = snprintf(buf, len, "%llu", (unsigned long long) s->ts.origin.tv_sec); @@ -91,13 +105,12 @@ int sample_scan(const char *line, struct sample *s, int *fl) end++; } - for (s->length = 0, ptr = end; ; - s->length++, ptr = end) { + for (ptr = end, s->length = 0; + s->length < s->capacity; + ptr = end, s->length++) { - /** @todo We only support floating point values at the moment */ - s->data[s->length].f = strtod(ptr, &end); - - if (end == ptr) /* there are no valid FP values anymore */ + s->data[s->length].f = strtod(ptr, &end); /** @todo We only support floating point values at the moment */ + if (end == ptr) /* There are no valid FP values anymore */ break; } diff --git a/src/pipe.c b/src/pipe.c index e54d1652c..894733de5 100644 --- a/src/pipe.c +++ b/src/pipe.c @@ -100,7 +100,7 @@ static void * send_loop(void *ctx) if (ret < 0) error("Failed to allocate memory for receive pool."); - ret = pool_get_many(&sendd.pool, (void **) smps, node->vectorize); + ret = sample_get_many(&sendd.pool, smps, node->vectorize); if (ret < 0) error("Failed to get %u samples out of send pool (%d).", node->vectorize, ret); @@ -144,10 +144,10 @@ static void * recv_loop(void *ctx) if (ret < 0) error("Failed to allocate memory for receive pool."); - ret = pool_get_many(&recvv.pool, (void **) smps, node->vectorize); - if (ret < 0) + ret = sample_get_many(&recvv.pool, smps, node->vectorize); + if (ret < 0) error("Failed to get %u samples out of receive pool (%d).", node->vectorize, ret); - + /* Print header */ fprintf(stdout, "# %-20s\t\t%s\n", "sec.nsec+offset", "data[]"); fflush(stdout); @@ -162,7 +162,7 @@ static void * recv_loop(void *ctx) } pthread_testcancel(); } - + return NULL; } @@ -201,7 +201,7 @@ int main(int argc, char *argv[]) usage(argv[0]); } } - + /* Setup signals */ struct sigaction sa_quit = { .sa_flags = SA_SIGINFO, @@ -211,7 +211,7 @@ int main(int argc, char *argv[]) sigemptyset(&sa_quit.sa_mask); sigaction(SIGTERM, &sa_quit, NULL); sigaction(SIGINT, &sa_quit, NULL); - + /* Initialize log, configuration.. */ list_init(&nodes); @@ -225,7 +225,7 @@ int main(int argc, char *argv[]) node = list_lookup(&nodes, argv[2]); if (!node) error("Node '%s' does not exist!", argv[2]); - + if (reverse) node_reverse(node);