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 'develop' of git.rwth-aachen.de:acs/public/villas/VILLASnode into develop

Browse source
This commit is contained in:
Steffen Vogel 2019-04-06 12:11:36 -04:00
commit 52108f4ece
51 changed files with 1073 additions and 527 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

151
.gitlab-ci.yml
View file

@ -4,6 +4,7 @@ variables:
PREFIX: /usr/
RSYNC_OPTS: --recursive --ignore-missing-args --chown ${DEPLOY_USER}:${DEPLOY_USER}
CRITERION_OPTS: --ignore-warnings
DOCKER_FILE: packaging/docker/Dockerfile.dev
DOCKER_TAG: ${CI_COMMIT_REF_NAME}
DOCKER_IMAGE: villas/node
DOCKER_IMAGE_DEV: villas/node-dev
@ -15,10 +16,9 @@ variables:
stages:
- prepare
- build
- build2
- test
- packaging
- deploy
- docker
# For some reason, GitLab CI prunes the contents of the submodules so we need to restore them.
before_script:
@ -28,34 +28,63 @@ before_script:
##############################################################################
# Build docker image which is used to build & test VILLASnode
docker-dev:
prepare:fedora:docker-dev:
stage: prepare
script:
- docker build
--file packaging/docker/Dockerfile.dev
--file ${DOCKER_FILE}
--tag ${DOCKER_IMAGE_DEV}:${DOCKER_TAG} .
tags:
- shell
- linux
prepare:raspbian:docker-dev:
extends: prepare:fedora:docker-dev
variables:
DOCKER_FILE: packaging/docker/Dockerfile.dev-raspbian
DOCKER_IMAGE_DEV: villas/node-dev-raspbian
prepare:ubuntu:docker-dev:
extends: prepare:fedora:docker-dev
variables:
DOCKER_FILE: packaging/docker/Dockerfile.dev-ubuntu
DOCKER_IMAGE_DEV: villas/node-dev-ubuntu
# Stage: build
##############################################################################
build:source:
build:fedora:x86_64:
stage: build
script:
- mkdir -p build && cd build
- cmake ..
- cmake .. ${CMAKE_OPTS}
- make ${MAKE_OPTS}
artifacts:
expire_in: 1 week
name: ${CI_PROJECT_NAME}-${CI_BUILD_REF}
paths:
- build/
image: ${DOCKER_IMAGE_DEV}:${DOCKER_TAG}
tags:
- docker
build:fedora-minimal:x86_64:
extends: build:fedora:x86_64
variables:
CMAKE_OPTS: -DWITH_HOOKS=OFF -DWITH_WEB=OFF -DWITH_API=OFF -DWITH_CONFIG=OFF -DWITH_SRC=OFF -DWITH_TOOLS=OFF -DWITH_TESTS=OFF -DWITH_PLUGINS=OFF -DWITH_CLIENTS=OFF -DWITH_DOC=OFF
build:ubuntu:x86_64:
extends: build:fedora:x86_64
variables:
LD_PRELOAD: /lib/x86_64-linux-gnu/libSegFault.so
DOCKER_IMAGE_DEV: villas/node-dev-ubuntu
build:raspbian:armv6l:
extends: build:fedora:x86_64
variables:
LD_PRELOAD: /lib/arm-linux-gnueabihf/libSegFault.so
DOCKER_IMAGE_DEV: villas/node-dev-raspbian
build:docs:
stage: build
artifacts:
@ -73,8 +102,57 @@ build:docs:
- tags
- master
build2:packages:
stage: build2
# Stage: test
##############################################################################
test:unit:
stage: test
dependencies:
- build:fedora:x86_64
script:
- mkdir -p build && cd build
- cmake .. && make ${MAKE_OPTS} unit-tests
- "tests/unit/unit-tests || true"
image: ${DOCKER_IMAGE_DEV}:${DOCKER_TAG}
tags:
- docker
test:integration:
stage: test
dependencies:
- build:fedora:x86_64
script:
- mkdir -p build && cd build
- cmake ..
- make ${MAKE_OPTS} run-integration-tests
artifacts:
name: ${CI_PROJECT_NAME}-integration-tests-${CI_BUILD_REF}
when: always
paths:
- build/tests/integration/
image: ${DOCKER_IMAGE_DEV}:${DOCKER_TAG}
tags:
- docker
# Stage: packaging
##############################################################################
packaging:docker:
stage: packaging
script:
- docker build
--build-arg BUILDER_IMAGE=${DOCKER_IMAGE_DEV}:${DOCKER_TAG}
--file packaging/docker/Dockerfile.app
--tag ${DOCKER_IMAGE}:${DOCKER_TAG} .
- docker push ${DOCKER_IMAGE}:${DOCKER_TAG}
- docker push ${DOCKER_IMAGE_DEV}:${DOCKER_TAG}
tags:
- shell
- linux
packaging:rpm:
stage: packaging
dependencies:
- build:docs
script:
@ -94,39 +172,6 @@ build2:packages:
- tags
- master
# Stage: test
##############################################################################
test:unit:
stage: test
dependencies:
- build:source
script:
- mkdir -p build && cd build
- cmake .. && make ${MAKE_OPTS} unit-tests
- "tests/unit/unit-tests || true"
image: ${DOCKER_IMAGE_DEV}:${DOCKER_TAG}
tags:
- docker
test:integration:
stage: test
dependencies:
- build:source
script:
- mkdir -p build && cd build
- cmake ..
- make ${MAKE_OPTS} run-integration-tests
artifacts:
name: ${CI_PROJECT_NAME}-integration-tests-${CI_BUILD_REF}
when: always
paths:
- build/tests/integration/
image: ${DOCKER_IMAGE_DEV}:${DOCKER_TAG}
tags:
- docker
# Stage: deploy
##############################################################################
@ -134,8 +179,8 @@ deploy:web:
stage: deploy
script:
- ssh ${DEPLOY_USER}@${DEPLOY_HOST} mkdir -p ${DEPLOY_PATH}/{coverage,doc}/${CI_BUILD_REF_NAME}/
- rsync ${RSYNC_OPTS} build/doc/html/ ${DEPLOY_USER}@${DEPLOY_HOST}:${DEPLOY_PATH}/doc/$CI_BUILD_REF_NAME/
- rsync ${RSYNC_OPTS} web/ ${DEPLOY_USER}@${DEPLOY_HOST}:${DEPLOY_PATH}/
- rsync ${RSYNC_OPTS} build/doc/html/ ${DEPLOY_USER}@${DEPLOY_HOST}:${DEPLOY_PATH}/doc/$CI_BUILD_REF_NAME/
- rsync ${RSYNC_OPTS} web/ ${DEPLOY_USER}@${DEPLOY_HOST}:${DEPLOY_PATH}/
dependencies:
- build:docs
only:
@ -151,24 +196,8 @@ deploy:packages:
- rsync ${RSYNC_OPTS} build/*.tar.gz ${DEPLOY_USER}@${DEPLOY_HOST}:${DEPLOY_PATH}/dist/
- ssh ${DEPLOY_USER}@${DEPLOY_HOST} createrepo ${DEPLOY_PATH}/../packages
dependencies:
- build2:packages
- packaging:rpm
tags:
- villas-deploy
only:
- tags
# Stage: docker
##############################################################################
docker:
stage: docker
script:
- docker build
--build-arg BUILDER_IMAGE=${DOCKER_IMAGE_DEV}:${DOCKER_TAG}
--file packaging/docker/Dockerfile.app
--tag ${DOCKER_IMAGE}:${DOCKER_TAG} .
- docker push ${DOCKER_IMAGE}:${DOCKER_TAG}
- docker push ${DOCKER_IMAGE_DEV}:${DOCKER_TAG}
tags:
- shell
- linux

88
CMakeLists.txt
View file

@ -52,6 +52,7 @@ include(CheckIncludeFile)
include(FeatureSummary)
include(GNUInstallDirs)
include(GetVersion)
include(CMakeDependentOption)
# Compiler flags
if(BUILD32)
@ -103,7 +104,7 @@ pkg_check_modules(RABBITMQ_C IMPORTED_TARGET librabbitmq>=0.8.0)
pkg_check_modules(COMEDILIB IMPORTED_TARGET comedilib>=0.11.0)
pkg_check_modules(LIBZMQ IMPORTED_TARGET libzmq>=2.2.0)
pkg_check_modules(LIBULDAQ IMPORTED_TARGET libuldaq>=1.0.0)
pkg_check_modules(UUID IMPORTED_TARGET REQUIRED uuid)
pkg_check_modules(UUID IMPORTED_TARGET REQUIRED uuid>=2.29)
pkg_check_modules(NANOMSG IMPORTED_TARGET nanomsg)
if(NOT NANOMSG_FOUND)
pkg_check_modules(NANOMSG IMPORTED_TARGET libnanomsg>=1.0.0)
@ -114,17 +115,37 @@ if(NOT RE_FOUND)
endif()
# Build options
option(WITH_HOOKS "Build with support for processing hook plugins" ON)
option(WITH_IO "Build with support format plugins" ON)
option(WITH_WEB "Build with internal webserver" ON)
option(WITH_API "Build with remote control API" ON)
option(WITH_CONFIG "Build with support for libconfig configuration syntax" ON)
option(WITH_SRC "Build villas-* executables" ${TOPLEVEL_PROJECT})
option(WITH_TOOLS "Build auxilary tools" ${TOPLEVEL_PROJECT})
option(WITH_TESTS "Run tests" ${TOPLEVEL_PROJECT})
option(WITH_PLUGINS "Build plugins" ${TOPLEVEL_PROJECT})
option(WITH_CLIENTS "Build client applications" ${TOPLEVEL_PROJECT})
option(WITH_DOC "Build documentation" ${TOPLEVEL_PROJECT})
cmake_dependent_option(WITH_HOOKS "Build with support for processing hook plugins" ON "" OFF)
cmake_dependent_option(WITH_WEB "Build with internal webserver" ON "LIBWEBSOCKETS_FOUND" OFF)
cmake_dependent_option(WITH_API "Build with remote control API" ON "" OFF)
cmake_dependent_option(WITH_CONFIG "Build with support for libconfig configuration syntax" ON "LIBCONFIG_FOUND" OFF)
cmake_dependent_option(WITH_SRC "Build executables" ON "TOPLEVEL_PROJECT" OFF)
cmake_dependent_option(WITH_TOOLS "Build auxilary tools" ON "TOPLEVEL_PROJECT" OFF)
cmake_dependent_option(WITH_TESTS "Run tests" ON "TOPLEVEL_PROJECT" OFF)
cmake_dependent_option(WITH_PLUGINS "Build plugins" ON "TOPLEVEL_PROJECT" OFF)
cmake_dependent_option(WITH_CLIENTS "Build client applications" ON "TOPLEVEL_PROJECT" OFF)
cmake_dependent_option(WITH_DOC "Build documentation" ON "TOPLEVEL_PROJECT" OFF)
cmake_dependent_option(WITH_NODE_AMQP "Build with amqp node-type" ON "RABBITMQ_C_FOUND" OFF)
cmake_dependent_option(WITH_NODE_COMEDI "Build with comedi node-type" ON "COMEDILIB_FOUND" OFF)
cmake_dependent_option(WITH_NODE_FILE "Build with file node-type" ON "" OFF)
cmake_dependent_option(WITH_NODE_IEC61850 "Build with iec61850 node-types" ON "LIBIEC61850_FOUND" OFF)
cmake_dependent_option(WITH_NODE_INFINIBAND "Build with infiniband node-type" ON "IBVERBS_FOUND; RDMACM_FOUND" OFF)
cmake_dependent_option(WITH_NODE_INFLUXDB "Build with influxdb node-type" ON "" OFF)
cmake_dependent_option(WITH_NODE_LOOPBACK "Build with loopback node-type" ON "" OFF)
cmake_dependent_option(WITH_NODE_MQTT "Build with mqtt node-type" ON "Mosquitto_FOUND" OFF)
cmake_dependent_option(WITH_NODE_NANOMSG "Build with nanomsg node-type" ON "NANOMSG_FOUND" OFF)
cmake_dependent_option(WITH_NODE_NGSI "Build with ngsi node-type" ON "" OFF)
cmake_dependent_option(WITH_NODE_OPAL "Build with opal node-type" ON "BUILD32; Opal_FOUND" OFF)
cmake_dependent_option(WITH_NODE_RTP "Build with rtp node-type" ON "RE_FOUND" OFF)
cmake_dependent_option(WITH_NODE_SHMEM "Build with shmem node-type" ON "HAS_SEMAPHORE; HAS_MMAN" OFF)
cmake_dependent_option(WITH_NODE_SIGNAL "Build with signal node-type" ON "" OFF)
cmake_dependent_option(WITH_NODE_SOCKET "Build with socket node-type" ON "LIBNL3_ROUTE_FOUND" OFF)
cmake_dependent_option(WITH_NODE_STATS "Build with stats node-type" ON "" OFF)
cmake_dependent_option(WITH_NODE_TEST_RTT "Build with test_rtt node-type" ON "" OFF)
cmake_dependent_option(WITH_NODE_ULDAQ "Build with uldaq node-type" ON "LIBULDAQ_FOUND" OFF)
cmake_dependent_option(WITH_NODE_WEBSOCKET "Build with websocket node-type" ON "WITH_WEB; LIBWEBSOCKETS_FOUND" OFF)
cmake_dependent_option(WITH_NODE_ZEROMQ "Build with zeromq node-type" ON "LIBZMQ_FOUND" OFF)
# Add more build configurations
include(cmake/config/Debug.cmake)
@ -172,16 +193,37 @@ configure_file(
)
# Show feature summary
add_feature_info(HOOKS WITH_HOOKS "Build with support for processing hook plugins")
add_feature_info(IO WITH_IO "Build with support format plugins")
add_feature_info(WEB WITH_WEB "Build with internal webserver")
add_feature_info(API WITH_API "Build with remote control API")
add_feature_info(CONFIG WITH_CONFIG "Build with support for libconfig configuration syntax")
add_feature_info(TOOLS WITH_TOOLS "Build auxilary tools")
add_feature_info(TESTS WITH_TESTS "Run tests")
add_feature_info(PLUGINS WITH_PLUGINS "Build plugins")
add_feature_info(CLIENTS WITH_CLIENTS "Build client applications")
add_feature_info(DOC WITH_DOC "Build documentation")
add_feature_info(HOOKS WITH_HOOKS "Build with support for processing hook plugins")
add_feature_info(WEB WITH_WEB "Build with internal webserver")
add_feature_info(API WITH_API "Build with remote control API")
add_feature_info(CONFIG WITH_CONFIG "Build with support for libconfig configuration syntax")
add_feature_info(SRC WITH_SRC "Build executables")
add_feature_info(TOOLS WITH_TOOLS "Build auxilary tools")
add_feature_info(TESTS WITH_TESTS "Run tests")
add_feature_info(PLUGINS WITH_PLUGINS "Build plugins")
add_feature_info(CLIENTS WITH_CLIENTS "Build client applications")
add_feature_info(DOC WITH_DOC "Build documentation")
add_feature_info(NODE_AMQP WITH_NODE_AMQP "Build with amqp node-type")
add_feature_info(NODE_COMEDI WITH_NODE_COMEDI "Build with comedi node-type")
add_feature_info(NODE_FILE WITH_NODE_FILE "Build with file node-type")
add_feature_info(NODE_IEC61850 WITH_NODE_IEC61850 "Build with iec61850 node-types")
add_feature_info(NODE_INFINIBAND WITH_NODE_INFINIBAND "Build with infiniband node-type")
add_feature_info(NODE_INFLUXDB WITH_NODE_INFLUXDB "Build with influxdb node-type")
add_feature_info(NODE_LOOPBACK WITH_NODE_LOOPBACK "Build with loopback node-type")
add_feature_info(NODE_MQTT WITH_NODE_MQTT "Build with mqtt node-type")
add_feature_info(NODE_NANOMSG WITH_NODE_NANOMSG "Build with nanomsg node-type")
add_feature_info(NODE_NGSI WITH_NODE_NGSI "Build with ngsi node-type")
add_feature_info(NODE_OPAL WITH_NODE_OPAL "Build with opal node-type")
add_feature_info(NODE_RTP WITH_NODE_RTP "Build with rtp node-type")
add_feature_info(NODE_SHMEM WITH_NODE_SHMEM "Build with shmem node-type")
add_feature_info(NODE_SIGNAL_GENERATOR WITH_NODE_SIGNAL "Build with signal node-type")
add_feature_info(NODE_SOCKET WITH_NODE_SOCKET "Build with socket node-type")
add_feature_info(NODE_STATS WITH_NODE_STATS "Build with stats node-type")
add_feature_info(NODE_TEST_RTT WITH_NODE_TEST_RTT "Build with test_rtt node-type")
add_feature_info(NODE_ULDAQ WITH_NODE_ULDAQ "Build with uldaq node-type")
add_feature_info(NODE_WEBSOCKET WITH_NODE_WEBSOCKET "Build with websocket node-type")
add_feature_info(NODE_ZEROMQ WITH_NODE_ZEROMQ "Build with zeromq node-type")
if(TOPLEVEL_PROJECT)
feature_summary(WHAT ALL VAR FEATURES)
@ -194,6 +236,8 @@ if(TOPLEVEL_PROJECT)
message(STATUS " VARIANT: ${CMAKE_PROJECT_VARIANT}")
message(STATUS " BUILD_ID: ${CMAKE_PROJECT_BUILD_ID}")
message(STATUS " BUILD_DATE: ${CMAKE_PROJECT_BUILD_DATE}")
message(STATUS " ARCH: ${CMAKE_SYSTEM_PROCESSOR}")
message(STATUS " OS: ${CMAKE_SYSTEM_NAME}")
endif()
include(VILLASnodePackaging)

2
clients/opal/models/send_receive/include/config.h
View file

@ -36,6 +36,6 @@
/* Default protocol */
#ifndef PROTOCOL
#define PROTOCOL VILLAS
#endif
#endif /* PROTOCOL */
#endif /* _CONFIG_H_ */

