Mirrors/VILLASnode
1
0
Fork 0
mirror of https://git.rwth-aachen.de/acs/public/villas/node/ synced 2025-03-09 00:00:00 +01:00
Code Issues Releases Wiki Activity

Merge branch 'feature-gtnet' into develop

Browse source
This commit is contained in:
Steffen Vogel 2016-09-10 22:52:46 -04:00
commit 0097132b4f
11 changed files with 283 additions and 183 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
etc/loopback.conf
View file

@ -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"]
}
);

2
include/villas/msg_format.h
View file

@ -13,7 +13,7 @@
#include <stdint.h>
#ifdef __linux__
#define _BSD_SOURCE 1
#define _BSD_SOURCE 1
#include <endian.h>
#elif defined(__PPC__) /* Xilinx toolchain */
#include <lwip/arch.h>

14
include/villas/nodes/socket.h
View file

@ -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 */

15
include/villas/sample.h
View file

@ -4,7 +4,7 @@
* @author Steffen Vogel <stvogel@eonerc.rwth-aachen.de>
* @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 <time.h>
#include <sys/types.h>
/* 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.

2
lib/cfg.c
View file

@ -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);

2
lib/hooks/hooks-internal.c
View file

@ -2,7 +2,7 @@
*
* @author Steffen Vogel <stvogel@eonerc.rwth-aachen.de>
* @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.
*********************************************************************************/

14
lib/nodes/opal.c
View file

@ -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;
}

361
lib/nodes/socket.c
View file

@ -20,13 +20,12 @@
#include <arpa/inet.h>
#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,

2
lib/path.c
View file

@ -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;
}

25
lib/sample.c
View file

@ -8,9 +8,23 @@
#include <ctype.h>
#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;
}

16
src/pipe.c
View file

@ -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);