aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--INSTALL7
-rw-r--r--configure.ac19
-rw-r--r--doc/example.ini25
-rw-r--r--src/DabMod.cpp64
-rw-r--r--src/DabModulator.cpp8
-rw-r--r--src/DabModulator.h4
-rw-r--r--src/InputReader.h4
-rw-r--r--src/InputZeroMQReader.cpp2
-rw-r--r--src/OutputUHD.cpp52
-rw-r--r--src/OutputUHD.h5
-rw-r--r--src/OutputZeroMQ.cpp4
-rw-r--r--src/OutputZeroMQ.h4
-rw-r--r--src/RemoteControl.cpp146
-rw-r--r--src/RemoteControl.h127
14 files changed, 411 insertions, 60 deletions
diff --git a/INSTALL b/INSTALL
index 55385ba..bb14565 100644
--- a/INSTALL
+++ b/INSTALL
@@ -2,8 +2,8 @@ Required dependencies:
======================
* Boost 1.41 or later
+ * Optional UHD for USRP
* Optional ZeroMQ http://www.zeromq.org
- Use --disable-input-zeromq if you don't have it
* Optional FFTW 3.x (included KISS FFT is used as fallback)
Simple install procedure:
@@ -11,7 +11,7 @@ Simple install procedure:
% tar xjf odr-dabmod-X.Y.Z.tar.bz2 # Unpack the source
% cd odr-dabmod-X.Y.Z # Change to the source directory
- % ./configure --disable-debug --with-debug-malloc=yes
+ % ./configure --enable-zeromq --enable-output-uhd
# Run the configure script
% make # Build ODR-DabMod
[ as root ]
@@ -21,7 +21,8 @@ Configure options
=================
The configure script can be launch with a variety of options:
- --enable-input-zeromq Enable ZeroMQ input (to be used with ODR-DabMux)
+ --enable-zeromq Enable ZeroMQ input (to be used with ODR-DabMux),
+ output and remotecontrol.
--enable-output-uhd Includes the binding to the UHD driver for USRPs
You have the choice between two FFT libraries: KISS FFT and FFTW. KISS FFT is a
diff --git a/configure.ac b/configure.ac
index 615bf6d..739f54a 100644
--- a/configure.ac
+++ b/configure.ac
@@ -71,12 +71,8 @@ AC_ARG_ENABLE([kiss_fft],
[], [enable_kiss=no])
# ZeroMQ message queue input
-AC_ARG_ENABLE([input_zeromq],
- AS_HELP_STRING([--enable-input-zeromq], [Enable ZeroMQ input]))
-
-# ZeroMQ message IQ output
-AC_ARG_ENABLE([output_zeromq],
- AS_HELP_STRING([--enable-output-zeromq], [Enable ZeroMQ output]))
+AC_ARG_ENABLE([zeromq],
+ AS_HELP_STRING([--enable-zeromq], [Enable ZeroMQ input, output and remote control]))
# UHD support control
AC_ARG_ENABLE([output_uhd],
@@ -91,13 +87,10 @@ AS_IF([test "x$enable_fftw" = "xyes"],
AC_MSG_NOTICE([Found FFTW3]),
AC_MSG_NOTICE([Using Kiss FFT]) )
-echo "Checking input zeromq"
+echo "Checking zeromq"
-AS_IF([test "x$enable_input_zeromq" = "xyes"],
- [AC_DEFINE(HAVE_INPUT_ZEROMQ, [1], [Define if ZeroMQ input is enabled]) ,
- AC_CHECK_LIB(zmq, zmq_init, ZMQ_LIBS="-lzmq" ,[AC_MSG_ERROR([ZeroMQ libzmq is required])])])
-AS_IF([test "x$enable_output_zeromq" = "xyes"],
- [AC_DEFINE(HAVE_OUTPUT_ZEROMQ, [1], [Define if ZeroMQ output is enabled]) ,
+AS_IF([test "x$enable_zeromq" = "xyes"],
+ [AC_DEFINE(HAVE_ZEROMQ, [1], [Define if ZeroMQ is enabled]) ,
AC_CHECK_LIB(zmq, zmq_init, ZMQ_LIBS="-lzmq" ,[AC_MSG_ERROR([ZeroMQ libzmq is required])])])
AS_IF([test "x$enable_debug" = "xno"],
[OPTIM="-O2" DEBUG="" EXTRA="$EXTRA -DNDEBUG"],
@@ -203,7 +196,7 @@ echo "***********************************************"
echo
enabled=""
disabled=""
-for feat in debug prof trace fftw fft_simd output_uhd input_zeromq output_zeromq
+for feat in debug prof trace fftw fft_simd output_uhd zeromq
do
eval var=\$enable_$feat
AS_IF([test "x$var" = "xyes"],
diff --git a/doc/example.ini b/doc/example.ini
index ada7a03..43adacb 100644
--- a/doc/example.ini
+++ b/doc/example.ini
@@ -8,6 +8,29 @@
telnet=1
telnetport=2121
+; Enable zmq remote control.
+; The zmq remote control is intended for machine-to-machine
+; integration and requires that ODR-DabMod is built with zmq support.
+; The zmq remote control may run in parallel with Telnet.
+;
+; Protocol:
+; ODR-DabMod binds a zmq rep socket so clients must connect
+; using either req or dealer socket.
+; [] denotes message part as zmq multi-part message are used for delimitation.
+; All message parts are utf-8 encoded strings and match the Telnet command set.
+; Explicit codes are denoted with "".
+; The following commands are supported:
+; REQ: ["ping"]
+; REP: ["ok"]
+;
+; REQ: ["get"][module name][parameter]
+; REP: [value] _OR_ ["fail"][error description]
+;
+; REQ: ["set"][module name][parameter][value]
+; REP: ["ok"] _OR_ ["fail"][error description]
+zmqctrl=1
+zmqctrlendpoint=tcp://127.0.0.1:9400
+
[log]
; Write to a logfile or to syslog.
; Setting filename to stderr is very useful during tests and development
@@ -59,7 +82,7 @@ gainmode=2
; and wide-band noise will be generated.
;
; Be aware that there is a dependency with resampling.
-digital_gain=1.0
+digital_gain=0.8
; Output sample rate. Values other than 2048000 enable
; resampling.
diff --git a/src/DabMod.cpp b/src/DabMod.cpp
index dc61ae2..82f03e5 100644
--- a/src/DabMod.cpp
+++ b/src/DabMod.cpp
@@ -72,13 +72,13 @@
typedef std::complex<float> complexf;
-bool running = true;
+volatile sig_atomic_t running = 1;
void signalHandler(int signalNb)
{
PDEBUG("signalHandler(%i)\n", signalNb);
- running = false;
+ running = 0;
}
@@ -223,16 +223,23 @@ int main(int argc, char* argv[])
FormatConverter* format_converter = NULL;
ModOutput* output = NULL;
- BaseRemoteController* rc = NULL;
+ RemoteControllers rcs;
Logger logger;
InputFileReader inputFileReader(logger);
-#if defined(HAVE_INPUT_ZEROMQ)
+#if defined(HAVE_ZEROMQ)
InputZeroMQReader inputZeroMQReader(logger);
#endif
InputReader* inputReader;
- signal(SIGINT, signalHandler);
+ struct sigaction sa;
+ memset(&sa, 0, sizeof(struct sigaction));
+ sa.sa_handler = &signalHandler;
+
+ if (sigaction(SIGINT, &sa, NULL) == -1) {
+ perror("sigaction");
+ return EXIT_FAILURE;
+ }
// Set timezone to UTC
setenv("TZ", "", 1);
@@ -366,15 +373,12 @@ int main(int argc, char* argv[])
"\n";
std::cerr << "Compiled with features: " <<
-#if defined(HAVE_INPUT_ZEROMQ)
- "input_zeromq " <<
+#if defined(HAVE_ZEROMQ)
+ "zeromq " <<
#endif
#if defined(HAVE_OUTPUT_UHD)
"output_uhd " <<
#endif
-#if defined(HAVE_OUTPUT_ZEROMQ)
- "output_zeromq " <<
-#endif
"\n";
if (use_configuration_file && use_configuration_cmdline) {
@@ -413,7 +417,7 @@ int main(int argc, char* argv[])
try {
int telnetport = pt.get<int>("remotecontrol.telnetport");
RemoteControllerTelnet* telnetrc = new RemoteControllerTelnet(telnetport);
- rc = telnetrc;
+ rcs.add_controller(telnetrc);
}
catch (std::exception &e) {
std::cerr << "Error: " << e.what() << "\n";
@@ -422,6 +426,22 @@ int main(int argc, char* argv[])
}
}
+#if defined(HAVE_ZEROMQ)
+ if (pt.get("remotecontrol.zmqctrl", 0) == 1) {
+ try {
+ std::string zmqCtrlEndpoint = pt.get("remotecontrol.zmqctrlendpoint", "");
+ std::cerr << "ZmqCtrlEndpoint: " << zmqCtrlEndpoint << std::endl;
+ RemoteControllerZmq* zmqrc = new RemoteControllerZmq(zmqCtrlEndpoint);
+ rcs.add_controller(zmqrc);
+ }
+ catch (std::exception &e) {
+ std::cerr << "Error: " << e.what() << "\n";
+ std::cerr << " zmq remote control enabled, but no endpoint defined.\n";
+ goto END_MAIN;
+ }
+ }
+#endif
+
// input params:
if (pt.get("input.loop", 0) == 1) {
loop = true;
@@ -592,7 +612,7 @@ int main(int argc, char* argv[])
useUHDOutput = 1;
}
#endif
-#if defined(HAVE_OUTPUT_ZEROMQ)
+#if defined(HAVE_ZEROMQ)
else if (output_selected == "zmq") {
outputName = pt.get<std::string>("zmqoutput.listen");
useZeroMQOutput = 1;
@@ -631,9 +651,9 @@ int main(int argc, char* argv[])
#endif
}
- if (!rc) {
+ if (rcs.get_no_controllers() == 0) {
logger.level(warn) << "No Remote-Control started";
- rc = new RemoteControllerDummy();
+ rcs.add_controller(new RemoteControllerDummy());
}
@@ -735,7 +755,7 @@ int main(int argc, char* argv[])
inputReader = &inputFileReader;
}
else if (inputTransport == "zeromq") {
-#if !defined(HAVE_INPUT_ZEROMQ)
+#if !defined(HAVE_ZEROMQ)
fprintf(stderr, "Error, ZeroMQ input transport selected, but not compiled in!\n");
ret = -1;
goto END_MAIN;
@@ -778,7 +798,7 @@ int main(int argc, char* argv[])
outputuhd_conf.sampleRate = outputRate;
try {
output = new OutputUHD(outputuhd_conf, logger);
- ((OutputUHD*)output)->enrol_at(*rc);
+ ((OutputUHD*)output)->enrol_at(rcs);
}
catch (std::exception& e) {
logger.level(error) << "UHD initialisation failed:" << e.what();
@@ -786,7 +806,7 @@ int main(int argc, char* argv[])
}
}
#endif
-#if defined(HAVE_OUTPUT_ZEROMQ)
+#if defined(HAVE_ZEROMQ)
else if (useZeroMQOutput) {
/* We normalise the same way as for the UHD output */
normalise = 1.0f / normalise_factor;
@@ -798,7 +818,7 @@ int main(int argc, char* argv[])
flowgraph = new Flowgraph();
data.setLength(6144);
input = new InputMemory(&data);
- modulator = new DabModulator(modconf, rc, logger, outputRate, clockRate,
+ modulator = new DabModulator(modconf, &rcs, logger, outputRate, clockRate,
dabMode, gainMode, digitalgain, normalise, filterTapsFilename);
flowgraph->connect(input, modulator);
if (format_converter) {
@@ -843,10 +863,8 @@ int main(int argc, char* argv[])
/* Check every once in a while if the remote control
* is still working */
- if (rc && (frame % 250) == 0 && rc->fault_detected()) {
- fprintf(stderr,
- "Detected Remote Control fault, restarting it\n");
- rc->restart();
+ if (rcs.get_no_controllers() > 0 && (frame % 250) == 0) {
+ rcs.check_faults();
}
}
if (framesize == 0) {
@@ -855,7 +873,7 @@ int main(int argc, char* argv[])
else {
fprintf(stderr, "Input read error.\n");
}
- running = false;
+ running = 0;
}
} catch (std::exception& e) {
fprintf(stderr, "EXCEPTION: %s\n", e.what());
diff --git a/src/DabModulator.cpp b/src/DabModulator.cpp
index 7f246d8..2664a08 100644
--- a/src/DabModulator.cpp
+++ b/src/DabModulator.cpp
@@ -53,7 +53,7 @@
DabModulator::DabModulator(
struct modulator_offset_config& modconf,
- BaseRemoteController* rc,
+ RemoteControllers* rcs,
Logger& logger,
unsigned outputRate, unsigned clockRate,
unsigned dabMode, GainMode gainMode,
@@ -71,7 +71,7 @@ DabModulator::DabModulator(
myEtiReader(EtiReader(modconf, myLogger)),
myFlowgraph(NULL),
myFilterTapsFilename(filterTapsFilename),
- myRC(rc)
+ myRCs(rcs)
{
PDEBUG("DabModulator::DabModulator(%u, %u, %u, %u) @ %p\n",
outputRate, clockRate, dabMode, gainMode, this);
@@ -201,13 +201,13 @@ int DabModulator::process(Buffer* const dataIn, Buffer* dataOut)
cifOfdm = new OfdmGenerator((1 + myNbSymbols), myNbCarriers, mySpacing);
cifGain = new GainControl(mySpacing, myGainMode, myDigGain, myNormalise);
- cifGain->enrol_at(*myRC);
+ cifGain->enrol_at(*myRCs);
cifGuard = new GuardIntervalInserter(myNbSymbols, mySpacing,
myNullSize, mySymSize);
if (myFilterTapsFilename != "") {
cifFilter = new FIRFilter(myFilterTapsFilename);
- cifFilter->enrol_at(*myRC);
+ cifFilter->enrol_at(*myRCs);
}
myOutput = new OutputMemory();
diff --git a/src/DabModulator.h b/src/DabModulator.h
index 21f9f61..84c9926 100644
--- a/src/DabModulator.h
+++ b/src/DabModulator.h
@@ -47,7 +47,7 @@ class DabModulator : public ModCodec
public:
DabModulator(
struct modulator_offset_config& modconf,
- BaseRemoteController* rc,
+ RemoteControllers* rcs,
Logger& logger,
unsigned outputRate = 2048000, unsigned clockRate = 0,
unsigned dabMode = 0, GainMode gainMode = GAIN_VAR,
@@ -77,7 +77,7 @@ protected:
Flowgraph* myFlowgraph;
OutputMemory* myOutput;
std::string myFilterTapsFilename;
- BaseRemoteController* myRC;
+ RemoteControllers* myRCs;
size_t myNbSymbols;
size_t myNbCarriers;
diff --git a/src/InputReader.h b/src/InputReader.h
index 164c5ac..3e0dcab 100644
--- a/src/InputReader.h
+++ b/src/InputReader.h
@@ -31,7 +31,7 @@
#endif
#include <cstdio>
-#if defined(HAVE_INPUT_ZEROMQ)
+#if defined(HAVE_ZEROMQ)
# include "zmq.hpp"
# include "ThreadsafeQueue.h"
#endif
@@ -130,7 +130,7 @@ class InputFileReader : public InputReader
// after 2**32 * 24ms ~= 3.3 years
};
-#if defined(HAVE_INPUT_ZEROMQ)
+#if defined(HAVE_ZEROMQ)
/* A ZeroMQ input. See www.zeromq.org for more info */
struct InputZeroMQThreadData
diff --git a/src/InputZeroMQReader.cpp b/src/InputZeroMQReader.cpp
index cfb56b2..f7f5702 100644
--- a/src/InputZeroMQReader.cpp
+++ b/src/InputZeroMQReader.cpp
@@ -29,7 +29,7 @@
# include "config.h"
#endif
-#if defined(HAVE_INPUT_ZEROMQ)
+#if defined(HAVE_ZEROMQ)
#include <string>
#include <cstring>
diff --git a/src/OutputUHD.cpp b/src/OutputUHD.cpp
index 8063e75..741731e 100644
--- a/src/OutputUHD.cpp
+++ b/src/OutputUHD.cpp
@@ -56,10 +56,12 @@ OutputUHD::OutputUHD(
// Since we don't know the buffer size, we cannot initialise
// the buffers at object initialisation.
first_run(true),
- activebuffer(1)
+ activebuffer(1),
+ myDelayBuf(196608)
{
myMuting = 0; // is remote-controllable
+ myStaticDelay = 0; // is remote-controllable
#if FAKE_UHD
MDEBUG("OutputUHD:Using fake UHD output");
@@ -87,7 +89,9 @@ OutputUHD::OutputUHD(
/* register the parameters that can be remote controlled */
RC_ADD_PARAMETER(txgain, "UHD analog daughterboard TX gain");
RC_ADD_PARAMETER(freq, "UHD transmission frequency");
- RC_ADD_PARAMETER(muting, "mute the output by stopping the transmitter");
+ RC_ADD_PARAMETER(muting, "Mute the output by stopping the transmitter");
+ RC_ADD_PARAMETER(staticdelay, "Set static delay (uS) between 0 and 96000");
+ RC_ADD_PARAMETER(iqbalance, "Set I/Q balance between 0 and 1.0");
uhd::set_thread_priority_safe();
@@ -236,6 +240,10 @@ OutputUHD::~OutputUHD()
{
MDEBUG("OutputUHD::~OutputUHD() @ %p\n", this);
worker.stop();
+ if (!first_run) {
+ free(uwd.frame0.buf);
+ free(uwd.frame1.buf);
+ }
}
int OutputUHD::process(Buffer* dataIn, Buffer* dataOut)
@@ -298,13 +306,30 @@ int OutputUHD::process(Buffer* dataIn, Buffer* dataOut)
uwd.sourceContainsTimestamp = myConf.enableSync &&
myEtiReader->sourceContainsTimestamp();
+ // calculate delay
+ uint32_t noSampleDelay = (myStaticDelay * 2048) / 1000;
+ uint32_t noByteDelay = noSampleDelay * sizeof(complexf);
+
+ uint8_t* pInData = (uint8_t*) dataIn->getData();
if (activebuffer == 0) {
- memcpy(uwd.frame0.buf, dataIn->getData(), uwd.bufsize);
+ uint8_t *pTmp = (uint8_t*) uwd.frame0.buf;
+ // copy remain from delaybuf
+ memcpy(pTmp, &myDelayBuf[0], noByteDelay);
+ // copy new data
+ memcpy(&pTmp[noByteDelay], pInData, uwd.bufsize - noByteDelay);
+ // copy remaining data to delay buf
+ memcpy(&myDelayBuf[0], &pInData[uwd.bufsize - noByteDelay], noByteDelay);
uwd.frame0.ts = ts;
}
else if (activebuffer == 1) {
- memcpy(uwd.frame1.buf, dataIn->getData(), uwd.bufsize);
+ uint8_t *pTmp = (uint8_t*) uwd.frame1.buf;
+ // copy remain from delaybuf
+ memcpy(pTmp, &myDelayBuf[0], noByteDelay);
+ // copy new data
+ memcpy(&pTmp[noByteDelay], pInData, uwd.bufsize - noByteDelay);
+ // copy remaining data to delay buf
+ memcpy(&myDelayBuf[0], &pInData[uwd.bufsize - noByteDelay], noByteDelay);
uwd.frame1.ts = ts;
}
@@ -627,6 +652,22 @@ void OutputUHD::set_parameter(const string& parameter, const string& value)
else if (parameter == "muting") {
ss >> myMuting;
}
+ else if (parameter == "staticdelay") {
+ int adjust;
+ ss >> adjust;
+ int newStaticDelay = myStaticDelay + adjust;
+ if (newStaticDelay > 96000)
+ myStaticDelay = newStaticDelay - 96000;
+ else if (newStaticDelay < 0)
+ myStaticDelay = newStaticDelay + 96000;
+ else
+ myStaticDelay = newStaticDelay;
+ }
+ else if (parameter == "iqbalance") {
+ ss >> myConf.frequency;
+ myUsrp->set_tx_freq(myConf.frequency);
+ myConf.frequency = myUsrp->get_tx_freq();
+ }
else {
stringstream ss;
ss << "Parameter '" << parameter
@@ -647,6 +688,9 @@ const string OutputUHD::get_parameter(const string& parameter) const
else if (parameter == "muting") {
ss << myMuting;
}
+ else if (parameter == "staticdelay") {
+ ss << myStaticDelay;
+ }
else {
ss << "Parameter '" << parameter <<
"' is not exported by controllable " << get_rc_name();
diff --git a/src/OutputUHD.h b/src/OutputUHD.h
index a2ffb7d..7eb6733 100644
--- a/src/OutputUHD.h
+++ b/src/OutputUHD.h
@@ -188,6 +188,7 @@ struct OutputUHDConfig {
class OutputUHD: public ModOutput, public RemoteControllable {
public:
+
OutputUHD(
OutputUHDConfig& config,
Logger& logger);
@@ -229,6 +230,10 @@ class OutputUHD: public ModOutput, public RemoteControllable {
// muting can only be changed using the remote control
bool myMuting;
+ private:
+ // data
+ int myStaticDelay;
+ std::vector<complexf> myDelayBuf;
size_t lastLen;
};
diff --git a/src/OutputZeroMQ.cpp b/src/OutputZeroMQ.cpp
index 0e759dd..793e473 100644
--- a/src/OutputZeroMQ.cpp
+++ b/src/OutputZeroMQ.cpp
@@ -30,7 +30,7 @@
#include <string.h>
#include <sstream>
-#if defined(HAVE_OUTPUT_ZEROMQ)
+#if defined(HAVE_ZEROMQ)
OutputZeroMQ::OutputZeroMQ(std::string endpoint, Buffer* dataOut)
: ModOutput(ModFormat(1), ModFormat(0)),
@@ -63,5 +63,5 @@ int OutputZeroMQ::process(Buffer* dataIn, Buffer* dataOut)
return dataIn->getLength();
}
-#endif // HAVE_OUTPUT_ZEROMQ_H
+#endif // HAVE_ZEROMQ
diff --git a/src/OutputZeroMQ.h b/src/OutputZeroMQ.h
index 1c48fe7..a80eab4 100644
--- a/src/OutputZeroMQ.h
+++ b/src/OutputZeroMQ.h
@@ -31,7 +31,7 @@
# include "config.h"
#endif
-#if defined(HAVE_OUTPUT_ZEROMQ)
+#if defined(HAVE_ZEROMQ)
#include "ModOutput.h"
#include "zmq.hpp"
@@ -54,7 +54,7 @@ class OutputZeroMQ : public ModOutput
std::string m_name;
};
-#endif // HAVE_OUTPUT_ZEROMQ_H
+#endif // HAVE_ZEROMQ
#endif // OUTPUT_ZEROMQ_H
diff --git a/src/RemoteControl.cpp b/src/RemoteControl.cpp
index 5bbd2f8..65da3b7 100644
--- a/src/RemoteControl.cpp
+++ b/src/RemoteControl.cpp
@@ -66,7 +66,8 @@ void RemoteControllerTelnet::process(long)
try {
boost::asio::io_service io_service;
- tcp::acceptor acceptor(io_service, tcp::endpoint(tcp::v4(), m_port));
+ tcp::acceptor acceptor(io_service, tcp::endpoint(
+ boost::asio::ip::address::from_string("127.0.0.1"), m_port) );
while (m_running) {
in_message = "";
@@ -246,3 +247,146 @@ void RemoteControllerTelnet::reply(tcp::socket& socket, string message)
ignored_error);
}
+
+#if defined(HAVE_ZEROMQ)
+
+void RemoteControllerZmq::restart()
+{
+ m_restarter_thread = boost::thread(&RemoteControllerZmq::restart_thread, this);
+}
+
+// This runs in a separate thread, because
+// it would take too long to be done in the main loop
+// thread.
+void RemoteControllerZmq::restart_thread()
+{
+ m_running = false;
+
+ if (!m_endpoint.empty()) {
+ m_child_thread.interrupt();
+ m_child_thread.join();
+ }
+
+ m_child_thread = boost::thread(&RemoteControllerZmq::process, this);
+}
+
+void RemoteControllerZmq::recv_all(zmq::socket_t* pSocket, std::vector<std::string> &message)
+{
+ int more = -1;
+ size_t more_size = sizeof(more);
+
+ while (more != 0)
+ {
+ zmq::message_t msg;
+ pSocket->recv(&msg);
+ message.push_back(std::string((char*)msg.data(), msg.size()));
+ pSocket->getsockopt(ZMQ_RCVMORE, &more, &more_size);
+ }
+}
+
+void RemoteControllerZmq::send_ok_reply(zmq::socket_t *pSocket)
+{
+ zmq::message_t msg(2);
+ char repCode[2] = {'o', 'k'};
+ memcpy ((void*) msg.data(), repCode, 2);
+ pSocket->send(msg, 0);
+}
+
+void RemoteControllerZmq::send_fail_reply(zmq::socket_t *pSocket, const std::string &error)
+{
+ zmq::message_t msg1(4);
+ char repCode[4] = {'f', 'a', 'i', 'l'};
+ memcpy ((void*) msg1.data(), repCode, 4);
+ pSocket->send(msg1, ZMQ_SNDMORE);
+
+ zmq::message_t msg2(error.length());
+ memcpy ((void*) msg2.data(), error.c_str(), error.length());
+ pSocket->send(msg2, 0);
+}
+
+void RemoteControllerZmq::process()
+{
+ // create zmq reply socket for receiving ctrl parameters
+ zmq::socket_t repSocket(m_zmqContext, ZMQ_REP);
+ std::cout << "Starting zmq remote control thread" << std::endl;
+ try
+ {
+ // connect the socket
+ int hwm = 100;
+ int linger = 0;
+ repSocket.setsockopt(ZMQ_RCVHWM, &hwm, sizeof(hwm));
+ repSocket.setsockopt(ZMQ_SNDHWM, &hwm, sizeof(hwm));
+ repSocket.setsockopt(ZMQ_LINGER, &linger, sizeof(linger));
+ repSocket.bind(m_endpoint.c_str());
+
+ // create pollitem that polls the ZMQ sockets
+ zmq::pollitem_t pollItems[] = { {repSocket, 0, ZMQ_POLLIN, 0} };
+ for(;;)
+ {
+ zmq::poll(pollItems, 1, 100);
+ std::vector<std::string> msg;
+ if (pollItems[0].revents & ZMQ_POLLIN)
+ {
+ recv_all(&repSocket, msg);
+ std::string command((char*)msg[0].data(), msg[0].size());
+
+ if (msg.size() == 1 && command == "ping")
+ {
+ send_ok_reply(&repSocket);
+ }
+ else if (msg.size() == 3 && command == "get")
+ {
+ std::string module((char*) msg[1].data(), msg[1].size());
+ std::string parameter((char*) msg[2].data(), msg[2].size());
+
+ try
+ {
+ std::string value = get_param_(module, parameter);
+ zmq::message_t *pMsg = new zmq::message_t(value.size());
+ memcpy ((void*) pMsg->data(), value.data(), value.size());
+ repSocket.send(*pMsg, 0);
+ delete pMsg;
+ }
+ catch (ParameterError &err)
+ {
+ send_fail_reply(&repSocket, err.what());
+ }
+ }
+ else if (msg.size() == 4 && command == "set")
+ {
+ std::string module((char*) msg[1].data(), msg[1].size());
+ std::string parameter((char*) msg[2].data(), msg[2].size());
+ std::string value((char*) msg[3].data(), msg[3].size());
+
+ try
+ {
+ set_param_(module, parameter, value);
+ send_ok_reply(&repSocket);
+ }
+ catch (ParameterError &err)
+ {
+ send_fail_reply(&repSocket, err.what());
+ }
+ }
+ else
+ send_fail_reply(&repSocket, "Unsupported command");
+ }
+
+ // check if thread is interrupted
+ boost::this_thread::interruption_point();
+ }
+ }
+ catch (boost::thread_interrupted&) {}
+ catch (zmq::error_t &e)
+ {
+ std::cerr << "ZMQ error: " << std::string(e.what()) << std::endl;
+ }
+ catch (std::exception& e)
+ {
+ std::cerr << "Remote control caught exception: " << e.what() << std::endl;
+ m_fault = true;
+ }
+ repSocket.close();
+}
+#endif
+
diff --git a/src/RemoteControl.h b/src/RemoteControl.h
index 09e7492..89a1583 100644
--- a/src/RemoteControl.h
+++ b/src/RemoteControl.h
@@ -29,6 +29,14 @@
#ifndef _REMOTECONTROL_H
#define _REMOTECONTROL_H
+#ifdef HAVE_CONFIG_H
+# include "config.h"
+#endif
+
+#if defined(HAVE_ZEROMQ)
+#include <zmq.hpp>
+#endif
+
#include <list>
#include <map>
#include <string>
@@ -85,6 +93,39 @@ class BaseRemoteController {
virtual ~BaseRemoteController() {}
};
+/* Holds all our remote controllers, i.e. we may have more than
+ * one type of controller running.
+ */
+class RemoteControllers {
+ public:
+ void add_controller(BaseRemoteController *rc) {
+ m_controllers.push_back(rc);
+ }
+
+ void add_controllable(RemoteControllable *rc) {
+ for (std::list<BaseRemoteController*>::iterator it = m_controllers.begin();
+ it != m_controllers.end(); ++it) {
+ (*it)->enrol(rc);
+ }
+ }
+
+ void check_faults() {
+ for (std::list<BaseRemoteController*>::iterator it = m_controllers.begin();
+ it != m_controllers.end(); ++it) {
+ if ((*it)->fault_detected())
+ {
+ fprintf(stderr,
+ "Detected Remote Control fault, restarting it\n");
+ (*it)->restart();
+ }
+ }
+ }
+ size_t get_no_controllers() { return m_controllers.size(); }
+
+ private:
+ std::list<BaseRemoteController*> m_controllers;
+};
+
/* Objects that support remote control must implement the following class */
class RemoteControllable {
public:
@@ -100,8 +141,8 @@ class RemoteControllable {
virtual std::string get_rc_name() const { return m_name; }
/* Tell the controllable to enrol at the given controller */
- virtual void enrol_at(BaseRemoteController& controller) {
- controller.enrol(this);
+ virtual void enrol_at(RemoteControllers& controllers) {
+ controllers.add_controllable(this);
}
/* Return a list of possible parameters that can be set */
@@ -254,6 +295,88 @@ class RemoteControllerTelnet : public BaseRemoteController {
int m_port;
};
+#if defined(HAVE_ZEROMQ)
+/* Implements a Remote controller using zmq transportlayer
+ * that listens on localhost
+ */
+class RemoteControllerZmq : public BaseRemoteController {
+ public:
+ RemoteControllerZmq()
+ : m_running(false), m_fault(false),
+ m_zmqContext(1),
+ m_endpoint("") { }
+
+ RemoteControllerZmq(std::string endpoint)
+ : m_running(true), m_fault(false),
+ m_child_thread(&RemoteControllerZmq::process, this),
+ m_zmqContext(1),
+ m_endpoint(endpoint) { }
+
+ ~RemoteControllerZmq() {
+ m_running = false;
+ m_fault = false;
+ if (!m_endpoint.empty()) {
+ m_child_thread.interrupt();
+ m_child_thread.join();
+ }
+ }
+
+ void enrol(RemoteControllable* controllable) {
+ m_cohort.push_back(controllable);
+ }
+
+ virtual bool fault_detected() { return m_fault; }
+
+ virtual void restart();
+
+ private:
+ void restart_thread();
+
+ void recv_all(zmq::socket_t* pSocket, std::vector<std::string> &message);
+ void send_ok_reply(zmq::socket_t *pSocket);
+ void send_fail_reply(zmq::socket_t *pSocket, const std::string &error);
+ void process();
+
+
+ RemoteControllerZmq& operator=(const RemoteControllerZmq& other);
+ RemoteControllerZmq(const RemoteControllerZmq& other);
+
+ RemoteControllable* get_controllable_(std::string name) {
+ for (std::list<RemoteControllable*>::iterator it = m_cohort.begin();
+ it != m_cohort.end(); ++it) {
+ if ((*it)->get_rc_name() == name)
+ {
+ return *it;
+ }
+ }
+ throw ParameterError("Module name unknown");
+ }
+
+ std::string get_param_(std::string name, std::string param) {
+ RemoteControllable* controllable = get_controllable_(name);
+ return controllable->get_parameter(param);
+ }
+
+ void set_param_(std::string name, std::string param, std::string value) {
+ RemoteControllable* controllable = get_controllable_(name);
+ return controllable->set_parameter(param, value);
+ }
+
+ bool m_running;
+
+ /* This is set to true if a fault occurred */
+ bool m_fault;
+ boost::thread m_restarter_thread;
+
+ boost::thread m_child_thread;
+
+ /* This controller commands the controllables in the cohort */
+ std::list<RemoteControllable*> m_cohort;
+
+ zmq::context_t m_zmqContext;
+ std::string m_endpoint;
+};
+#endif
/* The Dummy remote controller does nothing, and never fails
*/
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-08 01:04:10 +0000