2
clients/opal/models/send_receive/include/msg_format.h
View file

@ -73,7 +73,7 @@ struct msg
unsigned version: 4; /**< Specifies the format of the remaining message (see MGS_VERSION) */
#else
#error Invalid byte-order
#endif
#endif /* BYTEORDER */
uint8_t id; /**< An id which identifies the source of this sample */
uint16_t length; /**< The number of values in msg::data[]. */

2
common

@ -1 +1 @@
Subproject commit 581a9b192f5b430384633c97b3efec3eaccffffd
Subproject commit 84c6f87f6f4c7ab0799161e05576e871dc045a9e

8
etc/websocket-client.conf
View file

@ -36,4 +36,12 @@ nodes = {
"http://127.0.0.1:8088/test_session"
]
}
demo-relay = {
type = "websocket",
destinations = [
"http://web.villas.fein-aachen.org/ws/relay/test_session"
]
}
}

6
include/villas/config.hpp
View file

@ -25,7 +25,7 @@
#include <jansson.h>
#include <villas/config.h>
#include <villas/node/config.h>
#include <villas/advio.h>
namespace villas {
@ -50,10 +50,10 @@ protected:
/** Decode configuration file. */
void decode();
#ifdef LIBCONFIG_FOUND
#ifdef WITH_CONFIG
/** Convert libconfig .conf file to libjansson .json file. */
void libconfigDecode();
#endif /* LIBCONFIG_FOUND */
#endif /* WITH_CONFIG */
/** Load configuration from standard input (stdim). */
void loadFromStdio();

11
include/villas/config_helper.hpp
View file

@ -24,19 +24,18 @@
#include <jansson.h>
#ifdef LIBCONFIG_FOUND
#include <libconfig.h>
#endif /* LIBCONFIG_FOUND */
#include <villas/node/config.h>
#include <villas/sample.h>
#ifdef LIBCONFIG_FOUND
#ifdef WITH_CONFIG
#include <libconfig.h>
/** Convert a libconfig object to a jansson object */
json_t *config_to_json(config_setting_t *cfg);
/** Convert a jansson object into a libconfig object. */
int json_to_config(json_t *json, config_setting_t *parent);
#endif /* LIBCONFIG_FOUND */
#endif /* WITH_CONFIG */
int json_object_extend_str(json_t *orig, const char *str);

2
include/villas/hooks/limit_rate.hpp
View file

@ -44,8 +44,6 @@ protected:
timespec last;
public:
using Hook::Hook;
void setRate(double rate)
{
deadtime = 1.0 / rate;

4
include/villas/node/config.h.in
View file

@ -51,15 +51,13 @@
#cmakedefine WITH_WEB
#cmakedefine WITH_API
#cmakedefine WITH_HOOKS
#cmakedefine WITH_IO
#cmakedefine WITH_CONFIG
/* OS Headers */
#cmakedefine HAS_EVENTFD
#cmakedefine HAS_SEMAPHORE
/* Available Libraries */
#cmakedefine LIBWEBSOCKETS_FOUND
#cmakedefine HDF5_FOUND
#cmakedefine PROTOBUF_FOUND
#cmakedefine LIBNL3_ROUTE_FOUND
#cmakedefine IBVERBS_FOUND

4
include/villas/node/exceptions.hpp
View file

@ -46,7 +46,7 @@ public:
{ }
};
#ifdef LIBCONFIG_FOUND
#ifdef WITH_CONFIG
class LibconfigParseError : public ParseError {
protected:
@ -62,7 +62,7 @@ public:
config(c)
{ }
};
#endif /* LIBCONFIG_FOUND */
#endif /* WITH_CONFIG */
class JanssonParseError : public ParseError {

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

@ -29,20 +29,11 @@
#pragma once
#include <sys/socket.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <villas/node/config.h>
#include <villas/node.h>
#include <villas/socket_addr.h>
#include <villas/io.h>
#if defined(LIBNL3_ROUTE_FOUND) && defined(__linux__)
#define WITH_SOCKET_LAYER_ETH
#include <linux/if_packet.h>
#endif /* LIBNL3_ROUTE_FOUND */
#ifdef __cplusplus
extern "C" {
#endif
@ -53,24 +44,6 @@ struct format_type;
/** The maximum length of a packet which contains stuct msg. */
#define SOCKET_INITIAL_BUFFER_LEN (64*1024)
enum socket_layer {
SOCKET_LAYER_ETH,
SOCKET_LAYER_IP,
SOCKET_LAYER_UDP,
SOCKET_LAYER_UNIX
};
union sockaddr_union {
struct sockaddr sa;
struct sockaddr_storage ss;
struct sockaddr_in sin;
struct sockaddr_in6 sin6;
struct sockaddr_un sun;
#ifdef WITH_SOCKET_LAYER_ETH
struct sockaddr_ll sll;
#endif
};
struct socket {
int sd; /**< The socket descriptor */
int verify_source; /**< Verify the source address of incoming packets against socket::remote. */
@ -120,34 +93,6 @@ int socket_parse(struct node *n, json_t *cfg);
/** @see node_type::print */
char * socket_print(struct node *n);
/** Generate printable socket address depending on the address family
*
* A IPv4 address is formatted as dotted decimals followed by the port/protocol number
* A link layer address is formatted in hexadecimals digits seperated by colons and the inferface name
*
* @param sa A pointer to the socket address.
* @return The buffer containing the textual representation of the address. The caller is responsible to free() this buffer!
*/
char * socket_print_addr(struct sockaddr *saddr);
/** Parse a socket address depending on the address family
*
* A IPv4 address has the follwing format: [hostname/ip]:[port/protocol]
* A link layer address has the following format: [mac]%[interface]:[ethertype]
*
* @todo Add support for autodetection of address type
*
* @param str A string specifiying the socket address. See description for allowed formats.
* @param sa A pointer to the resolved address
* @param layer Specifies the address type in which the addr is given
* @param flags Flags for getaddrinfo(2)
* @retval 0 Success. Everything went well.
* @retval <0 Error. Something went wrong.
*/
int socket_parse_address(const char *str, struct sockaddr *sa, enum socket_layer layer, int flags);
int socket_compare_addr(struct sockaddr *x, struct sockaddr *y);
/** @} */
#ifdef __cplusplus

92
include/villas/socket_addr.h Normal file
View file

@ -0,0 +1,92 @@
/** Node type: socket
*
* @file
* @author Steffen Vogel <stvogel@eonerc.rwth-aachen.de>
* @copyright 2014-2019, Institute for Automation of Complex Power Systems, EONERC
* @license GNU General Public License (version 3)
*
* VILLASnode
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*********************************************************************************/
#pragma once
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <villas/node/config.h>
#if defined(LIBNL3_ROUTE_FOUND) && defined(__linux__)
#define WITH_SOCKET_LAYER_ETH
#include <linux/if_packet.h>
#include <netinet/ether.h>
#endif /* LIBNL3_ROUTE_FOUND */
enum socket_layer {
SOCKET_LAYER_ETH,
SOCKET_LAYER_IP,
SOCKET_LAYER_UDP,
SOCKET_LAYER_UNIX
};
union sockaddr_union {
struct sockaddr sa;
struct sockaddr_storage ss;
struct sockaddr_in sin;
struct sockaddr_in6 sin6;
struct sockaddr_un sun;
#ifdef WITH_SOCKET_LAYER_ETH
struct sockaddr_ll sll;
#endif
};
#ifdef __cplusplus
extern "C" {
#endif
/** Generate printable socket address depending on the address family
*
* A IPv4 address is formatted as dotted decimals followed by the port/protocol number
* A link layer address is formatted in hexadecimals digits seperated by colons and the inferface name
*
* @param sa A pointer to the socket address.
* @return The buffer containing the textual representation of the address. The caller is responsible to free() this buffer!
*/
char * socket_print_addr(struct sockaddr *saddr);
/** Parse a socket address depending on the address family
*
* A IPv4 address has the follwing format: [hostname/ip]:[port/protocol]
* A link layer address has the following format: [mac]%[interface]:[ethertype]
*
* @todo Add support for autodetection of address type
*
* @param str A string specifiying the socket address. See description for allowed formats.
* @param sa A pointer to the resolved address
* @param layer Specifies the address type in which the addr is given
* @param flags Flags for getaddrinfo(2)
* @retval 0 Success. Everything went well.
* @retval <0 Error. Something went wrong.
*/
int socket_parse_address(const char *str, struct sockaddr *sa, enum socket_layer layer, int flags);
int socket_compare_addr(struct sockaddr *x, struct sockaddr *y);
#ifdef __cplusplus
}
#endif

1
include/villas/stats.h
View file

@ -54,6 +54,7 @@ enum stats_metric {
STATS_METRIC_GAP_SAMPLE, /**< Histogram for inter sample timestamps (as sent by remote). */
STATS_METRIC_GAP_RECEIVED, /**< Histogram for inter sample arrival time (as seen by this instance). */
STATS_METRIC_OWD, /**< Histogram for one-way-delay (OWD) of received samples. */
STATS_METRIC_AGE, /**< Processing time of packets within VILLASnode. */
/* RTP metrics */
STATS_METRIC_RTP_LOSS_FRACTION, /**< Fraction lost since last RTP SR/RR. */

2
include/villas/super_node.h
View file

@ -23,7 +23,7 @@
#pragma once
#include <villas/config.h>
#include <villas/node/config.h>
#ifdef __cplusplus
extern "C" {

2
include/villas/web.hpp
View file

@ -69,7 +69,7 @@ public:
*
* The web interface is based on the libwebsockets library.
*/
Web(Api *a);
Web(Api *a = nullptr);
void start();
void stop();

23
lib/CMakeLists.txt
View file

@ -62,30 +62,27 @@ set(LIB_SRC
signal.c
stats.c
super_node.cpp
socket_addr.c
io.c
format_type.c
)
if(IBVERBS_FOUND AND RDMACM_FOUND)
if(WITH_NODE_INFINIBAND)
list(APPEND LIB_SRC memory/ib.c)
endif()
add_subdirectory(nodes)
list(APPEND WHOLE_ARCHIVES nodes)
if(LIBCONFIG_FOUND)
add_subdirectory(formats)
list(APPEND WHOLE_ARCHIVES formats)
if(WITH_CONFIG)
list(APPEND INCLUDE_DIRS ${LIBCONFIG_INCLUDE_DIRS})
list(APPEND LIBRARIES PkgConfig::LIBCONFIG)
endif()
if(WITH_IO)
list(APPEND LIB_SRC
io.c
format_type.c
)
add_subdirectory(formats)
list(APPEND WHOLE_ARCHIVES formats)
endif()
if(WITH_HOOKS)
list(APPEND LIB_SRC
hook.cpp
@ -105,7 +102,7 @@ if(WITH_WEB)
list(APPEND LIBRARIES PkgConfig::LIBWEBSOCKETS)
endif()
if(WITH_API AND WITH_WEB)
if(WITH_API)
list(APPEND LIB_SRC
api.cpp
)

11
lib/api/CMakeLists.txt
View file

@ -26,9 +26,6 @@ set(API_SRC
server.cpp
sessions/socket.cpp
sessions/wsi.cpp
sessions/http.cpp
sessions/websocket.cpp
actions/capabiltities.cpp
actions/shutdown.cpp
@ -40,6 +37,14 @@ set(API_SRC
actions/node.cpp
)
if(WITH_WEB)
list(APPEND API_SRC
sessions/wsi.cpp
sessions/http.cpp
sessions/websocket.cpp
)
endif()
add_library(api STATIC ${API_SRC})
target_include_directories(api PUBLIC ${INCLUDE_DIRS})
target_link_libraries(api INTERFACE ${LIBRARIES} PUBLIC villas-common)

4
lib/api/sessions/http.cpp
View file

@ -147,9 +147,11 @@ int api_http_protocol_cb(struct lws *wsi, enum lws_callback_reasons reason, void
void *user_ctx = lws_context_user(ctx);
Web *w = static_cast<Web *>(user_ctx);
Http *s = static_cast<Http *>(user);
Api *a = w->getApi();
Http *s = static_cast<Http *>(user);
if (a == nullptr)
return -1;
switch (reason) {
case LWS_CALLBACK_HTTP_BIND_PROTOCOL:

6
lib/api/sessions/websocket.cpp
View file

@ -104,10 +104,12 @@ int api_ws_protocol_cb(lws *wsi, enum lws_callback_reasons reason, void *user, v
lws_context *ctx = lws_get_context(wsi);
void *user_ctx = lws_context_user(ctx);
Web *w = static_cast<Web*>(user_ctx);
Web *w = static_cast<Web *>(user_ctx);
WebSocket *s = static_cast<WebSocket *>(user);
Api *a = w->getApi();
WebSocket *s = static_cast<WebSocket *>(user);
if (a == nullptr)
return -1;
switch (reason) {
case LWS_CALLBACK_ESTABLISHED:

6
lib/config.cpp
View file

@ -113,15 +113,16 @@ void Config::decode()
root = json_loadf(local_file, 0, &err);
if (root == nullptr) {
#ifdef LIBCONFIG_FOUND
#ifdef WITH_CONFIG
/* We try again to parse the config in the legacy format */
libconfigDecode();
#else
throw JanssonParseError(err);
#endif /* LIBCONFIG_FOUND */
#endif /* WITH_CONFIG */
}
}
#ifdef WITH_CONFIG
void Config::libconfigDecode()
{
int ret;
@ -158,6 +159,7 @@ void Config::libconfigDecode()
config_destroy(&cfg);
}
#endif /* WITH_CONFIG */
void Config::prettyPrintError(json_error_t err)
{

4
lib/config_helper.cpp
View file

@ -29,7 +29,7 @@
#include <villas/config_helper.hpp>
#include <villas/utils.h>
#ifdef LIBCONFIG_FOUND
#ifdef WITH_CONFIG
static int json_to_config_type(int type)
{
@ -166,7 +166,7 @@ int json_to_config(json_t *json, config_setting_t *parent)
return 0;
}
#endif /* LIBCONFIG_FOUND */
#endif /* WITH_CONFIG */
void json_object_extend_key_value_token(json_t *obj, const char *key, const char *value)
{

7
lib/hooks/CMakeLists.txt
View file

@ -38,14 +38,9 @@ set(HOOK_SRC
skip_first.cpp
stats.cpp
ts.cpp
print.cpp
)
if(WITH_IO)
list(APPEND HOOK_SRC
print.cpp
)
endif()
add_library(hooks STATIC ${HOOK_SRC})
target_include_directories(hooks PUBLIC ${INCLUDE_DIRS})
target_link_libraries(hooks INTERFACE ${LIBRARIES} PUBLIC villas-common)

81
lib/hooks/ebm.cpp
View file

@ -24,8 +24,12 @@
* @{
*/
#include <vector>
#include <villas/hook.hpp>
#include <villas/log.h>
#include <villas/sample.h>
#include <villas/timing.h>
namespace villas {
namespace node {
@ -33,19 +37,86 @@ namespace node {
class EBMHook : public Hook {
protected:
char *signal_name;
unsigned signal_index;
std::vector<std::pair<int, int>> phases;
double total_energy;
double energy;
sample *last;
public:
using Hook::Hook;
virtual int process(sample *smp)
virtual void parse(json_t *cfg)
{
int ret;
size_t i;
json_error_t err;
json_t *json_phases, *json_phase;
ret = json_unpack_ex(cfg, &err, 0, "{ s: o }",
"phases", &json_phases
);
if (ret)
throw ConfigError(cfg, err, "node-config-hook-ebm");
if (!json_is_array(json_phases))
throw ConfigError(json_phases, "node-config-hook-ebm-phases");
json_array_foreach(json_phases, i, json_phase) {
int voltage, current;
ret = json_unpack_ex(json_phase, &err, 0, "[ i, i ]",
&voltage, &current
);
if (ret)
throw ConfigError(cfg, err, "node-config-hook-ebm-phases");
phases.emplace_back(voltage, current);
}
}
virtual void start()
{
assert(state == STATE_PREPARED);
energy = 0;
last = nullptr;
state = STATE_STARTED;
}
virtual void periodic()
{
assert(state == STATE_STARTED);
return HOOK_ERROR;
info("Energy: %f", energy);
}
virtual int process(sample *smp)
{
double P, P_last, dt;
assert(state == STATE_STARTED);
if (last) {
for (auto phase : phases) {
/* Trapazoidal rule */
dt = time_delta(&last->ts.origin, &smp->ts.origin);
P = smp->data[phase.first].f * smp->data[phase.second].f;
P_last = last->data[phase.first].f * last->data[phase.second].f;
energy += dt * (P_last + P) / 2.0;
}
sample_decref(last);
}
sample_incref(smp);
last = smp;
return HOOK_OK;
}
};

135
lib/hooks/stats.cpp
View file

@ -37,11 +37,103 @@
namespace villas {
namespace node {
class StatsHook;
class StatsWriteHook : public Hook {
protected:
StatsHook *parent;
public:
StatsWriteHook(struct path *p, struct node *n, int fl, int prio, bool en = true) :
Hook(p, n, fl, prio, en)
{
state = STATE_CHECKED;
}
virtual int process(sample *smp)
{
stats *s = node->stats;
timespec now = time_now();
stats_update(s, STATS_METRIC_AGE, time_delta(&smp->ts.received, &now));
return HOOK_OK;
}
};
class StatsReadHook : public Hook {
protected:
sample *last;
public:
StatsReadHook(struct path *p, struct node *n, int fl, int prio, bool en = true) :
Hook(p, n, fl, prio, en)
{
state = STATE_CHECKED;
}
virtual void start()
{
assert(state == STATE_PREPARED);
last = nullptr;
state = STATE_STARTED;
}
virtual void stop()
{
assert(state == STATE_STARTED);
if (last)
sample_decref(last);
state = STATE_STOPPED;
}
virtual int process(sample *smp)
{
stats *s = node->stats;
if (last) {
if (smp->flags & last->flags & SAMPLE_HAS_TS_RECEIVED)
stats_update(s, STATS_METRIC_GAP_RECEIVED, time_delta(&last->ts.received, &smp->ts.received));
if (smp->flags & last->flags & SAMPLE_HAS_TS_ORIGIN)
stats_update(s, STATS_METRIC_GAP_SAMPLE, time_delta(&last->ts.origin, &smp->ts.origin));
if ((smp->flags & SAMPLE_HAS_TS_ORIGIN) && (smp->flags & SAMPLE_HAS_TS_RECEIVED))
stats_update(s, STATS_METRIC_OWD, time_delta(&smp->ts.origin, &smp->ts.received));
if (smp->flags & last->flags & SAMPLE_HAS_SEQUENCE) {
int dist = smp->sequence - (int32_t) last->sequence;
if (dist != 1)
stats_update(s, STATS_METRIC_SMPS_REORDERED, dist);
}
}
sample_incref(smp);
if (last)
sample_decref(last);
last = smp;
return HOOK_OK;
}
};
class StatsHook : public Hook {
protected:
struct stats stats;
StatsReadHook *readHook;
StatsWriteHook *writeHook;
enum stats_format format;
int verbose;
int warmup;
@ -50,8 +142,6 @@ protected:
AFILE *output;
char *uri;
sample *last;
public:
StatsHook(struct path *p, struct node *n, int fl, int prio, bool en = true) :
@ -68,8 +158,14 @@ public:
/* Register statistic object to path.
*
* This allows the path code to update statistics. */
if (node)
node->stats = &stats;
node->stats = &stats;
/* Add child hooks */
readHook = new StatsReadHook(p, n, fl, prio, en);
writeHook = new StatsWriteHook(p, n, fl, prio, en);
vlist_push(&node->in.hooks, (void *) readHook);
vlist_push(&node->out.hooks, (void *) writeHook);
}
~StatsHook()
@ -165,37 +261,6 @@ public:
state = STATE_PARSED;
}
virtual int process(sample *smp)
{
struct stats *s = &stats;
if (last) {
if (smp->flags & last->flags & SAMPLE_HAS_TS_RECEIVED)
stats_update(s, STATS_METRIC_GAP_RECEIVED, time_delta(&last->ts.received, &smp->ts.received));
if (smp->flags & last->flags & SAMPLE_HAS_TS_ORIGIN)
stats_update(s, STATS_METRIC_GAP_SAMPLE, time_delta(&last->ts.origin, &smp->ts.origin));
if ((smp->flags & SAMPLE_HAS_TS_ORIGIN) && (smp->flags & SAMPLE_HAS_TS_RECEIVED))
stats_update(s, STATS_METRIC_OWD, time_delta(&smp->ts.origin, &smp->ts.received));
if (smp->flags & last->flags & SAMPLE_HAS_SEQUENCE) {
int dist = smp->sequence - (int32_t) last->sequence;
if (dist != 1)
stats_update(s, STATS_METRIC_SMPS_REORDERED, dist);
}
}
sample_incref(smp);
if (last)
sample_decref(last);
last = smp;
return HOOK_OK;
}
};
/* Register hook */

2
lib/memory/hugepage.c
View file

@ -34,7 +34,7 @@
/* Required to allocate hugepages on Apple OS X */
#ifdef __MACH__
#include <mach/vm_statistics.h>
#endif
#endif /* __MACH__ */
#include <villas/kernel/kernel.h>
#include <villas/log.h>

3
lib/node_direction.c
View file

@ -21,6 +21,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*********************************************************************************/
#include <villas/config.h>
#include <villas/utils.h>
#include <villas/hook.h>
#include <villas/hook_list.h>
@ -226,7 +227,7 @@ struct vlist * node_direction_get_signals(struct node_direction *nd)
#ifdef WITH_HOOKS
if (vlist_length(&nd->hooks) > 0)
return hook_list_get_signals(&nd->hooks);
#endif
#endif /* WITH_HOOKS */
return &nd->signals;
}

73
lib/nodes/CMakeLists.txt
View file

@ -20,35 +20,50 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
###################################################################################
set(NODE_SRC
influxdb.c
stats.c
signal_generator.c
loopback.c
)
set(NODE_SRC)
if(LIBNL3_ROUTE_FOUND)
list(APPEND LIBRARIES PkgConfig::LIBNL3_ROUTE)
list(APPEND INCLUDE_DIRS ${LIBNL3_ROUTE_INCLUDE_DIRS})
endif()
if(WITH_IO)
list(APPEND NODE_SRC
test_rtt.c
file.c
socket.c
)
if(WITH_NODE_INFLUXDB)
list(APPEND NODE_SRC influxdb.c)
endif()
if(WITH_NODE_STATS)
list(APPEND NODE_SRC stats.c)
endif()
if(WITH_NODE_SIGNAL)
list(APPEND NODE_SRC signal_generator.c)
endif()
if(WITH_NODE_LOOPBACK)
list(APPEND NODE_SRC loopback.c)
endif()
if(WITH_NODE_TEST_RTT)
list(APPEND NODE_SRC test_rtt.c)
endif()
if(WITH_NODE_SOCKET)
list(APPEND NODE_SRC socket.c)
endif()
if(WITH_NODE_FILE)
list(APPEND NODE_SRC file.c)
endif()
# Enable Universal Library for Linux DAQ devices (libuldaq)
if(LIBULDAQ_FOUND)
if(WITH_NODE_ULDAQ)
list(APPEND NODE_SRC uldaq.c)
list(APPEND INCLUDE_DIRS ${LIBULDAQ_INCLUDE_DIRS})
list(APPEND LIBRARIES PkgConfig::LIBULDAQ uldaq)
endif()
# Enable shared memory node-type
if(HAS_SEMAPHORE AND HAS_MMAN)
if(WITH_NODE_SHMEM)
list(APPEND NODE_SRC shmem.c)
if(CMAKE_SUSTEM_NAME STREQUAL Linux)
@ -57,75 +72,77 @@ if(HAS_SEMAPHORE AND HAS_MMAN)
endif()
# Enable IEC61850 node-types when libiec61850 is available
if(LIBIEC61850_FOUND)
if(WITH_NODE_IEC61850)
list(APPEND NODE_SRC iec61850_sv.c iec61850.c)
list(APPEND INCLUDE_DIRS ${LIBIEC61850_INCLUDE_DIRS})
list(APPEND LIBRARIES PkgConfig::LIBIEC61850 ${LIBIEC61850_LIBRARIES})
endif()
# Enable OPAL-RT Asynchronous Process support (will result in 32bit binary!!!)
if(OPAL_FOUND AND BUILD32)
if(WITH_NODE_OPAL)
list(APPEND NODE_SRC opal.c)
list(APPEND INCLUDE_DIRS ${OPAL_INCLUDE_DIRS})
list(APPEND LIBRARIES ${OPAL_LIBRARIES})
endif()
# Enable nanomsg node type when libnanomsg is available
if(NANOMSG_FOUND AND WITH_IO)
if(WITH_NODE_NANOMSG)
list(APPEND NODE_SRC nanomsg.c)
list(APPEND INCLUDE_DIRS ${NANOMSG_INCLUDE_DIRS})
list(APPEND LIBRARIES PkgConfig::NANOMSG)
endif()
# Enable ZeroMQ node type when libzmq is available
if(LIBZMQ_FOUND AND WITH_IO)
if(WITH_NODE_ZEROMQ)
list(APPEND NODE_SRC zeromq.c)
list(APPEND INCLUDE_DIRS ${LIBZMQ_INCLUDE_DIRS})
list(APPEND LIBRARIES PkgConfig::LIBZMQ)
endif()
# Enable NGSI support
list(APPEND NODE_SRC ngsi.c)
list(APPEND INCLUDE_DIRS ${CURL_INCLUDE_DIRS})
list(APPEND LIBRARIES ${CURL_LIBRARIES})
if(WITH_NODE_NGSI)
list(APPEND NODE_SRC ngsi.c)
list(APPEND INCLUDE_DIRS ${CURL_INCLUDE_DIRS})
list(APPEND LIBRARIES ${CURL_LIBRARIES})
endif()
# Enable WebSocket support
if(LIBWEBSOCKETS_FOUND AND WITH_WEB AND WITH_IO)
if(WITH_NODE_WEBSOCKET)
list(APPEND NODE_SRC websocket.c)
list(APPEND INCLUDE_DIRS ${LIBWEBSOCKETS_INCLUDE_DIRS})
list(APPEND LIBRARIES ${LIBWEBSOCKETS_LDLIBS})
endif()
# Enable AMQP support
if(RABBITMQ_C_FOUND AND WITH_IO)
if(WITH_NODE_AMQP)
list(APPEND NODE_SRC amqp.c)
list(APPEND INCLUDE_DIRS ${RABBITMQ_C_INCLUDE_DIRS})
list(APPEND LIBRARIES PkgConfig::RABBITMQ_C)
endif()
# Enable MQTT support
if(MOSQUITTO_FOUND AND WITH_IO)
if(WITH_NODE_MQTT)
list(APPEND NODE_SRC mqtt.c)
list(APPEND INCLUDE_DIRS ${MOSQUITTO_INCLUDE_DIRS})
list(APPEND LIBRARIES ${MOSQUITTO_LIBRARIES})
endif()
# Enable Comedi support
if(COMEDILIB_FOUND)
if(WITH_NODE_COMEDI)
list(APPEND NODE_SRC comedi.c)
list(APPEND INCLUDE_DIRS ${COMEDILIB_INCLUDE_DIRS})
list(APPEND LIBRARIES PkgConfig::COMEDILIB)
endif()
# Enable infiniband support
if(IBVERBS_FOUND AND RDMACM_FOUND)
# Enable Infiniband support
if(WITH_NODE_INFINIBAND)
list(APPEND NODE_SRC infiniband.c)
list(APPEND INCLUDE_DIRS ${IBVERBS_INCLUDE_DIRS} ${RDMACM_INCLUDE_DIRS})
list(APPEND LIBRARIES ${IBVERBS_LIBRARIES} ${RDMACM_LIBRARIES})
endif()
# Enable RTP node type when libre is available
if(RE_FOUND AND WITH_IO)
if(WITH_NODE_RTP)
list(APPEND NODE_SRC rtp.cpp)
list(APPEND INCLUDE_DIRS ${RE_INCLUDE_DIRS})
list(APPEND LIBRARIES PkgConfig::RE)

10
lib/nodes/infiniband.c
View file

@ -49,14 +49,14 @@ static int ib_disconnect(struct node *n)
ib->conn.available_recv_wrs -= wcs;
for (int j = 0; j < wcs; j++)
sample_decref((struct sample *) (wc[j].wr_id));
sample_decref((struct sample *) (intptr_t) (wc[j].wr_id));
}
/* Send Queue */
while ((wcs = ibv_poll_cq(ib->ctx.send_cq, ib->send_cq_size, wc)))
for (int j = 0; j < wcs; j++)
if (wc[j].wr_id > 0)
sample_decref((struct sample *) (wc[j].wr_id));
sample_decref((struct sample *) (intptr_t) (wc[j].wr_id));
/* Destroy QP */
rdma_destroy_qp(ib->ctx.id);
@ -253,8 +253,8 @@ int ib_parse(struct node *n, json_t *cfg)
debug(LOG_IB | 4, "Set buffer subtraction to %i in node %s", buffer_subtraction, node_name(n));
/* Translate IP:PORT to a struct addrinfo */
char* ip_adr = strtok_r(local, ":", &lasts);
char* port = strtok_r(NULL, ":", &lasts);
char *ip_adr = strtok_r(local, ":", &lasts);
char *port = strtok_r(NULL, ":", &lasts);
ret = getaddrinfo(ip_adr, port, NULL, &ib->conn.src_addr);
if (ret)
@ -959,7 +959,7 @@ int ib_write(struct node *n, struct sample *smps[], unsigned cnt, unsigned *rele
/* The remaining work requests will be bad. Ripple through list
* and prepare them to be released
*/
debug(LOG_IB | 4, "Bad WR occured with ID: 0x%lx and S/G address: 0x%px: %i",
debug(LOG_IB | 4, "Bad WR occured with ID: 0x%zx and S/G address: 0x%px: %i",
bad_wr->wr_id, bad_wr->sg_list, ret);
while (1) {

12
lib/nodes/nanomsg.c
View file

@ -281,6 +281,15 @@ int nanomsg_poll_fds(struct node *n, int fds[])
return 1;
}
int nanomsg_netem_fds(struct node *n, int fds[])
{
struct nanomsg *m = (struct nanomsg *) n->_vd;
fds[0] = m->out.socket;
return 1;
}
static struct plugin p = {
.name = "nanomsg",
.description = "scalability protocols library (libnanomsg)",
@ -296,7 +305,8 @@ static struct plugin p = {
.stop = nanomsg_stop,
.read = nanomsg_read,
.write = nanomsg_write,
.poll_fds = nanomsg_poll_fds
.poll_fds = nanomsg_poll_fds,
.netem_fds = nanomsg_netem_fds
}
};

4
lib/nodes/rtp.cpp
View file

@ -652,6 +652,7 @@ static void register_plugin() {
p.description = "real-time transport protocol (libre)";
#endif
p.type = PLUGIN_TYPE_NODE;
p.node.instances.state = STATE_DESTROYED;
p.node.vectorize = 0;
p.node.size = sizeof(struct rtp);
p.node.type.start = rtp_type_start;
@ -678,6 +679,3 @@ static void deregister_plugin() {
}
} /* extern C */
REGISTER_PLUGIN(&p)
LIST_INIT_STATIC(&p.node.instances)

205
lib/nodes/socket.c
View file

@ -22,12 +22,9 @@
#include <unistd.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <netdb.h>
#include <netinet/ip.h>
#include <arpa/inet.h>
#include <errno.h>
#include <arpa/inet.h>
#include <netinet/ip.h>
#include <villas/nodes/socket.h>
#include <villas/utils.h>
@ -551,204 +548,6 @@ int socket_parse(struct node *n, json_t *cfg)
return 0;
}
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:
inet_ntop(AF_INET6, &sa->sin6.sin6_addr, buf, 64);
break;
case AF_INET:
inet_ntop(AF_INET, &sa->sin.sin_addr, buf, 64);
break;
#ifdef WITH_SOCKET_LAYER_ETH
case AF_PACKET:
strcatf(&buf, "%02x", sa->sll.sll_addr[0]);
for (int i = 1; i < sa->sll.sll_halen; i++)
strcatf(&buf, ":%02x", sa->sll.sll_addr[i]);
break;
#endif /* WITH_SOCKET_LAYER_ETH */
case AF_UNIX:
strcatf(&buf, "%s", sa->sun.sun_path);
break;
default:
error("Unknown address family: '%u'", sa->sa.sa_family);
}
/* Port / Interface */
switch (sa->sa.sa_family) {
case AF_INET6:
case AF_INET:
strcatf(&buf, ":%hu", ntohs(sa->sin.sin_port));
break;
#ifdef WITH_SOCKET_LAYER_ETH
case AF_PACKET: {
struct nl_cache *cache = nl_cache_mngt_require("route/link");
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;
}
#endif /* WITH_SOCKET_LAYER_ETH */
}
return buf;
}
int socket_parse_address(const char *addr, struct sockaddr *saddr, enum socket_layer layer, int flags)
{
/** @todo: Add support for IPv6 */
union sockaddr_union *sa = (union sockaddr_union *) saddr;
char *copy = strdup(addr);
int ret;
if (layer == SOCKET_LAYER_UNIX) { /* Format: "/path/to/socket" */
sa->sun.sun_family = AF_UNIX;
if (strlen(addr) > sizeof(sa->sun.sun_path) - 1)
error("Length of unix socket path is too long!");
memcpy(sa->sun.sun_path, addr, strlen(addr) + 1);
ret = 0;
}
#ifdef WITH_SOCKET_LAYER_ETH
else if (layer == SOCKET_LAYER_ETH) { /* Format: "ab:cd:ef:12:34:56%ifname:protocol" */
/* Split string */
char *lasts;
char *node = strtok_r(copy, "%", &lasts);
char *ifname = strtok_r(NULL, ":", &lasts);
char *proto = strtok_r(NULL, "\0", &lasts);
/* Parse link layer (MAC) address */
struct ether_addr *mac = ether_aton(node);
if (!mac)
error("Failed to parse MAC address: %s", node);
memcpy(&sa->sll.sll_addr, &mac->ether_addr_octet, ETHER_ADDR_LEN);
/* Get interface index from name */
nl_init();
struct nl_cache *cache = nl_cache_mngt_require("route/link");
struct rtnl_link *link = rtnl_link_get_by_name(cache, ifname);
if (!link)
error("Failed to get network interface: '%s'", ifname);
sa->sll.sll_protocol = htons(proto ? strtol(proto, NULL, 0) : ETH_P_VILLAS);
sa->sll.sll_halen = ETHER_ADDR_LEN;
sa->sll.sll_family = AF_PACKET;
sa->sll.sll_ifindex = rtnl_link_get_ifindex(link);
ret = 0;
}
#endif /* WITH_SOCKET_LAYER_ETH */
else { /* Format: "192.168.0.10:12001" */
struct addrinfo hint = {
.ai_flags = flags,
.ai_family = AF_UNSPEC
};
/* Split string */
char *lasts;
char *node = strtok_r(copy, ":", &lasts);
char *service = strtok_r(NULL, "\0", &lasts);
if (node && !strcmp(node, "*"))
node = NULL;
if (service && !strcmp(service, "*"))
service = NULL;
switch (layer) {
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 SOCKET_LAYER_UDP:
hint.ai_socktype = SOCK_DGRAM;
hint.ai_protocol = IPPROTO_UDP;
break;
default:
error("Invalid address type");
}
/* Lookup address */
struct addrinfo *result;
ret = getaddrinfo(node, (layer == SOCKET_LAYER_IP) ? NULL : service, &hint, &result);
if (!ret) {
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);
}
}
free(copy);
return ret;
}
int socket_compare_addr(struct sockaddr *x, struct sockaddr *y)
{
#define CMP(a, b) if (a != b) return a < b ? -1 : 1
union sockaddr_union *xu = (void *) x, *yu = (void *) y;
CMP(x->sa_family, y->sa_family);
switch (x->sa_family) {
case AF_UNIX:
return strcmp(xu->sun.sun_path, yu->sun.sun_path);
case AF_INET:
CMP(ntohl(xu->sin.sin_addr.s_addr), ntohl(yu->sin.sin_addr.s_addr));
CMP(ntohs(xu->sin.sin_port), ntohs(yu->sin.sin_port));
return 0;
case AF_INET6:
CMP(ntohs(xu->sin6.sin6_port), ntohs(yu->sin6.sin6_port));
// CMP(xu->sin6.sin6_flowinfo, yu->sin6.sin6_flowinfo);
// CMP(xu->sin6.sin6_scope_id, yu->sin6.sin6_scope_id);
return memcmp(xu->sin6.sin6_addr.s6_addr, yu->sin6.sin6_addr.s6_addr, sizeof(xu->sin6.sin6_addr.s6_addr));
#ifdef WITH_SOCKET_LAYER_ETH
case AF_PACKET:
CMP(ntohs(xu->sll.sll_protocol), ntohs(yu->sll.sll_protocol));
CMP(xu->sll.sll_ifindex, yu->sll.sll_ifindex);
// CMP(xu->sll.sll_pkttype, yu->sll.sll_pkttype);
// CMP(xu->sll.sll_hatype, yu->sll.sll_hatype);
CMP(xu->sll.sll_halen, yu->sll.sll_halen);
return memcmp(xu->sll.sll_addr, yu->sll.sll_addr, xu->sll.sll_halen);
#endif /* WITH_SOCKET_LAYER_ETH */
default:
return -1;
}
#undef CMP
}
int socket_fds(struct node *n, int fds[])
{
struct socket *s = (struct socket *) n->_vd;

20
lib/nodes/zeromq.c
View file

@ -542,6 +542,23 @@ int zeromq_poll_fds(struct node *n, int fds[])
return 1;
}
int zeromq_netem_fds(struct node *n, int fds[])
{
int ret;
struct zeromq *z = (struct zeromq *) n->_vd;
int fd;
size_t len = sizeof(fd);
ret = zmq_getsockopt(z->out.socket, ZMQ_FD, &fd, &len);
if (ret)
return ret;
fds[0] = fd;
return 1;
}
static struct plugin p = {
.name = "zeromq",
.description = "ZeroMQ Distributed Messaging (libzmq)",
@ -559,7 +576,8 @@ static struct plugin p = {
.destroy = zeromq_destroy,
.read = zeromq_read,
.write = zeromq_write,
.poll_fds = zeromq_poll_fds
.poll_fds = zeromq_poll_fds,
.netem_fds = zeromq_netem_fds,
}
};

15
lib/path.c
View file

@ -125,9 +125,11 @@ int path_init(struct path *p)
if (ret)
return ret;
#ifdef WITH_HOOKS
ret = hook_list_init(&p->hooks);
if (ret)
return ret;
#endif /* WITH_HOOKS */
p->_name = NULL;
@ -140,7 +142,7 @@ int path_init(struct path *p)
p->enabled = 1;
p->poll = -1;
p->queuelen = DEFAULT_QUEUE_LENGTH;
p->original_sequence_no = 0;
p->original_sequence_no = -1;
p->state = STATE_INITIALIZED;
@ -288,6 +290,9 @@ int path_prepare(struct path *p)
return ret;
}
if (p->original_sequence_no == -1)
p->original_sequence_no = vlist_length(&p->sources) == 1;
p->state = STATE_PREPARED;
return 0;
@ -541,9 +546,12 @@ int path_start(struct path *p)
mask = bitset_dump(&p->mask);
info("Starting path %s: #signals=%zu, mode=%s, poll=%s, mask=%s, rate=%.2f, enabled=%s, reversed=%s, queuelen=%d, #hooks=%zu, #sources=%zu, #destinations=%zu, original_sequence_no=%s",
info("Starting path %s: #signals=%zu, #hooks=%zu, #sources=%zu, #destinations=%zu, mode=%s, poll=%s, mask=%s, rate=%.2f, enabled=%s, reversed=%s, queuelen=%d, original_sequence_no=%s",
path_name(p),
vlist_length(&p->signals),
vlist_length(&p->hooks),
vlist_length(&p->sources),
vlist_length(&p->destinations),
mode,
p->poll ? "yes" : "no",
mask,
@ -551,9 +559,6 @@ int path_start(struct path *p)
path_is_enabled(p) ? "yes" : "no",
path_is_reversed(p) ? "yes" : "no",
p->queuelen,
vlist_length(&p->hooks),
vlist_length(&p->sources),
vlist_length(&p->destinations),
p->original_sequence_no ? "yes" : "no"
);

6
lib/path_source.c
View file

@ -62,7 +62,7 @@ int path_source_destroy(struct path_source *ps)
int path_source_read(struct path_source *ps, struct path *p, int i)
{
int recv, tomux, allocated, cnt, toenqueue, enqueued = 0;
int ret, recv, tomux, allocated, cnt, toenqueue, enqueued = 0;
unsigned release;
cnt = ps->node->in.vectorize;
@ -123,7 +123,9 @@ int path_source_read(struct path_source *ps, struct path *p, int i)
muxed_smps[i]->ts = tomux_smps[i]->ts;
muxed_smps[i]->flags |= tomux_smps[i]->flags & (SAMPLE_HAS_TS_ORIGIN | SAMPLE_HAS_TS_RECEIVED);
mapping_list_remap(&ps->mappings, muxed_smps[i], tomux_smps[i]);
ret = mapping_list_remap(&ps->mappings, muxed_smps[i], tomux_smps[i]);
if (ret)
return ret;
}
sample_copy(p->last_sample, muxed_smps[tomux-1]);

228
lib/socket_addr.c Normal file
View file

@ -0,0 +1,228 @@
/** Various functions to work with socket addresses
*
* @author Steffen Vogel <stvogel@eonerc.rwth-aachen.de>
* @copyright 2014-2019, Institute for Automation of Complex Power Systems, EONERC
* @license GNU General Public License (version 3)
*
* VILLASnode
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <netdb.h>
#include <stdlib.h>
#include <string.h>
#include <arpa/inet.h>
#include <villas/socket_addr.h>
#include <villas/utils.h>
#include <villas/kernel/nl.h>
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:
inet_ntop(AF_INET6, &sa->sin6.sin6_addr, buf, 64);
break;
case AF_INET:
inet_ntop(AF_INET, &sa->sin.sin_addr, buf, 64);
break;
#ifdef WITH_SOCKET_LAYER_ETH
case AF_PACKET:
strcatf(&buf, "%02x", sa->sll.sll_addr[0]);
for (int i = 1; i < sa->sll.sll_halen; i++)
strcatf(&buf, ":%02x", sa->sll.sll_addr[i]);
break;
#endif /* WITH_SOCKET_LAYER_ETH */
case AF_UNIX:
strcatf(&buf, "%s", sa->sun.sun_path);
break;
default:
error("Unknown address family: '%u'", sa->sa.sa_family);
}
/* Port / Interface */
switch (sa->sa.sa_family) {
case AF_INET6:
case AF_INET:
strcatf(&buf, ":%hu", ntohs(sa->sin.sin_port));
break;
#ifdef WITH_SOCKET_LAYER_ETH
case AF_PACKET: {
struct nl_cache *cache = nl_cache_mngt_require("route/link");
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;
}
#endif /* WITH_SOCKET_LAYER_ETH */
}
return buf;
}
int socket_parse_address(const char *addr, struct sockaddr *saddr, enum socket_layer layer, int flags)
{
/** @todo: Add support for IPv6 */
union sockaddr_union *sa = (union sockaddr_union *) saddr;
char *copy = strdup(addr);
int ret;
if (layer == SOCKET_LAYER_UNIX) { /* Format: "/path/to/socket" */
sa->sun.sun_family = AF_UNIX;
if (strlen(addr) > sizeof(sa->sun.sun_path) - 1)
error("Length of unix socket path is too long!");
memcpy(sa->sun.sun_path, addr, strlen(addr) + 1);
ret = 0;
}
#ifdef WITH_SOCKET_LAYER_ETH
else if (layer == SOCKET_LAYER_ETH) { /* Format: "ab:cd:ef:12:34:56%ifname:protocol" */
/* Split string */
char *lasts;
char *node = strtok_r(copy, "%", &lasts);
char *ifname = strtok_r(NULL, ":", &lasts);
char *proto = strtok_r(NULL, "\0", &lasts);
/* Parse link layer (MAC) address */
struct ether_addr *mac = ether_aton(node);
if (!mac)
error("Failed to parse MAC address: %s", node);
memcpy(&sa->sll.sll_addr, &mac->ether_addr_octet, ETHER_ADDR_LEN);
/* Get interface index from name */
nl_init();
struct nl_cache *cache = nl_cache_mngt_require("route/link");
struct rtnl_link *link = rtnl_link_get_by_name(cache, ifname);
if (!link)
error("Failed to get network interface: '%s'", ifname);
sa->sll.sll_protocol = htons(proto ? strtol(proto, NULL, 0) : ETH_P_VILLAS);
sa->sll.sll_halen = ETHER_ADDR_LEN;
sa->sll.sll_family = AF_PACKET;
sa->sll.sll_ifindex = rtnl_link_get_ifindex(link);
ret = 0;
}
#endif /* WITH_SOCKET_LAYER_ETH */
else { /* Format: "192.168.0.10:12001" */
struct addrinfo hint = {
.ai_flags = flags,
.ai_family = AF_UNSPEC
};
/* Split string */
char *lasts;
char *node = strtok_r(copy, ":", &lasts);
char *service = strtok_r(NULL, "\0", &lasts);
if (node && !strcmp(node, "*"))
node = NULL;
if (service && !strcmp(service, "*"))
service = NULL;
switch (layer) {
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 SOCKET_LAYER_UDP:
hint.ai_socktype = SOCK_DGRAM;
hint.ai_protocol = IPPROTO_UDP;
break;
default:
error("Invalid address type");
}
/* Lookup address */
struct addrinfo *result;
ret = getaddrinfo(node, (layer == SOCKET_LAYER_IP) ? NULL : service, &hint, &result);
if (!ret) {
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);
}
}
free(copy);
return ret;
}
int socket_compare_addr(struct sockaddr *x, struct sockaddr *y)
{
#define CMP(a, b) if (a != b) return a < b ? -1 : 1
union sockaddr_union *xu = (void *) x, *yu = (void *) y;
CMP(x->sa_family, y->sa_family);
switch (x->sa_family) {
case AF_UNIX:
return strcmp(xu->sun.sun_path, yu->sun.sun_path);
case AF_INET:
CMP(ntohl(xu->sin.sin_addr.s_addr), ntohl(yu->sin.sin_addr.s_addr));
CMP(ntohs(xu->sin.sin_port), ntohs(yu->sin.sin_port));
return 0;
case AF_INET6:
CMP(ntohs(xu->sin6.sin6_port), ntohs(yu->sin6.sin6_port));
// CMP(xu->sin6.sin6_flowinfo, yu->sin6.sin6_flowinfo);
// CMP(xu->sin6.sin6_scope_id, yu->sin6.sin6_scope_id);
return memcmp(xu->sin6.sin6_addr.s6_addr, yu->sin6.sin6_addr.s6_addr, sizeof(xu->sin6.sin6_addr.s6_addr));
#ifdef WITH_SOCKET_LAYER_ETH
case AF_PACKET:
CMP(ntohs(xu->sll.sll_protocol), ntohs(yu->sll.sll_protocol));
CMP(xu->sll.sll_ifindex, yu->sll.sll_ifindex);
// CMP(xu->sll.sll_pkttype, yu->sll.sll_pkttype);
// CMP(xu->sll.sll_hatype, yu->sll.sll_hatype);
CMP(xu->sll.sll_halen, yu->sll.sll_halen);
return memcmp(xu->sll.sll_addr, yu->sll.sll_addr, xu->sll.sll_halen);
#endif /* WITH_SOCKET_LAYER_ETH */
default:
return -1;
}
#undef CMP
}

31
lib/stats.c
View file

@ -37,6 +37,7 @@ struct stats_metric_description stats_metrics[] = {
{ "gap_sent", STATS_METRIC_GAP_SAMPLE, "seconds", "Inter-message timestamps (as sent by remote)" },
{ "gap_received", STATS_METRIC_GAP_RECEIVED, "seconds", "Inter-message arrival time (as received by this instance)" },
{ "owd", STATS_METRIC_OWD, "seconds", "One-way-delay (OWD) of received messages" },
{ "age", STATS_METRIC_AGE, "seconds", "Processing time of packets within the from receive to sent" },
{ "rtp.loss_fraction", STATS_METRIC_RTP_LOSS_FRACTION, "percent", "Fraction lost since last RTP SR/RR." },
{ "rtp.pkts_lost", STATS_METRIC_RTP_PKTS_LOST, "packets", "Cumulative number of packtes lost" },
{ "rtp.jitter", STATS_METRIC_RTP_JITTER, "seconds", "Interarrival jitter" },
@ -139,13 +140,18 @@ json_t * stats_json_periodic(struct stats *s, struct node *n)
{
assert(s->state == STATE_INITIALIZED);
return json_pack("{ s: s, s: i, s: f, s: f, s: i, s: i }",
return json_pack("{ s: s, s: i, s: i, s: i, s: i, s: f, s: f, s: f, s: f, s: f, s: f }",
"node", node_name(n),
"processed", hist_total(&s->histograms[STATS_METRIC_OWD]),
"owd", hist_last(&s->histograms[STATS_METRIC_OWD]),
"rate", 1.0 / hist_last(&s->histograms[STATS_METRIC_GAP_SAMPLE]),
"recv", hist_total(&s->histograms[STATS_METRIC_OWD]),
"sent", hist_total(&s->histograms[STATS_METRIC_AGE]),
"dropped", hist_total(&s->histograms[STATS_METRIC_SMPS_REORDERED]),
"skipped", hist_total(&s->histograms[STATS_METRIC_SMPS_SKIPPED])
"skipped", hist_total(&s->histograms[STATS_METRIC_SMPS_SKIPPED]),
"owd_last", 1.0 / hist_last(&s->histograms[STATS_METRIC_OWD]),
"owd_mean", 1.0 / hist_mean(&s->histograms[STATS_METRIC_OWD]),
"rate_last", 1.0 / hist_last(&s->histograms[STATS_METRIC_GAP_SAMPLE]),
"rate_mean", 1.0 / hist_mean(&s->histograms[STATS_METRIC_GAP_SAMPLE]),
"age_mean", hist_mean(&s->histograms[STATS_METRIC_AGE]),
"age_max", hist_highest(&s->histograms[STATS_METRIC_AGE])
);
}
@ -160,13 +166,15 @@ void stats_reset(struct stats *s)
static struct table_column stats_cols[] = {
{ 10, "Node", "%s", NULL, TABLE_ALIGN_LEFT },
{ 10, "Recv", "%ju", "pkts", TABLE_ALIGN_RIGHT },
// { 10, "Sent", "%ju", "pkts", TABLE_ALIGN_RIGHT },
{ 10, "Sent", "%ju", "pkts", TABLE_ALIGN_RIGHT },
{ 10, "Drop", "%ju", "pkts", TABLE_ALIGN_RIGHT },
{ 10, "Skip", "%ju", "pkts", TABLE_ALIGN_RIGHT },
{ 10, "OWD last", "%f", "secs", TABLE_ALIGN_RIGHT },
{ 10, "OWD mean", "%f", "secs", TABLE_ALIGN_RIGHT },
{ 10, "Rate last", "%f", "pkt/sec", TABLE_ALIGN_RIGHT },
{ 10, "Rate mean", "%f", "pkt/sec", TABLE_ALIGN_RIGHT },
{ 10, "Drop", "%ju", "pkts", TABLE_ALIGN_RIGHT },
{ 10, "Skip", "%ju", "pkts", TABLE_ALIGN_RIGHT }
{ 10, "Age mean", "%f", "secs", TABLE_ALIGN_RIGHT },
{ 10, "Age Max", "%f", "sec", TABLE_ALIGN_RIGHT },
};
static struct table stats_table = {
@ -194,12 +202,15 @@ void stats_print_periodic(struct stats *s, FILE *f, enum stats_format fmt, int v
table_row(&stats_table,
node_name_short(n),
hist_total(&s->histograms[STATS_METRIC_OWD]),
hist_total(&s->histograms[STATS_METRIC_AGE]),
hist_total(&s->histograms[STATS_METRIC_SMPS_REORDERED]),
hist_total(&s->histograms[STATS_METRIC_SMPS_SKIPPED]),
hist_last(&s->histograms[STATS_METRIC_OWD]),
hist_mean(&s->histograms[STATS_METRIC_OWD]),
1.0 / hist_last(&s->histograms[STATS_METRIC_GAP_RECEIVED]),
1.0 / hist_mean(&s->histograms[STATS_METRIC_GAP_RECEIVED]),
hist_total(&s->histograms[STATS_METRIC_SMPS_REORDERED]),
hist_total(&s->histograms[STATS_METRIC_SMPS_SKIPPED])
hist_mean(&s->histograms[STATS_METRIC_AGE]),
hist_highest(&s->histograms[STATS_METRIC_AGE])
);
break;

7
lib/super_node.cpp
View file

@ -50,13 +50,17 @@ SuperNode::SuperNode() :
idleStop(false),
#ifdef WITH_API
api(this),
#endif
#ifdef WITH_WEB
#ifdef WITH_API
web(&api),
#else
web(),
#endif
#endif
priority(0),
affinity(0),
hugepages(DEFAULT_NR_HUGEPAGES)
#endif
{
nodes.state = STATE_DESTROYED;
paths.state = STATE_DESTROYED;
@ -454,6 +458,7 @@ void SuperNode::stop()
#ifdef WITH_API
api.stop();
#endif
#ifdef WITH_WEB
web.stop();
#endif

4
lib/web.cpp
View file

@ -61,14 +61,14 @@ lws_protocols protocols[] = {
.rx_buffer_size = 0
},
#endif /* WITH_API */
#ifdef LIBWEBSOCKETS_FOUND
#ifdef WITH_NODE_WEBSOCKET
{
.name = "live",
.callback = websocket_protocol_cb,
.per_session_data_size = sizeof(websocket_connection),
.rx_buffer_size = 0
},
#endif /* LIBWEBSOCKETS_FOUND */
#endif /* WITH_NODE_WEBSOCKET */
{
.name = nullptr /* terminator */
}

64
packaging/docker/Dockerfile.app-raspbian Normal file
View file

@ -0,0 +1,64 @@
# Dockerfile
#
# This image can be used for running VILLASnode
# by running:
# make docker
#
# @author Steffen Vogel <stvogel@eonerc.rwth-aachen.de>
# @copyright 2014-2019, Institute for Automation of Complex Power Systems, EONERC
# @license GNU General Public License (version 3)
#
# VILLASnode
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
###################################################################################
ARG BUILDER_IMAGE=villas/node-dev-raspbian
ARG DOCKER_TAG=latest
ARG GIT_REV=unknown
ARG GIT_BRANCH=unknown
ARG VERSION=unknown
ARG VARIANT=unknown
# This image is built by villas-node-git/packaging/docker/Dockerfile.dev
FROM $BUILDER_IMAGE
COPY . /villas/
RUN rm -rf /villas/build && mkdir /villas/build
WORKDIR /villas/build
RUN cmake -DCPACK_GENERATOR=RPM ..
RUN make -j$(nproc) install
# For WebSocket / API access
EXPOSE 80
EXPOSE 443
ENTRYPOINT ["villas"]
LABEL \
org.label-schema.schema-version = "1.0" \
org.label-schema.name = "VILLASnode" \
org.label-schema.license = "GPL-3.0" \
org.label-schema.vcs-ref="$GIT_REV" \
org.label-schema.vcs-branch="$GIT_BRANCH" \
org.label-schema.version="$VERSION" \
org.label-schema.variant="$VARIANT" \
org.label-schema.vendor = "Institute for Automation of Complex Power Systems, RWTH Aachen University" \
org.label-schema.author.name = "Steffen Vogel" \
org.label-schema.author.email = "stvogel@eonerc.rwth-aachen.de" \
org.label-schema.description = "A image containing for VILLASnode based on Fedora" \
org.label-schema.url = "http://fein-aachen.org/projects/villas-framework/" \
org.label-schema.vcs-url = "https://git.rwth-aachen.de/VILLASframework/VILLASnode" \
org.label-schema.usage = "https://villas.fein-aachen.org/doc/node-installation.html#node-installation-docker"

123
packaging/docker/Dockerfile.dev-raspbian Normal file
View file

@ -0,0 +1,123 @@
# Dockerfile for VILLASnode development.
#
# This Dockerfile builds an image which contains all library dependencies
# and tools to build VILLASnode.
# However, VILLASnode itself it not part of the image.
#
# This image can be used for developing VILLASnode
# by running:
# make docker
#
# @author Steffen Vogel <stvogel@eonerc.rwth-aachen.de>
# @copyright 2014-2019, Institute for Automation of Complex Power Systems, EONERC
# @license GNU General Public License (version 3)
#
# VILLASnode
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
###################################################################################
FROM balenalib/rpi-raspbian
ARG GIT_REV=unknown
ARG GIT_BRANCH=unknown
ARG VERSION=unknown
ARG VARIANT=unknown
# Toolchain
RUN apt-get update && apt-get install -y \
gcc g++ \
autoconf automake libtool \
pkg-config cmake make ninja-build \
texinfo git curl tar \
protobuf-compiler protobuf-c-compiler
# Several tools only needed for developement and testing
RUN apt-get install -y \
doxygen dia graphviz \
openssh-client \
jq netcat \
iproute2 \
python-pip \
valgrind gdb gdbserver \
xmlto asciidoctor \
rabbitmq-server mosquitto
# Dependencies
RUN apt-get install -y \
libssl-dev \
libprotobuf-dev \
libprotobuf-c-dev \
uuid-dev \
libconfig-dev \
libnl-3-dev libnl-route-3-dev \
libcurl4-openssl-dev \
libjansson-dev \
libzmq3-dev \
libnanomsg-dev \
librabbitmq-dev \
libmosquitto-dev \
libcomedi-dev \
libibverbs-dev \
librdmacm-dev \
libre-dev \
libusb-1.0-0-dev
# Build & Install libwebsockets
RUN cd /tmp && \
git clone -b v3.1-stable https://github.com/warmcat/libwebsockets && \
mkdir -p libwebsockets/build && cd libwebsockets/build && \
cmake .. && make -j$(nproc) install && \
rm -rf /tmp/*
# Build & Install libiec61850
RUN cd /tmp && \
git clone -b v1.3.1 https://github.com/mz-automation/libiec61850 && \
mkdir -p libiec61850/build && cd libiec61850/build && \
cmake .. && make -j$(nproc) install && \
rm -rf /tmp/*
RUN cd /tmp && \
git clone -b rpm https://github.com/stv0g/uldaq && \
cd uldaq && \
autoreconf -i && ./configure && make -j$(nproc) install && \
rm -rf /tmp/*
# Expose ports for HTTP and WebSocket frontend
EXPOSE 80
EXPOSE 443
ENV LD_LIBRARY_PATH /usr/local/lib:/usr/local/lib64
# Workaround for libnl3's search path for netem distributions
RUN ln -s /usr/lib64/tc /usr/lib/tc
WORKDIR /villas
ENTRYPOINT bash
LABEL \
org.label-schema.schema-version="1.0" \
org.label-schema.name="VILLASnode" \
org.label-schema.license="GPL-3.0" \
org.label-schema.vcs-ref="$GIT_REV" \
org.label-schema.vcs-branch="$GIT_BRANCH" \
org.label-schema.version="$VERSION" \
org.label-schema.variant="$VARIANT" \
org.label-schema.vendor="Institute for Automation of Complex Power Systems, RWTH Aachen University" \
org.label-schema.author.name="Steffen Vogel" \
org.label-schema.author.email="stvogel@eonerc.rwth-aachen.de" \
org.label-schema.description="A image containing all build-time dependencies for VILLASnode based on Ubuntu" \
org.label-schema.url="http://fein-aachen.org/projects/villas-framework/" \
org.label-schema.vcs-url="https://git.rwth-aachen.de/VILLASframework/VILLASnode" \
org.label-schema.usage="https://villas.fein-aachen.org/doc/node-installation.html#node-installation-docker"

30
src/CMakeLists.txt
View file

@ -47,27 +47,25 @@ if(WITH_WEB)
)
endif()
if(WITH_IO)
add_executable(villas-test-cmp villas-test-cmp.cpp)
target_link_libraries(villas-test-cmp PUBLIC villas)
add_executable(villas-test-cmp villas-test-cmp.cpp)
target_link_libraries(villas-test-cmp PUBLIC villas)
add_executable(villas-convert villas-convert.cpp)
target_link_libraries(villas-convert PUBLIC villas)
add_executable(villas-convert villas-convert.cpp)
target_link_libraries(villas-convert PUBLIC villas)
add_executable(villas-pipe villas-pipe.cpp)
target_link_libraries(villas-pipe PUBLIC villas Threads::Threads)
add_executable(villas-pipe villas-pipe.cpp)
target_link_libraries(villas-pipe PUBLIC villas Threads::Threads)
add_executable(villas-signal villas-signal.cpp)
target_link_libraries(villas-signal PUBLIC villas)
add_executable(villas-signal villas-signal.cpp)
target_link_libraries(villas-signal PUBLIC villas)
install(
TARGETS villas-convert villas-pipe villas-signal villas-test-cmp
COMPONENT bin
RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}
)
endif()
install(
TARGETS villas-convert villas-pipe villas-signal villas-test-cmp
COMPONENT bin
RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}
)
if(WITH_IO AND WITH_HOOKS)
if(WITH_HOOKS)
add_executable(villas-hook villas-hook.cpp)
target_link_libraries(villas-hook PUBLIC villas)

2
src/villas-hook.cpp
View file

@ -69,10 +69,12 @@ static void usage()
<< " -h show this help" << std::endl
<< " -V show the version of the tool" << std::endl << std::endl;
#ifdef WITH_HOOKS
std::cout << "Supported hooks:" << std::endl;
for (Plugin *p : Registry::lookup<HookFactory>())
std::cout << " - " << p->getName() << ": " << p->getDescription() << std::endl;
std::cout << std::endl;
#endif /* WITH_HOOKS */
std::cout << "Supported IO formats:" << std::endl;
plugin_dump(PLUGIN_TYPE_FORMAT);

12
src/villas-node.cpp
View file

@ -116,6 +116,7 @@ static void usage()
int main(int argc, char *argv[])
{
int ret;
const char *uri;
Logger logger = logging.get("node");
@ -129,7 +130,7 @@ int main(int argc, char *argv[])
opal_register_region(argc, argv);
const char *uri = "opal-shmem.conf";
uri = "opal-shmem.conf";
#else
/* Parse optional command line arguments */
@ -153,7 +154,14 @@ int main(int argc, char *argv[])
continue;
}
char *uri = argc == optind + 1 ? argv[optind] : nullptr;
if (argc == optind + 1)
uri = argv[optind];
else if (argc == optind)
uri = nullptr;
else {
usage();
exit(EXIT_FAILURE);
}
#endif /* ENABLE_OPAL_ASYNC */
logger->info("This is VILLASnode {} (built on {}, {})",

7
src/villas-pipe.cpp
View file

@ -374,16 +374,13 @@ check: if (optarg == endptr)
if (!node)
throw RuntimeError("Node {} does not exist!", nodestr);
#ifdef LIBWEBSOCKETS_FOUND
#ifdef WITH_NODE_WEBSOCKET
/* Only start web subsystem if villas-pipe is used with a websocket node */
if (node_type(node)->start == websocket_start) {
Web *w = sn.getWeb();
Api *a = sn.getApi();
w->start();
a->start();
}
#endif /* LIBWEBSOCKETS_FOUND */
#endif /* WITH_NODE_WEBSOCKET */
if (reverse)
node_reverse(node);

1
src/villas-relay.cpp
View file

@ -32,6 +32,7 @@
#include <jansson.h>
#include <villas/config.h>
#include <villas/compat.h>
#include <villas/log.hpp>
#include <villas/copyright.hpp>

4
src/villas-relay.hpp
View file

@ -27,6 +27,10 @@
#include <spdlog/spdlog.h>
#include <uuid/uuid.h>
#ifndef UUID_STR_LEN
#define UUID_STR_LEN 37
#endif
#include <libwebsockets.h>
/* Forward declarations */

18
tests/CMakeLists.txt
View file

@ -28,12 +28,14 @@ if(CRITERION_FOUND)
add_subdirectory(unit)
endif()
set(VALGRIND "valgrind --leak-check=full --show-leak-kinds=all --suppressions=${CMAKE_CURRENT_SOURCE_DIR}/valgrind.supp")
if(WITH_SRC AND WITH_HOOKS)
set(VALGRIND "valgrind --leak-check=full --show-leak-kinds=all --suppressions=${CMAKE_CURRENT_SOURCE_DIR}/valgrind.supp")
add_custom_target(run-valgrind
COMMAND ${VALGRIND} $<TARGET_FILE:villas-node> & sleep 2; kill %1
COMMAND ${VALGRIND} $<TARGET_FILE:villas-pipe> -t 2 ${PROJECT_SOURCE_DIR}/etc/websocket-loopback.conf ws1
COMMAND ${VALGRIND} $<TARGET_FILE:villas-signal> mixed -v 4 -l 10
COMMAND ${VALGRIND} $<TARGET_FILE:villas-hook> stats < <($<TARGET_FILE:villas-signal> mixed -l 5)
)
add_dependencies(run-valgrind villas-node villas-pipe villas-signal villas-hook)
add_custom_target(run-valgrind
COMMAND ${VALGRIND} $<TARGET_FILE:villas-node> & sleep 2; kill %1
COMMAND ${VALGRIND} $<TARGET_FILE:villas-pipe> -t 2 ${PROJECT_SOURCE_DIR}/etc/websocket-loopback.conf ws1
COMMAND ${VALGRIND} $<TARGET_FILE:villas-signal> mixed -v 4 -l 10
COMMAND ${VALGRIND} $<TARGET_FILE:villas-hook> stats < <($<TARGET_FILE:villas-signal> mixed -l 5)
)
add_dependencies(run-valgrind villas-node villas-pipe villas-signal villas-hook)
endif()

4
tests/unit/config_json.cpp
View file

@ -22,7 +22,7 @@
#include <criterion/criterion.h>
#ifdef LIBCONFIG_FOUND
#ifdef WITH_CONFIG
#include <jansson.h>
#include <libconfig.h>
@ -117,4 +117,4 @@ Test(utils, json_to_config)
json_decref(json);
}
#endif /* LIBCONFIG_FOUND */
#endif /* WITH_CONFIG */

2
tests/unit/io.cpp
View file

@ -62,7 +62,7 @@ static struct param params[] = {
{ "villas.binary", 10, 0 },
{ "csv", 10, 0 },
{ "json", 10, 0 },
#ifdef LIBPROTOBUF_FOUND
#ifdef PROTOBUF_FOUND
{ "protobuf", 10, 0 }
#endif
};

2
tools/CMakeLists.txt
View file

@ -20,7 +20,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
###################################################################################
if(LIBCONFIG_FOUND)
if(WITH_CONFIG)
add_executable(conf2json conf2json.cpp)
target_link_libraries(conf2json PUBLIC villas)