This commit is contained in:
Ren RenJuan
parent
8ae588ed40
commit
d82e77382d
|
|
@ -37,7 +37,7 @@ namespace fsm = boost::statechart;
|
|||
|
||||
#ifdef MD_MAIN
|
||||
|
||||
mdGTEEngine *engineFactory;
|
||||
mdDeviceFabrik *engineFactory;
|
||||
mdLogger *theseLogs;
|
||||
masterDaemonConfig *thisConfig;
|
||||
|
||||
|
|
@ -46,7 +46,7 @@ namespace fsm = boost::statechart;
|
|||
|
||||
#else
|
||||
|
||||
extern mdGTEEngine *engineFactory;
|
||||
extern mdDeviceFabrik *engineFactory;
|
||||
extern mdLogger *theseLogs;
|
||||
extern masterDaemonConfig *thisConfig;
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,86 @@
|
|||
#ifndef MASTER_DAEMON
|
||||
#define MASTER_DAEMON
|
||||
|
||||
#include <msgpack.hpp>
|
||||
|
||||
/*! \brief masterDaemon
|
||||
* server core.
|
||||
*
|
||||
* Contains the data layer and the XMLRPC API.
|
||||
*/
|
||||
|
||||
#ifdef MD_CORE
|
||||
|
||||
class masterDaemon *thisService;
|
||||
|
||||
#else
|
||||
|
||||
extern class masterDaemon *thisService;
|
||||
|
||||
#endif
|
||||
|
||||
using boost::asio::ip::udp;
|
||||
|
||||
|
||||
class masterDaemon : public mdProcess,
|
||||
public Listener<mdAttention>,
|
||||
public Listener<mdCDPulse>,
|
||||
public Listener<mdClientBirth>,
|
||||
public Listener<mdClientDeath>,
|
||||
public Listener<mdIncoming>,
|
||||
public Listener<mdResponse>,
|
||||
public Listener<mdTelemetryFrame>,
|
||||
public Listener<mdDeviceCommand> {
|
||||
public:
|
||||
|
||||
bool shuttingDown;
|
||||
|
||||
boost::asio::io_service io_;
|
||||
|
||||
int arCycles,
|
||||
dataClients[MAX_DATACLIENTS],
|
||||
instruments[MAX_INSTRUMENTS],
|
||||
nClievers,
|
||||
sentCommands;
|
||||
|
||||
masterDaemonConfig *cfg;
|
||||
mdDGChannel *bg,*fg;
|
||||
|
||||
std::string clievers[MAX_CLIEVER];
|
||||
|
||||
masterDaemon();
|
||||
~masterDaemon() {}
|
||||
|
||||
masterDaemon(masterDaemonConfig *cmdCfg) { int i;
|
||||
thisService = this;
|
||||
cfg = cmdCfg;
|
||||
nClievers = 0;
|
||||
shuttingDown = false;
|
||||
memset(dataClients,0,sizeof(dataClients));
|
||||
memset(instruments,0,sizeof(instruments));
|
||||
}
|
||||
|
||||
int getDeviceHandle(int deviceMajor,std::string &deviceMinor) {};
|
||||
int initBaseAPI(void);
|
||||
int initDataLayer(void);
|
||||
int releaseDevice(int handle) {return( -1);}
|
||||
int validateHandleForCmds(int handle) {return(-1);}
|
||||
void dispatch(mdWQitem*);
|
||||
void dispatch(const mdIncoming&);
|
||||
void listen();
|
||||
xmlrpc_c::value* fetchCommands(std::string subSystem) {};
|
||||
|
||||
virtual void processEvent(const mdAttention &ev);
|
||||
virtual void processEvent(const mdCDPulse &ev);
|
||||
virtual void processEvent(const mdClientBirth &ev);
|
||||
virtual void processEvent(const mdClientDeath &ev);
|
||||
virtual void processEvent(const mdIncoming &ev);
|
||||
virtual void processEvent(const mdResponse &ev);
|
||||
virtual void processEvent(const mdTelemetryFrame &ev);
|
||||
virtual void processEvent(const mdDeviceCommand &ev);
|
||||
|
||||
void run();
|
||||
|
||||
};
|
||||
|
||||
#endif
|
||||
|
|
@ -0,0 +1,90 @@
|
|||
|
||||
#include <boost/asio.hpp>
|
||||
|
||||
/*! \brief mdDevice
|
||||
* General abstraction of all MD clients
|
||||
*
|
||||
* For historical reasons all clients are considered to be
|
||||
* devices, though mdclient and US clients have a non-device
|
||||
* client type. The central device of the system has zero
|
||||
* device type.
|
||||
*
|
||||
*/
|
||||
|
||||
using namespace std;
|
||||
using boost::asio::ip::udp;
|
||||
|
||||
template<class T>
|
||||
class mdDevice {
|
||||
public:
|
||||
|
||||
bool isSingleton;
|
||||
int clieverGroup, // masterDaemonConfig.thisMachineContext
|
||||
handle,mdStdDevIdx;
|
||||
md_device type;
|
||||
mdState state;
|
||||
|
||||
udp::endpoint ip;
|
||||
|
||||
// Only Machine and Instrument use below
|
||||
InstructionSet cmds;
|
||||
|
||||
// Some parameters initially here are now all uniformly ODEs
|
||||
// defined in the COOL scripts.
|
||||
|
||||
~mdDevice() {}
|
||||
mdDevice(md_device t) : type(t) {clieverGroup = handle = mdStdDevIdx = -1;}
|
||||
|
||||
void lxi_control(T* device,std::string scpiText);
|
||||
T* registeR(md_device t);
|
||||
|
||||
xmlrpc_c::value* fetchCommands(std::string subSystem);
|
||||
void registerCmd(const char *cmdName,const mdIncoming &mdI);
|
||||
|
||||
};
|
||||
|
||||
class mdClientServer;
|
||||
class mdClientServer : public mdDevice<mdClientServer> {
|
||||
public:
|
||||
|
||||
mdClientServer() : mdDevice<mdClientServer>( MDDEV_CD ) {};
|
||||
mdClientServer *validateClient(int handle, mdResponse &r);
|
||||
};
|
||||
|
||||
class mdMachine;
|
||||
class mdMachine : public mdDevice<mdMachine> {
|
||||
public:
|
||||
|
||||
mdMachine() : mdDevice<mdMachine>( MACHINE )
|
||||
{ cmds["RST"] = new mdCommand(MD_SCPI,std::string("RST")); }
|
||||
mdMachine *validateClient(int handle, const mdClientBirth &c, mdResponse &r);
|
||||
void registerCmd(const char *cmdName,const mdIncoming &mdI);
|
||||
};
|
||||
|
||||
class mdInstrument;
|
||||
class mdInstrument : public mdDevice<mdInstrument> {
|
||||
public:
|
||||
|
||||
mdInstrument() : mdDevice<mdInstrument>( MDDEV_INSTRUMENT )
|
||||
{ cmds["RST"] = new mdCommand(MD_SCPI,std::string("RST")); }
|
||||
mdInstrument *validateClient(int handle, const mdClientBirth &c, mdResponse &r);
|
||||
void registerCmd(const char *cmdName,const mdIncoming &mdI);
|
||||
};
|
||||
|
||||
class mdDataClient;
|
||||
class mdDataClient : public mdDevice<mdDataClient> {
|
||||
public:
|
||||
|
||||
mdDataClient() : mdDevice<mdDataClient>( MDDEV_DATACLIENT ) {};
|
||||
mdDataClient *validateClient(int handle);
|
||||
};
|
||||
|
||||
class masterDaemon;
|
||||
class mdDeviceFabrik : public mdDevice<masterDaemon>
|
||||
{public:
|
||||
|
||||
mdDeviceFabrik() : mdDevice<masterDaemon>( MDDEV_MD ) {}
|
||||
void newFromHeartbeat(const mdClientBirth &itsAWhat);
|
||||
std::string newFromAPI(md_device type,std::string signature);
|
||||
|
||||
};
|
||||
|
|
@ -70,9 +70,9 @@ using boost::asio::ip::udp;
|
|||
#define MD_EPOCH date()
|
||||
#define MD_HEARTBEAT 1 // Network peer heartbeat in seconds.
|
||||
#define MD_MAX_DATAGRAM (32*1024) // half of the IPV4 max
|
||||
#define AUSREGCLIEVER_APP "Generic"
|
||||
#define CLIEVER_APP "AusReg Cliever"
|
||||
#define MD_COMPONENT "Master Daemon" // Cliever Component
|
||||
#define MD_NAME DACLIPS_APP " " MD_COMPONENT
|
||||
#define MD_NAME CLIEVER_APP " " MD_COMPONENT
|
||||
#define MD_VERSION " 1.0 " // Version
|
||||
#define MD_REFRESH 10 // default milliseconds between telemetry frame updates
|
||||
#define MD_TYPE "AUSREGCLIEVER" // i.e. what a MACHINE is, Change per your Cliever derivation
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ class mdAttention: public TimeStampedEvent<>, public PolymorphEvent {
|
|||
virtual void send() const { sendTypedEvent(*this); }
|
||||
|
||||
};
|
||||
|
||||
class mdClientBirth: public TimeStampedEvent<>, public PolymorphEvent {
|
||||
public:
|
||||
bool dgDetermined;
|
||||
|
|
|
|||
|
|
@ -66,7 +66,7 @@ void md() {
|
|||
theseLogs->logN(0,"AusReg Cliever <- MasterDaemon.");
|
||||
}
|
||||
|
||||
boost::thread foreground(runMasterDaemon);
|
||||
boost::thread foreground(runMasterDaemon);`
|
||||
boost::thread background(runClientLayer);
|
||||
|
||||
if (!foreground.joinable()) {
|
||||
|
|
@ -108,8 +108,8 @@ main(int const argc,
|
|||
thisConfig = new masterDaemonConfig();
|
||||
thisConfig->shellProcess = getpid();
|
||||
|
||||
mdOrKb = (strcspn(argv[0],"./") == strlen(argv[0])) ? argv[0] : strrchr(argv[0],'/') + 1;
|
||||
mdHasKb = strstr(argv[0],"clips") ? true : false;
|
||||
mdOrAC = (strcspn(argv[0],"./") == strlen(argv[0])) ? argv[0] : strrchr(argv[0],'/') + 1;
|
||||
mdHasEPPTk = strstr(argv[0],"epp") ? true : false;
|
||||
|
||||
if (argc < 3 || argc > 6) usage();
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,327 @@
|
|||
#define MD_CORE
|
||||
#include "auc-md.h"
|
||||
#include "masterDaemon.h"
|
||||
#include "coreapi.h"
|
||||
#include "../server/Listener.cpp"
|
||||
#include "../server/EventSender2.cpp"
|
||||
|
||||
void attention();
|
||||
void arCallback(const boost::system::error_code& error);
|
||||
void mdWQ();
|
||||
|
||||
void masterDaemon::dispatch(mdWQitem *next) {
|
||||
|
||||
bool success;
|
||||
boost::system::error_code ec;
|
||||
const char *failure;
|
||||
int sentBytes,step=1;
|
||||
mdResponse *ackOrNak = (mdResponse *)next->what;
|
||||
|
||||
switch(next->kind) {
|
||||
case DV_MDQUERY:
|
||||
failure = (const char *)&ackOrNak->reply.dg.payLoad[0];
|
||||
success = ackOrNak->reply.dg.hdr.dgType.value == 1;
|
||||
goto commonReply;
|
||||
case MD_NEWBORN:
|
||||
ackOrNak->reply.dg.hdr.msgType = MDDG_NEWBORN;
|
||||
failure = "Stillbirth";
|
||||
success = ackOrNak->mdStdDevIdx >= 0;
|
||||
commonReply:
|
||||
assert(cb.find(ackOrNak->mdStdDevIdx) != cb.end());
|
||||
ackOrNak->reply.dg.hdr.clientType = MDDEV_MD;
|
||||
if (success) {
|
||||
if (!cb[ackOrNak->mdStdDevIdx]->connection.open) {
|
||||
ackOrNak->bus->connect_to(ackOrNak->ip,ec,step,ackOrNak->mdStdDevIdx);
|
||||
if (cb[ackOrNak->mdStdDevIdx]->connection.open) {
|
||||
if ((sentBytes=ackOrNak->bus->send(ackOrNak->reply,ec)) != sizeof(mdDGReply))
|
||||
theseLogs->logNdebug(1,2,"incomplete blocking send: %d: %s",sentBytes,ec.message().c_str());
|
||||
}
|
||||
else
|
||||
theseLogs->logNdebug(1,2,"Couldn't get back channel to client: %s (in step %d).",ec.message().c_str(),step);
|
||||
}
|
||||
else if ((sentBytes=ackOrNak->bus->send(ackOrNak->reply,ec)) != sizeof(mdDGReply))
|
||||
theseLogs->logNdebug(1,2,"incomplete blocking send: %d: %s",sentBytes,ec.message().c_str());
|
||||
|
||||
} else {
|
||||
theseLogs->logNdebug(1,0,failure);
|
||||
}
|
||||
delete ackOrNak;
|
||||
break;
|
||||
}
|
||||
delete next;
|
||||
|
||||
}
|
||||
/*
|
||||
* Prenatal heartbeats and commands are not queued.
|
||||
* Everything else is.
|
||||
*
|
||||
*/
|
||||
void masterDaemon::dispatch(const mdIncoming &what) {
|
||||
|
||||
bool isObservation;
|
||||
const char *name,*xStr;
|
||||
int about=what.dg.hdr.handle;
|
||||
md_device thisKind;
|
||||
mdInstrument *d1;
|
||||
mdMachine *d2;
|
||||
|
||||
map<int,mdLiveClient*>::iterator iter = thisConfig->allClients.find(about);
|
||||
|
||||
switch(what.dg.hdr.msgType) {
|
||||
case MDDG_CDRESET:
|
||||
theseLogs->logN(0,"Shutdown request received from a Cliever");
|
||||
thisConfig->halt = true;
|
||||
break;
|
||||
case MDDG_MDQUERY:
|
||||
if( iter == thisConfig->allClients.end() )
|
||||
theseLogs->logN(1,"Query for device whose handle (%d) has disappeared, ignored.", about );
|
||||
else {
|
||||
thisKind = thisConfig->allClients[about]->devType;
|
||||
name = &what.dg.payLoad[0];
|
||||
xStr = &what.dg.payLoad[what.dg.hdr.primeOffset];
|
||||
switch(what.dg.hdr.dgSubType) {
|
||||
case MDDG_REGSCPI:
|
||||
theseLogs->logNdebug(NORMAL_DEBUG,4,"Src SCPI: '%s' from type: %d ('%s'), handle %d.", name, what.dg.hdr.clientType,xStr,about);
|
||||
if (thisKind == MACHINE) theMachine->registerCmd(name,what);
|
||||
else thisConfig->instruments[about]->registerCmd(name,what);
|
||||
break;
|
||||
case MDDG_REGODE:
|
||||
theseLogs->logNdebug(NORMAL_DEBUG,4,"Src ODE: '%s' from type: %d ('%s'), handle %d.", name, what.dg.hdr.clientType,xStr,about);
|
||||
goto regName;
|
||||
case MDDG_REGOBS:
|
||||
theseLogs->logNdebug(NORMAL_DEBUG,4,"Src Obs: '%s' from type: %d ('%s'), handle %d.", name, what.dg.hdr.clientType,xStr,about);
|
||||
regName:
|
||||
if (thisKind == MACHINE) theMachine->state.registerData(name,what);
|
||||
else thisConfig->instruments[about]->state.registerData(name,what);
|
||||
break;
|
||||
}}
|
||||
break;
|
||||
case MDDG_HEARTBEAT:
|
||||
if (!what.dg.hdr.sourceHandle) {
|
||||
if (what.dg.hdr.clientType < MDDEV_CD || what.dg.hdr.clientType > MDDEV_DATACLIENT) {
|
||||
theseLogs->logN(1,"Heartbeat from unknown client type: %d, ignored.", what.dg.hdr.clientType);
|
||||
break;
|
||||
}
|
||||
theseLogs->logNdebug(NORMAL_DEBUG*4,1,"Heartbeat from new %s ...",clientTypes[what.dg.hdr.clientType]);
|
||||
if (what.dg.hdr.primeOffset >= 5) {
|
||||
mdClientBirth itsAWhat;
|
||||
name = (char *)(&what.dg.payLoad[what.dg.hdr.primeOffset]);
|
||||
theseLogs->logNdebug(NORMAL_DEBUG*4,1," ... its telemetry port is %s.",what.dg.payLoad);
|
||||
theseLogs->logNdebug(NORMAL_DEBUG*4,1," ... '%s' will be its _deviceName.",name);
|
||||
itsAWhat.dg = what.dg;
|
||||
itsAWhat.ip = what.ip;
|
||||
itsAWhat.ip.port((unsigned short)atoi(what.dg.payLoad));
|
||||
itsAWhat.dgDetermined = true;
|
||||
itsAWhat.send();
|
||||
}
|
||||
else
|
||||
theseLogs->logN(1,"Heartbeat didn't appear to say what port to use, ignored.");
|
||||
}
|
||||
else {
|
||||
theseLogs->logNdebug(MAX_DEBUG,2,"Heartbeat from client with handle: %d.",what.dg.hdr.sourceHandle);
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
}
|
||||
int masterDaemon::initBaseAPI(void) {
|
||||
|
||||
int rc=OK;
|
||||
|
||||
try {
|
||||
|
||||
theseLogs->logN(0,"Create Generic Core API");
|
||||
|
||||
xmlrpc_c::methodPtr const registerDeviceP(new registerDevice(thisService->cfg));
|
||||
xmlrpc_c::methodPtr const getMDversionP(new getMDversion);
|
||||
xmlrpc_c::methodPtr const getP(new getter);
|
||||
xmlrpc_c::methodPtr const setP(new setter);
|
||||
xmlrpc_c::methodPtr const getCmdListP(new cmdListFetch);
|
||||
xmlrpc_c::methodPtr const getCmdP(new cmd);
|
||||
xmlrpc_c::methodPtr const createP(new create);
|
||||
|
||||
thisConfig->api_registry.addMethod("device.registeR", registerDeviceP );
|
||||
thisConfig->api_registry.addMethod("state.getMDversion", getMDversionP );
|
||||
thisConfig->api_registry.addMethod("state.create", createP);
|
||||
thisConfig->api_registry.addMethod("state.get", getP );
|
||||
thisConfig->api_registry.addMethod("state.set", setP );
|
||||
thisConfig->api_registry.addMethod("behavior.getCommandList", getCmdListP );
|
||||
thisConfig->api_registry.addMethod("behavior.command", getCmdP );
|
||||
|
||||
}
|
||||
catch(...)
|
||||
{ rc = NOT_OK; }
|
||||
|
||||
return rc;
|
||||
|
||||
}
|
||||
void masterDaemon::listen() {
|
||||
|
||||
EventSender<mdAttention>::add(*this);
|
||||
assert(EventSender<mdAttention>::getNumListeners() == 1);
|
||||
EventSender<mdCDPulse>::add(*this);
|
||||
assert(EventSender<mdCDPulse>::getNumListeners() == 1);
|
||||
EventSender<mdClientBirth>::add(*this);
|
||||
assert(EventSender<mdClientBirth>::getNumListeners() == 1);
|
||||
EventSender<mdClientDeath>::add(*this);
|
||||
assert(EventSender<mdClientDeath>::getNumListeners() == 1);
|
||||
EventSender<mdDeviceCommand>::add(*this);
|
||||
assert(EventSender<mdDeviceCommand>::getNumListeners() == 1);
|
||||
EventSender<mdIncoming>::add(*this);
|
||||
assert(EventSender<mdIncoming>::getNumListeners() == 1);
|
||||
EventSender<mdResponse>::add(*this);
|
||||
assert(EventSender<mdResponse>::getNumListeners() == 1);
|
||||
EventSender<mdTelemetryFrame>::add(*this);
|
||||
assert(EventSender<mdTelemetryFrame>::getNumListeners() == 1);
|
||||
|
||||
boost::thread mdAr(attention);
|
||||
|
||||
}
|
||||
void masterDaemon::processEvent( const mdAttention &thisAR )
|
||||
{
|
||||
assert(EventSender<mdAttention>::isSending());
|
||||
}
|
||||
void masterDaemon::processEvent( const mdCDPulse &thisPulse )
|
||||
{
|
||||
assert(EventSender<mdCDPulse>::isSending());
|
||||
}
|
||||
void masterDaemon::processEvent( const mdClientBirth &thisWhat )
|
||||
{
|
||||
assert(EventSender<mdClientBirth>::isSending());
|
||||
if (thisWhat.dgDetermined) {
|
||||
deviceFactory->newFromHeartbeat(thisWhat);
|
||||
}
|
||||
else {
|
||||
deviceFactory->newFromAPI(
|
||||
thisWhat.clientType,thisWhat.signature);
|
||||
}
|
||||
}
|
||||
void masterDaemon::processEvent( const mdClientDeath &thisWas )
|
||||
{
|
||||
assert(EventSender<mdClientDeath>::isSending());
|
||||
}
|
||||
void masterDaemon::processEvent( const mdDeviceCommand &thisCmd )
|
||||
{
|
||||
assert(EventSender<mdDeviceCommand>::isSending());
|
||||
}
|
||||
void masterDaemon::processEvent( const mdIncoming &thisDatagram )
|
||||
{
|
||||
assert(EventSender<mdIncoming>::isSending());
|
||||
thisService->dispatch(thisDatagram);
|
||||
}
|
||||
void masterDaemon::processEvent( const mdResponse &thisReply )
|
||||
{
|
||||
const void *queued = &thisReply;
|
||||
|
||||
assert(EventSender<mdResponse>::isSending());
|
||||
queue(new mdWQitem( queued, thisReply.dCat, 0 ));
|
||||
|
||||
}
|
||||
void masterDaemon::processEvent( const mdTelemetryFrame &thisFrame )
|
||||
{
|
||||
assert(EventSender<mdTelemetryFrame>::isSending());
|
||||
}
|
||||
void masterDaemon::run() {
|
||||
|
||||
deviceFactory = new mdDeviceFabrik();
|
||||
fg = new mdDGChannel( thisService->io_, 0 );
|
||||
|
||||
if (initBaseAPI()) return;
|
||||
listen();
|
||||
boost::thread work(mdWQ);
|
||||
assert(work.joinable());
|
||||
theseLogs->logNdebug(MAX_DEBUG,0,"Master Daemon worker started, foreground async i/o service joins MD worker.");
|
||||
io_.run();
|
||||
work.join();
|
||||
|
||||
}
|
||||
|
||||
void mdDGChannel::handle_receive_from(const boost::system::error_code& error,
|
||||
size_t bytes_recvd)
|
||||
{
|
||||
const char *c1;
|
||||
|
||||
if (!error && bytes_recvd > 0)
|
||||
{
|
||||
mdIncoming incoming(thisService->bg);
|
||||
|
||||
incoming.ip = thisService->bg->p_endpoint_;
|
||||
c1 = thisService->bg->p_endpoint_.address().to_string().c_str();
|
||||
|
||||
if (incoming.dg.hdr.clientType > 0 && incoming.dg.hdr.clientType < N_MDDEV_TYPES)
|
||||
{ theseLogs->logNdebug(MAX_DEBUG,3,"msgtype %d received from %s (a '%s').",incoming.dg.hdr.msgType,c1,clientTypes[incoming.dg.hdr.clientType]);
|
||||
incoming.send();
|
||||
} else
|
||||
theseLogs->logNdebug(1,2,"msgtype %d received from unknown MD client type at %s, ignored.",incoming.dg.hdr.msgType,c1);
|
||||
}
|
||||
passive_.async_receive_from(
|
||||
boost::asio::buffer(data_, MD_MAX_DATAGRAM), p_endpoint_,
|
||||
boost::bind(&mdDGChannel::handle_receive_from, this,
|
||||
boost::asio::placeholders::error,
|
||||
boost::asio::placeholders::bytes_transferred));
|
||||
}
|
||||
|
||||
|
||||
// Callbacks, thread runners, and other ordinary functions.
|
||||
|
||||
void attention() {
|
||||
|
||||
bool announced = false;
|
||||
|
||||
// The housekeeping routine is the only cyclic event in MD
|
||||
boost::asio::deadline_timer t0(thisService->io_, boost::posix_time::seconds(MD_HAUSHALT));
|
||||
while (!thisService->shuttingDown)
|
||||
{t0.async_wait(arCallback);
|
||||
sleep(2*MD_HAUSHALT);
|
||||
if (!announced) {announced = true;
|
||||
theseLogs->logNdebug(MAX_DEBUG,0,"First invocation of housekeeping.");}
|
||||
}
|
||||
|
||||
}
|
||||
void arCallback(const boost::system::error_code& error) {
|
||||
|
||||
mdAttention housekeep( thisService->arCycles++ );
|
||||
housekeep.send();
|
||||
|
||||
}
|
||||
void runMasterDaemon() {
|
||||
|
||||
cb[0] = new mdCB();
|
||||
thisService = new masterDaemon( thisConfig );
|
||||
thisService->run();
|
||||
|
||||
}
|
||||
void mdWQ() {
|
||||
|
||||
while(!thisConfig->shutdown) {
|
||||
if (!thisConfig->halt && thisService->q.size())
|
||||
{ thisService->dispatch(thisService->q.top()); thisService->q.pop(); }
|
||||
else
|
||||
boost::this_thread::yield();
|
||||
}
|
||||
|
||||
}
|
||||
void runDataLayer() {
|
||||
|
||||
boost::asio::io_service io_;
|
||||
|
||||
theseLogs->logN(1,"Background dgram service thread starting on port %d.",thisConfig->telemetryPort);
|
||||
|
||||
try {
|
||||
|
||||
thisService->bg = new mdDGChannel(io_, thisConfig->telemetryPort);
|
||||
io_.run();
|
||||
|
||||
}
|
||||
catch (std::exception& e)
|
||||
{
|
||||
theseLogs->logN(1,"Fatal error on main bus background: %s .",e.what());
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
theseLogs->logN(0,"Unknown failure in background datalayer.");
|
||||
}
|
||||
|
||||
theseLogs->logNdebug(1,0,"mainbus background thread exited.");
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,346 @@
|
|||
#include "cliever-md.h"
|
||||
#include "masterDaemon.h"
|
||||
#include "../server/Listener.cpp"
|
||||
#include "../server/EventSender2.cpp"
|
||||
using namespace std;
|
||||
|
||||
/*! \brief Client object implementatios
|
||||
*
|
||||
*/
|
||||
|
||||
int getHandle() {
|
||||
|
||||
map<int,mdLiveClient*>::iterator it;
|
||||
|
||||
bool collision = thisConfig->allClients.size() > 0 ? false: true, found;
|
||||
int value; srand ( time(NULL) );
|
||||
|
||||
value = (rand() % (MAX_DEVICE * 10)) + 1;
|
||||
|
||||
while(collision) {
|
||||
for(found = false, it = thisConfig->allClients.begin();
|
||||
it != thisConfig->allClients.end() && !found;
|
||||
found = (it->first == value ? true : false), it++);
|
||||
if (!found) collision = false;
|
||||
else value = (rand() % (MAX_DEVICE * 10)) + 1;
|
||||
|
||||
}
|
||||
return value;
|
||||
|
||||
}
|
||||
template<class T> T* mdDevice<T>::registeR(md_device t) {
|
||||
|
||||
T *value=NULL;
|
||||
int h = getHandle();
|
||||
|
||||
if (value=mdDevice<T>::validateClient(h)) {
|
||||
theseLogs->logN(2,"Handle %d assigned to new client of type: %s",clientTypes[t]);
|
||||
} else {
|
||||
theseLogs->logN(2,"Validation failed for client type: %s",clientTypes[t]);
|
||||
return value;
|
||||
}
|
||||
|
||||
value->handle = h;
|
||||
value->create();
|
||||
thisConfig->allClients[h] = value; // validateClient has already added to group
|
||||
return value ;
|
||||
|
||||
}
|
||||
template<class T>
|
||||
void mdDevice<T>::lxi_control(T *device, std::string fullText) {
|
||||
|
||||
T *target = device;
|
||||
char *ip,*port,*command,*timeout,*argv[5];
|
||||
|
||||
argv[1] = ip = (char *)malloc(32);
|
||||
argv[2] = port = (char *)malloc(16);
|
||||
argv[3] = timeout = (char *)malloc(32);
|
||||
argv[4] = command = (char *)malloc(1024);
|
||||
|
||||
sprintf(ip,"--ip %s",target->ip.c_str());
|
||||
sprintf(port,"--port %s",target->port.c_str());
|
||||
sprintf(timeout,"--timeout %s",target->timeout.c_str());
|
||||
sprintf(command,"--scpi %s",fullText.c_str());
|
||||
|
||||
lxi_control(5,argv);
|
||||
|
||||
free(ip);
|
||||
free(port);
|
||||
free(timeout);
|
||||
free(command);
|
||||
|
||||
}
|
||||
mdClientServer* mdClientServer::validateClient(int handle, mdResponse &r) {
|
||||
|
||||
bool isNew=true;
|
||||
int i,m=-1,n=-1;
|
||||
mdClientServer *value=NULL;
|
||||
|
||||
if (!thisConfig->nClievers) {m =0; isNew = true;}
|
||||
else for(i=0;i<MAX_CLIEVER && isNew;i++)
|
||||
{if (m < 0 && thisConfig->clievers[i].empty()) m = i;
|
||||
if (ip.address().to_string() == thisConfig->clievers[i]) isNew = false;
|
||||
}
|
||||
|
||||
if (isNew && thisConfig->nClievers < MAX_CLIEVER)
|
||||
{ for (n=i=0;i<thisConfig->nClievers && n < 0;i++)
|
||||
if (!thisConfig->cliever[i]) n = i;
|
||||
thisConfig->nClievers++;
|
||||
thisConfig->cliever[n] = value = this;
|
||||
mdStdDevIdx = n + 1;
|
||||
ip = r.ip;
|
||||
thisConfig->clievers[n] = ip.address().to_string();
|
||||
}
|
||||
else {
|
||||
theseLogs->logN(1,"Either a Cliever already active at %s or limit number (%d) reached.",ip.address().to_string().c_str(),MAX_CLIEVER);
|
||||
}
|
||||
return value;
|
||||
|
||||
}
|
||||
mdMachine* mdMachine::validateClient(int handle, const mdClientBirth &c, mdResponse &r) {
|
||||
|
||||
char *cp;
|
||||
mdMachine *value=NULL;
|
||||
|
||||
if (!theMachine) {
|
||||
if (c.dg.hdr.dgType.clieverGroup) {
|
||||
theseLogs->logN(1,"Machine specified non-zero cliever group(%d) in GDOLMS 1.x, rejected.",c.dg.hdr.dgType.clieverGroup);
|
||||
goto done;
|
||||
}
|
||||
if (!thisConfig->cliever[c.dg.hdr.dgType.clieverGroup]) {
|
||||
theseLogs->logN(1,"The cliever for this device group (%d) is not online, machine birth rejected.",c.dg.hdr.dgType.clieverGroup);
|
||||
goto done;
|
||||
}
|
||||
strcpy(r.reply.dg.payLoad,thisConfig->clievers[c.dg.hdr.dgType.clieverGroup].c_str());
|
||||
cp = &r.reply.dg.payLoad[0] + strlen(r.reply.dg.payLoad) + 1;
|
||||
*((unsigned short *)cp) = thisConfig->cliever[c.dg.hdr.dgType.clieverGroup]->ip.port();
|
||||
theMachine = value = this;
|
||||
mdStdDevIdx = MAX_CLIEVER + 1;
|
||||
}
|
||||
|
||||
done:
|
||||
return value;
|
||||
|
||||
}
|
||||
mdInstrument* mdInstrument::validateClient(int handle, const mdClientBirth &c, mdResponse &r) {
|
||||
|
||||
mdInstrument *value=NULL;
|
||||
|
||||
if (thisConfig->instruments.size() < MAX_INSTRUMENTS) {
|
||||
thisConfig->instruments[handle] = value = this;
|
||||
// mdStdDevIdx =
|
||||
}
|
||||
else theseLogs->logN(1,"Too many instruments, configured limit is: %d.",MAX_INSTRUMENTS);
|
||||
|
||||
return value;
|
||||
|
||||
}
|
||||
mdDataClient* mdDataClient::validateClient(int handle) {
|
||||
|
||||
mdDataClient *value=NULL;
|
||||
|
||||
if (thisConfig->clients.size() < MAX_DATACLIENTS) {
|
||||
thisConfig->clients[handle] = value = this;
|
||||
}
|
||||
else theseLogs->logN(1,"Too many non-device clients, configured limit is: %d.",MAX_DATACLIENTS);
|
||||
|
||||
return value;
|
||||
|
||||
}
|
||||
std::string mdDeviceFabrik::newFromAPI(md_device type,std::string thisSpecialOne) {
|
||||
|
||||
}
|
||||
void mdDeviceFabrik::newFromHeartbeat(const mdClientBirth &thisOne) {
|
||||
|
||||
const char *kind,*outcome;
|
||||
void *resultat;
|
||||
|
||||
int i,mdStdDevIdx;
|
||||
md_device thisKind;
|
||||
mdCB *newControlBlock;
|
||||
mdLiveClient *newAllMap;
|
||||
mdClientServer *newCliever;
|
||||
mdMachine *newMachine;
|
||||
mdInstrument *newInstrument;
|
||||
mdResponse *result = new mdResponse(thisService->bg,thisOne.ip);
|
||||
|
||||
int maybe=getHandle();
|
||||
|
||||
result->dCat = MD_NEWBORN;
|
||||
result->reply.dg.hdr = thisOne.dg.hdr;
|
||||
result->reply.dg.hdr.dgType.isAckNak = true;
|
||||
result->reply.dg.hdr.dgType.value = true;
|
||||
result->ip = thisOne.ip;
|
||||
|
||||
switch(thisOne.dg.hdr.clientType) {
|
||||
case MDDEV_CD:
|
||||
thisKind = MDDEV_CD;
|
||||
kind = "cliever";
|
||||
newCliever = new mdClientServer();
|
||||
if (resultat = newCliever = newCliever->validateClient( maybe, *result )) {
|
||||
newCliever->ip = thisOne.ip;
|
||||
mdStdDevIdx = newCliever->mdStdDevIdx;
|
||||
}
|
||||
else delete newCliever;
|
||||
break;
|
||||
case MACHINE:
|
||||
thisKind = MACHINE;
|
||||
kind = "machine";
|
||||
newMachine = new mdMachine();
|
||||
if (resultat = newMachine = newMachine->validateClient( maybe, thisOne, *result )) {
|
||||
theMachine = newMachine;
|
||||
newMachine->ip = thisOne.ip;
|
||||
mdStdDevIdx = MAX_CLIEVER + thisOne.dg.hdr.dgType.clieverGroup;
|
||||
}
|
||||
else delete newMachine;
|
||||
break;
|
||||
case MDDEV_INSTRUMENT:
|
||||
thisKind = MDDEV_INSTRUMENT;
|
||||
kind = "instrument";
|
||||
newInstrument = new mdInstrument();
|
||||
if (!(resultat = newInstrument = newInstrument->validateClient( maybe, thisOne, *result )))
|
||||
delete newInstrument;
|
||||
else {newInstrument->ip = thisOne.ip;
|
||||
for (i=0;i<MAX_INSTRUMENTS && thisService->instruments[i];i++);
|
||||
thisService->instruments[i] = maybe;
|
||||
mdStdDevIdx = 2+i;
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
outcome = resultat ? "succeeded" : "failed";
|
||||
theseLogs->logN(2,"The %s instantiation request %s.",kind,outcome);
|
||||
|
||||
if (!resultat) { result->reply.dg.hdr.dgType.value = false;
|
||||
result->mdStdDevIdx = MDDEV_MD;
|
||||
}
|
||||
else { newAllMap = new mdLiveClient();
|
||||
newAllMap->devType = thisKind;
|
||||
newAllMap->mdStdDevIdx = mdStdDevIdx;
|
||||
thisConfig->allClients[maybe] = newAllMap;
|
||||
result->reply.dg.hdr.sinkHandle = maybe;
|
||||
result->mdStdDevIdx = mdStdDevIdx;
|
||||
cb[mdStdDevIdx] = newControlBlock = new mdCB;
|
||||
newControlBlock->handle = maybe;
|
||||
}
|
||||
|
||||
result->send();
|
||||
|
||||
}
|
||||
void mdMachine::registerCmd(const char *cmdName,const mdIncoming &thisOne) {
|
||||
|
||||
const char *msg;
|
||||
char *name;
|
||||
int value = OK;
|
||||
std::string arg = std::string(cmdName);
|
||||
std::map<int,mdLiveClient*>::iterator iter = thisConfig->allClients.find(thisOne.dg.hdr.handle);
|
||||
mdResponse *result = new mdResponse(thisService->bg,thisOne.ip);
|
||||
|
||||
result->reply.dg.hdr = thisOne.dg.hdr;
|
||||
result->dCat = DV_MDQUERY;
|
||||
|
||||
if( iter == thisConfig->allClients.end() ) {
|
||||
theseLogs->logN(1,"Cmd reg for device whose handle (%d) absent, ignored.", thisOne.dg.hdr.handle );
|
||||
value = MDERR_NOTREADY;
|
||||
goto done;
|
||||
}
|
||||
|
||||
result->mdStdDevIdx = iter->second->mdStdDevIdx;
|
||||
|
||||
if (cmds.empty()) {
|
||||
theseLogs->logN(1,"attempt to register '%s' but device not ready to accept command registration.",cmdName);
|
||||
value = MDERR_NOTREADY;
|
||||
goto done;
|
||||
}
|
||||
|
||||
if( cmds.find(arg) == cmds.end() ) {
|
||||
theseLogs->logN(1,"attempt to register '%s' whose rules has not yet been defined.",cmdName);
|
||||
value = MDERR_MISSING;
|
||||
goto done;
|
||||
}
|
||||
|
||||
// Currently presumes SCPI.
|
||||
|
||||
if (cmds[arg]->getHandler())
|
||||
{value = MDERR_CONFLICT; goto done;}
|
||||
else{
|
||||
cmds[arg]->setHandler(cmds[arg]);
|
||||
result->reply.dg.hdr.dgSubType = MDDG_REGSCPI;
|
||||
}
|
||||
|
||||
done:
|
||||
|
||||
if (value == OK) {
|
||||
msg = cmdName;
|
||||
result->reply.dg.hdr.dgType.value = 1;
|
||||
}
|
||||
else msg = thisConfig->err[value];
|
||||
|
||||
result->reply.dg.hdr.msgType = MDDG_MDQUERY;
|
||||
name = (char *)(&result->reply.dg.payLoad[0] + result->reply.dg.hdr.primeOffset);
|
||||
|
||||
strcpy(name,msg);
|
||||
result->reply.dg.hdr.payloadSize = result->reply.dg.hdr.primeOffset + strlen(name) + 1;
|
||||
result->send();
|
||||
|
||||
}
|
||||
//
|
||||
// TODO: figure out how to avoid this duplication in gcc and msvc 10.
|
||||
//
|
||||
void mdInstrument::registerCmd(const char *cmdName,const mdIncoming &thisOne) {
|
||||
|
||||
const char *msg;
|
||||
char *name;
|
||||
int value = OK;
|
||||
std::string arg = std::string(cmdName);
|
||||
std::map<int,mdLiveClient*>::iterator iter = thisConfig->allClients.find(thisOne.dg.hdr.handle);
|
||||
mdResponse *result = new mdResponse(thisService->bg,thisOne.ip);
|
||||
|
||||
result->reply.dg.hdr = thisOne.dg.hdr;
|
||||
result->dCat = DV_MDQUERY;
|
||||
|
||||
if( iter == thisConfig->allClients.end() ) {
|
||||
theseLogs->logN(1,"Cmd reg for device whose handle (%d) absent, ignored.", thisOne.dg.hdr.handle );
|
||||
value = MDERR_NOTREADY;
|
||||
goto done;
|
||||
}
|
||||
|
||||
result->mdStdDevIdx = iter->second->mdStdDevIdx;
|
||||
|
||||
if (cmds.empty()) {
|
||||
theseLogs->logN(1,"attempt to register '%s' but device not ready to accept command registration.",cmdName);
|
||||
value = MDERR_NOTREADY;
|
||||
goto done;
|
||||
}
|
||||
|
||||
if( cmds.find(arg) == cmds.end() ) {
|
||||
theseLogs->logN(1,"attempt to register '%s' which has no rules basis.",cmdName);
|
||||
value = MDERR_MISSING;
|
||||
goto done;
|
||||
}
|
||||
|
||||
// Currently presumes SCPI.
|
||||
|
||||
if (cmds[arg]->getHandler())
|
||||
{value = MDERR_CONFLICT; goto done;}
|
||||
else{
|
||||
cmds[arg]->setHandler(cmds[arg]);
|
||||
result->reply.dg.hdr.dgSubType = MDDG_REGSCPI;
|
||||
}
|
||||
|
||||
done:
|
||||
|
||||
if (value == OK) {
|
||||
msg = cmdName;
|
||||
result->reply.dg.hdr.dgType.value = 1;
|
||||
}
|
||||
else msg = thisConfig->err[value];
|
||||
|
||||
result->reply.dg.hdr.msgType = MDDG_MDQUERY;
|
||||
name = (char *)(&result->reply.dg.payLoad[0] + result->reply.dg.hdr.primeOffset);
|
||||
|
||||
strcpy(name,msg);
|
||||
result->reply.dg.hdr.payloadSize = result->reply.dg.hdr.primeOffset + strlen(name) + 1;
|
||||
result->send();
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue