clientlib/nanoprobe.c

Go to the documentation of this file.

#include <projectcommon.h>
#include <string.h>
#include <frameset.h>
#include <framesettypes.h>
#include <frametypes.h>
#include <compressframe.h>
#include <cryptframe.h>
#include <intframe.h>
#include <cstringframe.h>
#include <addrframe.h>
#include <ipportframe.h>
#include <seqnoframe.h>
#include <packetdecoder.h>
#include <netgsource.h>
#include <reliableudp.h>
#include <authlistener.h>
#include <nvpairframe.h>
#include <hblistener.h>
#include <hbsender.h>
#include <configcontext.h>
#include <pcap_min.h>
#include <jsondiscovery.h>
#include <switchdiscovery.h>
#include <fsprotocol.h>
#include <resourcecmd.h>
#include <resourcequeue.h>
#include <misc.h>
#include <nanoprobe.h>


void (*nanoprobe_deadtime_agent)(HbListener*)                   = NULL;
void (*nanoprobe_heartbeat_agent)(HbListener*)                  = NULL;
void (*nanoprobe_warntime_agent)(HbListener*, guint64 howlate)  = NULL;
void (*nanoprobe_comealive_agent)(HbListener*, guint64 howlate) = NULL;
WINEXPORT NanoHbStats   nano_hbstats = {0U, 0U, 0U, 0U, 0U};
gboolean                nano_connected = FALSE;
WINEXPORT int           errcount = 0;
WINEXPORT GMainLoop*    mainloop = NULL;

FSTATIC void            nanoobey_sendexpecthb(AuthListener*, FrameSet* fs, NetAddr*);
FSTATIC void            nanoobey_sendhb(AuthListener*, FrameSet* fs, NetAddr*);
FSTATIC void            nanoobey_expecthb(AuthListener*, FrameSet* fs, NetAddr*);
FSTATIC void            nanoobey_stopsendexpecthb(AuthListener*, FrameSet* fs, NetAddr*);
FSTATIC void            nanoobey_stopsendhb(AuthListener*, FrameSet* fs, NetAddr*);
FSTATIC void            nanoobey_stopexpecthb(AuthListener*, FrameSet* fs, NetAddr*);
FSTATIC void            nanoobey_setconfig(AuthListener*, FrameSet* fs, NetAddr*);
FSTATIC void            nanoobey_change_debug(gint plusminus, AuthListener*, FrameSet*, NetAddr*);
FSTATIC void            nanoobey_incrdebug(AuthListener*, FrameSet*, NetAddr*);
FSTATIC void            nanoobey_decrdebug(AuthListener*, FrameSet*, NetAddr*);
FSTATIC void            nanoobey_startdiscover(AuthListener*, FrameSet*, NetAddr*);
FSTATIC void            nanoobey_stopdiscover(AuthListener*, FrameSet*, NetAddr*);
FSTATIC void            nanoobey_dorscoperation(AuthListener*, FrameSet*, NetAddr*);
FSTATIC void            _nano_send_rscexitstatus(ConfigContext* request, gpointer user_data
,                               enum HowDied reason, int rc, int signal, gboolean core_dumped
,                               const char * stringresult);
FSTATIC void            nanoobey_cancelrscoperation(AuthListener*, FrameSet*, NetAddr*);
FSTATIC void            nano_schedule_discovery(const char *name, guint32 interval,const char* json
                        ,       ConfigContext*, NetGSource* transport, NetAddr* fromaddr);
FSTATIC void            nano_stop_discovery(const char * discoveryname, NetGSource*, NetAddr*);
FSTATIC gboolean        nano_startupidle(gpointer gcruft);
FSTATIC gboolean        nano_reqconfig(gpointer gcruft);
FSTATIC void            _real_heartbeat_agent(HbListener* who);
FSTATIC void            _real_deadtime_agent(HbListener* who);
FSTATIC void            _real_warntime_agent(HbListener* who, guint64 howlate);
FSTATIC void            _real_comealive_agent(HbListener* who, guint64 howlate);
FSTATIC void            _real_martian_agent(NetAddr* who);
FSTATIC HbListener*     _real_hblistener_new(NetAddr*, ConfigContext*);
FSTATIC gboolean        _nano_final_shutdown(gpointer unused);
FSTATIC gboolean        shutdown_when_outdone(gpointer unused);

HbListener* (*nanoprobe_hblistener_new)(NetAddr*, ConfigContext*) = _real_hblistener_new;

gboolean                nano_shutting_down = FALSE;
const char *            procname = "nanoprobe";

static NetAddr*         nanofailreportaddr = NULL;
static NetGSource*      nanotransport = NULL;
static guint            idle_shutdown_gsource = 0;
static ResourceQueue*   RscQ = NULL;

DEBUGDECLARATIONS

FSTATIC HbListener*
_real_hblistener_new(NetAddr* addr, ConfigContext* context)
{
        return hblistener_new(addr, context, 0);
}

void
nanoprobe_report_upstream(guint16 reporttype    
,                         NetAddr* who          
,                         const char * systemnm 
,                         guint64 howlate)      

{
                FrameSet*       fs;

                if (nano_shutting_down || NULL == nanofailreportaddr) {
                        DEBUGMSG("%s.%d: Ignoring request to send fstype=%d message upstream [%s]."
                        ,       __FUNCTION__, __LINE__, reporttype
                        ,       (nano_shutting_down ? "shutting down" : "not connected to CMA"));
                        return;
                }

                fs              = frameset_new(reporttype);
                // Construct and send a frameset reporting this event...
                if (howlate > 0) {
                        IntFrame*       lateframe       = intframe_new(FRAMETYPE_ELAPSEDTIME, 8);
                        lateframe->setint(lateframe, howlate);
                        frameset_append_frame(fs, &lateframe->baseclass);
                        UNREF2(lateframe);
                }
                // Add the address - if any...
                if (who != NULL) {
                        IpPortFrame*    peeraddr        = ipportframe_netaddr_new(FRAMETYPE_IPPORT, who);
                        frameset_append_frame(fs, &peeraddr->baseclass);
                        UNREF2(peeraddr);
                }
                // Add the system name - if any...
                if (systemnm != NULL) {
                        CstringFrame* usf = cstringframe_new(FRAMETYPE_HOSTNAME, 0);
                        usf->baseclass.setvalue(&usf->baseclass, g_strdup(systemnm), strlen(systemnm)+1
                        ,                       frame_default_valuefinalize);
                        frameset_append_frame(fs, &usf->baseclass);
                        UNREF2(usf);
                }
                DEBUGMSG3("%s - sending frameset of type %d", __FUNCTION__, reporttype);
                DUMP3("nanoprobe_report_upstream", &nanofailreportaddr->baseclass, NULL);
                nanotransport->_netio->sendareliablefs(nanotransport->_netio, nanofailreportaddr, DEFAULT_FSP_QID, fs);
                UNREF(fs);
}


FSTATIC void
_real_martian_agent(NetAddr* who)
{
        static guint64          last_martian_time = 0;          // microseconds
        static guint            recent_martian_count = 0;
        guint64                 now = g_get_monotonic_time();   // microseconds
        const guint64           uS = 1000000;

        ++nano_hbstats.martian_count;

        // If it's been more than MARTIAN_TIMEOUT seconds since the last
        // martian, then reset the count of recent martians
        if (now > (last_martian_time + (MARTIAN_TIMEOUT*uS))) {
                recent_martian_count = 0;       
        }

        last_martian_time = now;
        ++recent_martian_count;

        // This means if we only get one martian then none, we say nothing
        // This can happen as a result of timing - and it's OK.
        // But if we get more than one, we complain then and once every 10 afterwards
        if ((recent_martian_count % 10) == 2) {
                char *          addrstring;

                addrstring = who->baseclass.toString(who);
                g_warning("System at address %s is sending unexpected heartbeats.", addrstring);
                g_free(addrstring);

                nanoprobe_report_upstream(FRAMESETTYPE_HBMARTIAN, who, NULL, 0);
        }
}
FSTATIC void
_real_deadtime_agent(HbListener* who)
{
        ++nano_hbstats.dead_count;
        if (nanoprobe_deadtime_agent) {
                nanoprobe_deadtime_agent(who);
        }else{
                char *          addrstring;

                addrstring = who->listenaddr->baseclass.toString(who->listenaddr);
                g_warning("Peer at address %s is dead (has timed out).", addrstring);
                g_free(addrstring);

                nanoprobe_report_upstream(FRAMESETTYPE_HBDEAD, who->listenaddr, NULL, 0);
        }
}

FSTATIC void
_real_heartbeat_agent(HbListener* who)
{
        ++nano_hbstats.heartbeat_count;
        if (nanoprobe_heartbeat_agent) {
                nanoprobe_heartbeat_agent(who);
        }
}


FSTATIC void
_real_warntime_agent(HbListener* who, guint64 howlate)
{
        ++nano_hbstats.warntime_count;
        if (nanoprobe_warntime_agent) {
                nanoprobe_warntime_agent(who, howlate);
        }else{
                char *  addrstring;
                guint64 mslate = howlate / 1000;
                addrstring = who->listenaddr->baseclass.toString(who->listenaddr);
                g_warning("Heartbeat from peer at address %s was "FMT_64BIT"d ms late.", addrstring, mslate);
                g_free(addrstring);
                nanoprobe_report_upstream(FRAMESETTYPE_HBLATE, who->listenaddr, NULL, howlate);
        }
}
FSTATIC void
_real_comealive_agent(HbListener* who, guint64 howlate)
{
        ++nano_hbstats.comealive_count;
        if (nanoprobe_comealive_agent) {
                nanoprobe_comealive_agent(who, howlate);
        }else{
                char *  addrstring;
                double secsdead = ((double)((howlate+50000) / 100000))/10.0; // Round to nearest tenth of a second
                addrstring = who->listenaddr->baseclass.toString(who->listenaddr);
                g_warning("Peer at address %s came alive after being dead for %g seconds.", addrstring, secsdead);
                g_free(addrstring);
                nanoprobe_report_upstream(FRAMESETTYPE_HBBACKALIVE, who->listenaddr, NULL, howlate);
        }
}

void
nanoobey_sendhb(AuthListener* parent    
        ,   FrameSet*   fs              
        ,   NetAddr*    fromaddr)       
{

        GSList*         slframe;
        guint           addrcount = 0;
        ConfigContext*  config = parent->baseclass.config;
        guint16         sendinterval = 0;

        if (nano_shutting_down) {
                return;
        }

        g_return_if_fail(fs != NULL);
        (void)fromaddr;
        
        if (config->getint(config, CONFIGNAME_HBTIME) > 0) {
                sendinterval = config->getint(config, CONFIGNAME_HBTIME);
        }

        for (slframe = fs->framelist; slframe != NULL; slframe = g_slist_next(slframe)) {
                Frame* frame = CASTTOCLASS(Frame, slframe->data);
                int     frametype = frame->type;
                switch(frametype) {
                        IntFrame*       iframe;
                        IpPortFrame*    aframe;
                        HbSender*       hb;

                        case FRAMETYPE_HBINTERVAL:
                                iframe = CASTTOCLASS(IntFrame, frame);
                                sendinterval = (guint16) iframe->getint(iframe);
                                break;
                        case FRAMETYPE_IPPORT:
                                if (0 == sendinterval) {
                                        g_warning("Send interval is zero in %s", __FUNCTION__);
                                        continue;
                                }
                                aframe = CASTTOCLASS(IpPortFrame, frame);
                                addrcount++;
                                hb = hbsender_new(aframe->getnetaddr(aframe), parent->baseclass.transport
                                ,       sendinterval, 0);
                                (void)hb;
                                break;
                }
        }
}
void
nanoobey_expecthb(AuthListener* parent  
        ,   FrameSet*   fs              
        ,   NetAddr*    fromaddr)       
{

        GSList*         slframe;
        ConfigContext*  config = parent->baseclass.config;
        guint           addrcount = 0;

        guint64         deadtime = 0;
        guint64         warntime = 0;

        (void)fromaddr;

        if (nano_shutting_down) {
                return;
        }

        g_return_if_fail(fs != NULL);
        if (config->getint(config, CONFIGNAME_DEADTIME) > 0) {
                deadtime = config->getint(config, CONFIGNAME_DEADTIME);
        }
        if (config->getint(config, CONFIGNAME_WARNTIME) > 0) {
                warntime = config->getint(config, CONFIGNAME_WARNTIME);
        }

        for (slframe = fs->framelist; slframe != NULL; slframe = g_slist_next(slframe)) {
                Frame* frame = CASTTOCLASS(Frame, slframe->data);
                int     frametype = frame->type;
                switch(frametype) {
                        IntFrame*       iframe;

                        case FRAMETYPE_HBDEADTIME:
                                iframe = CASTTOCLASS(IntFrame, frame);
                                deadtime = iframe->getint(iframe);
                                break;

                        case FRAMETYPE_HBWARNTIME:
                                iframe = CASTTOCLASS(IntFrame, frame);
                                warntime = iframe->getint(iframe);
                                break;

                        case FRAMETYPE_IPPORT: {
                                HbListener*     hblisten;
                                IpPortFrame*    aframe;
                                NetGSource*     transport = parent->baseclass.transport;
                                aframe = CASTTOCLASS(IpPortFrame, frame);
                                addrcount++;
                                hblisten = hblistener_new(aframe->getnetaddr(aframe), config, 0);
                                hblisten->baseclass.associate(&hblisten->baseclass, transport);
                                if (deadtime > 0) {
                                        // Otherwise we get the default deadtime
                                        hblisten->set_deadtime(hblisten, deadtime);
                                }
                                if (warntime > 0) {
                                        // Otherwise we get the default warntime
                                        hblisten->set_warntime(hblisten, warntime);
                                }
                                hblisten->set_deadtime_callback(hblisten, _real_deadtime_agent);
                                hblisten->set_heartbeat_callback(hblisten, _real_heartbeat_agent);
                                hblisten->set_warntime_callback(hblisten, _real_warntime_agent);
                                hblisten->set_comealive_callback(hblisten, _real_comealive_agent);
                                // Intercept incoming heartbeat packets
                                transport->addListener(transport, FRAMESETTYPE_HEARTBEAT
                                ,                   &hblisten->baseclass);
                                // Unref this heartbeat listener, and forget our reference.
                                UNREF2(hblisten);
                                /*
                                 * That still leaves two references to 'hblisten':
                                 *   - in the transport dispatch table
                                 *   - in the global heartbeat listener table
                                 * And one reference to the previous 'hblisten' object:
                                 *   - in the global heartbeat listener table
                                 * Also note that we become the 'proxy' for all incoming heartbeats
                                 * but we dispatch them to the right HbListener object.
                                 * Since we've become the proxy for all incoming heartbeats, if
                                 * we displace and free the old proxy, this all still works nicely,
                                 * because the transport object gets rid of its old reference to the
                                 * old 'proxy' object.
                                 */
                        }
                        break;
                }
        }
}

void
nanoobey_sendexpecthb(AuthListener* parent      
        ,   FrameSet*   fs              
        ,   NetAddr*    fromaddr)       
{
        g_return_if_fail(fs != NULL && fs->fstype == FRAMESETTYPE_SENDEXPECTHB);

        if (nano_shutting_down) {
                return;
        }
        // This will cause us to ACK the packet twice -- not a problem...
        nanoobey_sendhb  (parent, fs, fromaddr);
        nanoobey_expecthb(parent, fs, fromaddr);
}
void
nanoobey_stopsendhb(AuthListener* parent
        ,   FrameSet*   fs              
        ,   NetAddr*    fromaddr)       
{
        GSList*         slframe;
        (void)parent;
        (void)fromaddr;

        for (slframe = fs->framelist; slframe != NULL; slframe = g_slist_next(slframe)) {
                Frame* frame = CASTTOCLASS(Frame, slframe->data);
                switch(frame->type) {
                        case FRAMETYPE_IPPORT: {
                                // This is _so_ much simpler than the code to send them ;-)
                                IpPortFrame*    aframe = CASTTOCLASS(IpPortFrame, frame);
                                hbsender_stopsend(aframe->getnetaddr(aframe));
                                break;
                        }
                }//endswitch
        }//endfor
}

void
nanoobey_stopexpecthb(AuthListener* parent
        ,   FrameSet*   fs              
        ,   NetAddr*    fromaddr)       
{
        GSList*         slframe;
        (void)parent;
        (void)fromaddr;

        for (slframe = fs->framelist; slframe != NULL; slframe = g_slist_next(slframe)) {
                Frame* frame = CASTTOCLASS(Frame, slframe->data);
                switch(frame->type) {
                        case FRAMETYPE_IPPORT: {
                                // This is _so_ much simpler than the code to listen for heartbeats...
                                IpPortFrame*    aframe = CASTTOCLASS(IpPortFrame, frame);
                                NetAddr*        destaddr = aframe->getnetaddr(aframe);
                                NetIO*          transport = parent->baseclass.transport->_netio;
                                hblistener_unlisten(destaddr);
                                transport->closeconn(transport, DEFAULT_FSP_QID, destaddr);
                                break;
                        }
                }//endswitch
        }//endfor
}

void
nanoobey_stopsendexpecthb(AuthListener* parent
        ,   FrameSet*   fs              
        ,   NetAddr*    fromaddr)       
{
        nanoobey_stopexpecthb(parent, fs, fromaddr);
        nanoobey_stopsendhb  (parent, fs, fromaddr);
}

/*
 * Act on (obey) a <b>FRAMESETTYPE_SETCONFIG</b> @ref FrameSet.
 * This frameset is sent during the initial configuration phase.
 * It contains name value pairs to save into our configuration (ConfigContext).
 * These might be {string,string} pairs or {string,ipaddr} pairs, or
 * {string, integer} pairs.  We process them all.
 * The frame types that we receive for these are:
 * <b>FRAMETYPE_PARAMNAME</b> - parameter name to set
 * <b>FRAMETYPE_CSTRINGVAL</b> - string value to associate with name
 * <b>FRAMETYPE_CINTVAL</b> - integer value to associate with naem
 * <b>FRAMETYPE_IPPORT</b> - IP address to associate with name
 */
void
nanoobey_setconfig(AuthListener* parent 
        ,      FrameSet*        fs      
        ,      NetAddr* fromaddr)       
{
        GSList*         slframe;
        ConfigContext*  cfg = parent->baseclass.config;
        char *          paramname = NULL;

        (void)fromaddr;

        for (slframe = fs->framelist; slframe != NULL; slframe = g_slist_next(slframe)) {
                Frame* frame = CASTTOCLASS(Frame, slframe->data);
                int     frametype = frame->type;
                switch (frametype) {

                        case FRAMETYPE_PARAMNAME: { // Parameter name to set
                                paramname = frame->value;
                                g_return_if_fail(paramname != NULL);
                        }
                        break;

                        case FRAMETYPE_CSTRINGVAL: { // String value to set 'paramname' to
                                g_return_if_fail(paramname != NULL);
                                cfg->setstring(cfg, paramname, frame->value);
                                paramname = NULL;
                        }
                        break;

                        case FRAMETYPE_CINTVAL: { // Integer value to set 'paramname' to
                                IntFrame* intf = CASTTOCLASS(IntFrame, frame);
                                g_return_if_fail(paramname != NULL);
                                cfg->setint(cfg, paramname, (gint)intf->getint(intf));
                                paramname = NULL;
                        }
                        break;


                        case FRAMETYPE_IPPORT: { // NetAddr value to set 'paramname' to
                                IpPortFrame* af = CASTTOCLASS(IpPortFrame, frame);
                                NetAddr*        addr = af->getnetaddr(af);
                                g_return_if_fail(paramname != NULL);
                                cfg->setaddr(cfg, paramname, addr);
                                paramname = NULL;
                        }
                        break;

                }//endswitch
        }//endfor
        DUMP3("nanoobey_setconfig: cfg is", &cfg->baseclass, NULL);
        if (cfg->getaddr(cfg, CONFIGNAME_CMAFAIL) != NULL) {
                if (nanofailreportaddr == NULL) {
                        nanofailreportaddr = cfg->getaddr(cfg, CONFIGNAME_CMAFAIL);
                        
                        
                }else if (cfg->getaddr(cfg, CONFIGNAME_CMAFAIL) != nanofailreportaddr) {
                        UNREF(nanofailreportaddr);
                        nanofailreportaddr = cfg->getaddr(cfg, CONFIGNAME_CMAFAIL);
                }
                DUMP3("nanoobey_setconfig: nanofailreportaddr", &nanofailreportaddr->baseclass, NULL);
                {
                        // Alias localhost to the CMA nanofailreportaddr (at least for now...)

                        NetAddr* localhost = netaddr_string_new("127.0.0.1");
                        NetIO* io = parent->baseclass.transport->_netio;
                        io->addalias(io, localhost, nanofailreportaddr);
                        UNREF(localhost);
                }
                REF(nanofailreportaddr);
        }
        g_message("Connected to CMA.  Happiness :-D");
        nano_connected = TRUE;
}//nanoobey_setconfig

FSTATIC void
nanoobey_change_debug(gint plusminus            
        ,             AuthListener* parent      
        ,             FrameSet* fs              
        ,             NetAddr*  fromaddr)       
{

        GSList*         slframe;
        guint           changecount = 0;

        (void)parent;
        (void)fromaddr;

        for (slframe = fs->framelist; slframe != NULL; slframe = g_slist_next(slframe)) {
                Frame* frame = CASTTOCLASS(Frame, slframe->data);
                int     frametype = frame->type;
                switch (frametype) {
                        case FRAMETYPE_CSTRINGVAL: { // String value to set 'paramname' to
                                ++changecount;
                                if (plusminus < 0) {
                                        proj_class_decr_debug((char*)frame->value);
                                }else{
                                        proj_class_incr_debug((char*)frame->value);
                                }
                        }
                        break;
                }
        }
        if (changecount == 0) {
                if (plusminus < 0) {
                        proj_class_decr_debug(NULL);
                }else{
                        proj_class_incr_debug(NULL);
                }
        }
}
FSTATIC void
nanoobey_incrdebug(AuthListener* parent 
        ,             FrameSet* fs              
        ,             NetAddr*  fromaddr)       
{
        nanoobey_change_debug(+1, parent, fs, fromaddr);
}
FSTATIC void
nanoobey_decrdebug(AuthListener* parent 
        ,             FrameSet* fs              
        ,             NetAddr*  fromaddr)       
{
        nanoobey_change_debug(-1, parent, fs, fromaddr);
}

FSTATIC void
nanoobey_startdiscover(AuthListener* parent     
        ,             FrameSet* fs              
        ,             NetAddr*  fromaddr)       
{

        GSList*         slframe;
        guint           interval = 0;
        const char *    discoveryname = NULL;

        (void)parent;
        (void)fromaddr;
        
        if (nano_shutting_down) {
                return;
        }

        DEBUGMSG3("%s - got frameset", __FUNCTION__);
        // Loop over the frames, looking for those we know what to do with ;-)
        for (slframe = fs->framelist; slframe != NULL; slframe = g_slist_next(slframe)) {
                Frame* frame = CASTTOCLASS(Frame, slframe->data);
                int     frametype = frame->type;

                switch (frametype) {
                        case FRAMETYPE_DISCNAME: { // Discovery instance name
                                CstringFrame* strf = CASTTOCLASS(CstringFrame, frame);
                                g_return_if_fail(strf != NULL);
                                g_return_if_fail(discoveryname == NULL);
                                discoveryname = strf->baseclass.value;
                                DEBUGMSG3("%s - got DISCOVERYNAME %s", __FUNCTION__, discoveryname);
                        }
                        break;

                        case FRAMETYPE_DISCINTERVAL: { // Discovery interval
                                IntFrame* intf = CASTTOCLASS(IntFrame, frame);
                                interval = (guint)intf->getint(intf);
                                DEBUGMSG3("%s - got DISCOVERYINTERVAL %d", __FUNCTION__, interval);
                        }
                        break;

                        case FRAMETYPE_DISCJSON: { // Discovery JSON string (parameters)
                                CstringFrame* strf = CASTTOCLASS(CstringFrame, frame);
                                const char *  jsonstring;
                                g_return_if_fail(strf != NULL);
                                jsonstring = strf->baseclass.value;
                                g_return_if_fail(discoveryname != NULL);
                                DEBUGMSG3("Got DISCJSON frame: %s %d %s" , discoveryname, interval, jsonstring);
                                nano_schedule_discovery(discoveryname, interval, jsonstring
                                ,                       parent->baseclass.config
                                ,                       parent->baseclass.transport
                                ,                       fromaddr);
                        }
                        interval = 0;
                        discoveryname = NULL;
                        break;

                }
        }
}

FSTATIC void
_nano_send_rscexitstatus(ConfigContext* request, gpointer user_data
,                       enum HowDied reason, int rc, int signal, gboolean core_dumped
,                       const char * stringresult)
{
        NetGSource*     transport = CASTTOCLASS(NetGSource, user_data);
        ConfigContext*  response = configcontext_new(0);
        FrameSet*       fs = frameset_new(FRAMESETTYPE_RSCOPREPLY);
        CstringFrame*   sf = cstringframe_new(FRAMETYPE_RSCJSONREPLY, 0);
        char*           rsp_json;

        struct {
                const char*     framename;
                int             framevalue;
        } pktframes[] = {
                {REQREASONENUMNAMEFIELD,        reason},
                {REQRCNAMEFIELD,                rc},
                {REQSIGNALNAMEFIELD,            signal},
        };
        unsigned        j;

        for (j=0; j < DIMOF(pktframes); ++j) {
                response->setint(response, pktframes[j].framename, pktframes[j].framevalue);
        }
        response->setbool(response, REQCOREDUMPNAMEFIELD, core_dumped);
        if (stringresult) {
                response->setstring(response, REQSTRINGRETNAMEFIELD, stringresult);
        }
        // Copy the request ID over from the original request
        response->setint(response, REQIDENTIFIERNAMEFIELD
        ,       request->getint(request, REQIDENTIFIERNAMEFIELD));
        rsp_json = response->baseclass.toString(&response->baseclass);
        UNREF(response);
        sf->baseclass.setvalue(&sf->baseclass, rsp_json, strlen(rsp_json)+1, g_free);
        frameset_append_frame(fs, &sf->baseclass);
        UNREF2(sf);
        transport->_netio->sendareliablefs(transport->_netio, nanofailreportaddr, DEFAULT_FSP_QID, fs);
        UNREF(fs);
}
FSTATIC void
nanoobey_dorscoperation(AuthListener* parent, FrameSet* fs, NetAddr*fromaddr)
{
        GSList*         slframe;

        (void)parent;
        (void)fromaddr;
        if (nano_shutting_down) {
                return;
        }
        if (NULL == RscQ) {
                RscQ = resourcequeue_new(0);
        }
        // Loop over the frames, looking for those we know what to do with ;-)
        for (slframe = fs->framelist; slframe != NULL; slframe = g_slist_next(slframe)) {
                Frame* frame = CASTTOCLASS(Frame, slframe->data);
                CstringFrame* csframe;
                ConfigContext*  cfg;
                if (frame->type != FRAMETYPE_RSCJSON) {
                        continue;
                }
                csframe = CASTTOCLASS(CstringFrame, frame);
                cfg = configcontext_new_JSON_string(csframe->baseclass.value);
                if (NULL == cfg) {
                        g_warning("%s.%d: Received malformed JSON string [%*s]"
                        ,       __FUNCTION__, __LINE__
                        ,       csframe->baseclass.length-1
                        ,       (char*)csframe->baseclass.value);
                        continue;
                }
                RscQ->Qcmd(RscQ, cfg, _nano_send_rscexitstatus, nanotransport);
                UNREF(cfg);
        }
}

FSTATIC void
nanoobey_cancelrscoperation(AuthListener* parent, FrameSet* fs, NetAddr* fromaddr)
{
        GSList*         slframe;

        (void)parent;
        (void)fromaddr;
        if (NULL == RscQ) {
                RscQ = resourcequeue_new(0);
        }

        for (slframe = fs->framelist; slframe != NULL; slframe = g_slist_next(slframe)) {
                Frame* frame = CASTTOCLASS(Frame, slframe->data);
                CstringFrame* csframe;
                ConfigContext*  cfg;
                if (frame->type != FRAMETYPE_RSCJSON) {
                        continue;
                }
                csframe = CASTTOCLASS(CstringFrame, frame);
                cfg = configcontext_new_JSON_string(csframe->baseclass.value);
                if (NULL == cfg) {
                        g_warning("%s.%d: Received malformed JSON string [%*s]"
                        ,       __FUNCTION__, __LINE__
                        ,       csframe->baseclass.length-1
                        ,       (char*)csframe->baseclass.value);
                        continue;
                }
                RscQ->cancel(RscQ, cfg);
                UNREF(cfg);
        }
}

FSTATIC void
nanoobey_stopdiscover(AuthListener* parent      
        ,             FrameSet* fs              
        ,             NetAddr*  fromaddr)       
{

        GSList*         slframe;

        (void)parent;
        (void)fromaddr;
        

        // Loop over the frames, looking for the one we know what to do with ;-)
        for (slframe = fs->framelist; slframe != NULL; slframe = g_slist_next(slframe)) {
                Frame* frame = CASTTOCLASS(Frame, slframe->data);
                int     frametype = frame->type;

                switch (frametype) {
                        case FRAMETYPE_DISCNAME: { // Discovery instance name
                                CstringFrame* strf = CASTTOCLASS(CstringFrame, frame);
                                const char *    discoveryname;
                                g_return_if_fail(strf != NULL);
                                discoveryname = strf->baseclass.value;
                                g_return_if_fail(discoveryname == NULL);
                                discovery_unregister(discoveryname);
                        }
                        break;

                }
        }
}


FSTATIC void    
nano_schedule_discovery(const char *instance,   
                        guint32 interval,       
                        const char* json,       
                        ConfigContext* config,  
                        NetGSource* transport,  
                        NetAddr* fromaddr)      
{
        ConfigContext*  jsonroot;
        JsonDiscovery*  discovery;
        const char*     disctype;

        (void)fromaddr;

        DEBUGMSG3("%s(%s,%d,%s)", __FUNCTION__, instance, interval, json);
        jsonroot = configcontext_new_JSON_string(json);
        g_return_if_fail(jsonroot != NULL);
        disctype = jsonroot->getstring(jsonroot, "type");
        g_return_if_fail(disctype != NULL);
        discovery = jsondiscovery_new(disctype, instance, interval, jsonroot
        ,                             transport, config, 0);
        UNREF(jsonroot);
        UNREF2(discovery);
}

struct startup_cruft {
        const char *    initdiscover;
        int             discover_interval;
        NetGSource*     iosource;
        ConfigContext*  context;
};

gboolean
nano_startupidle(gpointer gcruft)
{
        static enum istate {INIT=3, WAIT=5, DONE=7} state = INIT;
        struct startup_cruft* cruft = gcruft;
        const char *    cfgname = cruft->initdiscover;

        if (state == DONE || nano_shutting_down) {
                return FALSE;
        }
        if (state == INIT) {
                const char * jsontext = "{\"parameters\":{}}";
                ConfigContext*  jsondata = configcontext_new_JSON_string(jsontext);
                JsonDiscovery* jd = jsondiscovery_new
                (       cruft->initdiscover
                ,       cruft->initdiscover
                ,       cruft->discover_interval
                ,       jsondata
                ,       cruft->iosource, cruft->context, 0);
                UNREF(jsondata);
                UNREF2(jd);
                state = WAIT;
                return TRUE;
        }
        if (cruft->context->getstring(cruft->context, cfgname)) {
                state = DONE;
                // Call it once, and arrange for it to repeat until we hear back.
                g_timeout_add_seconds(5, nano_reqconfig, gcruft);
                nano_reqconfig(gcruft);
                return FALSE;
        }
        return TRUE;
}

gboolean
nano_reqconfig(gpointer gcruft)
{
        struct startup_cruft* cruft = gcruft;
        FrameSet*       fs;
        CstringFrame*   csf;
        CstringFrame*   usf;
        const char *    cfgname = cruft->initdiscover;
        ConfigContext*  context = cruft->context;
        NetAddr*        cmainit = context->getaddr(context, CONFIGNAME_CMAINIT);
        const char *            jsontext;
        char *                  sysname = NULL;

        if (nano_shutting_down) {
                return FALSE;
        }

        // We <i>have</i> to know our initial request address - or all is lost.
        // NOTE THAT THIS ADDRESS MIGHT BE MULTICAST AND MIGHT BE USED ONLY ONCE
        g_return_val_if_fail(cmainit != NULL, FALSE);

        // Our initial configuration message must contain these parameters.
        if (context->getaddr(context, CONFIGNAME_CMAADDR) != NULL
        &&  context->getaddr(context, CONFIGNAME_CMAFAIL) != NULL
        &&  context->getaddr(context, CONFIGNAME_CMADISCOVER) != NULL
        &&  context->getint(context, CONFIGNAME_CMAPORT) > 0) {
                return FALSE;
        }
        fs = frameset_new(FRAMESETTYPE_STARTUP);

        // Put in the system name
        usf = cstringframe_new(FRAMETYPE_HOSTNAME, 0);
        sysname = proj_get_sysname();
        usf->baseclass.setvalue(&usf->baseclass, g_strdup(sysname), strlen(sysname)+1
        ,                       frame_default_valuefinalize);
        frameset_append_frame(fs, &usf->baseclass);
        UNREF2(usf);

        // Put in the JSON discovery text
        jsontext = context->getstring(context, cfgname);
        csf = cstringframe_new(FRAMETYPE_JSDISCOVER, 0);
        csf->baseclass.setvalue(&csf->baseclass, g_strdup(jsontext), strlen(jsontext)+1
        ,                       frame_default_valuefinalize);

        frameset_append_frame(fs, &csf->baseclass);
        UNREF2(csf);

        // We've constructed the frameset - now send it - unreliably...
        // That's because the reply is typically from a different address
        // which would confuse the blazes out of the reliable comm code.
        cruft->iosource->sendaframeset(cruft->iosource, cmainit, fs);
        DEBUGMSG("%s.%d: Sent initial STARTUP frameset for %s."
        ,       __FUNCTION__, __LINE__, sysname);
        g_free(sysname); sysname = NULL; 
        UNREF(fs);
        return TRUE;
}

static PacketDecoder*   decoder = NULL;
static SwitchDiscovery* swdisc = NULL;
static AuthListener*    obeycollective = NULL;


ObeyFrameSetTypeMap collective_obeylist [] = {
        // This is the complete set of commands that nanoprobes know how to obey - so far...
        {FRAMESETTYPE_SENDHB,           nanoobey_sendhb},
        {FRAMESETTYPE_EXPECTHB,         nanoobey_expecthb},
        {FRAMESETTYPE_SENDEXPECTHB,     nanoobey_sendexpecthb},
        {FRAMESETTYPE_STOPSENDHB,       nanoobey_stopsendhb},
        {FRAMESETTYPE_STOPEXPECTHB,     nanoobey_stopexpecthb},
        {FRAMESETTYPE_STOPSENDEXPECTHB, nanoobey_stopsendexpecthb},
        {FRAMESETTYPE_SETCONFIG,        nanoobey_setconfig},
        {FRAMESETTYPE_INCRDEBUG,        nanoobey_incrdebug},
        {FRAMESETTYPE_DECRDEBUG,        nanoobey_decrdebug},
        {FRAMESETTYPE_DODISCOVER,       nanoobey_startdiscover},
        {FRAMESETTYPE_STOPDISCOVER,     nanoobey_stopdiscover},
        {FRAMESETTYPE_DORSCOP,          nanoobey_dorscoperation},
        {FRAMESETTYPE_STOPRSCOP,        nanoobey_cancelrscoperation},
        {0,                             NULL},
};

PacketDecoder*
nano_packet_decoder(void)
{
        static FrameTypeToFrame decodeframes[] = FRAMETYPEMAP;
        // Set up our packet decoder
        decoder = packetdecoder_new(0, decodeframes, DIMOF(decodeframes));
        return decoder;
}


WINEXPORT void
nano_start_full(const char *initdiscoverpath    
        ,       guint discover_interval         
        ,       NetGSource* io                  
        ,       ConfigContext* config)          
{
        static struct startup_cruft cruftiness;
        struct startup_cruft initcrufty = {
                initdiscoverpath,
                discover_interval,
                io,
                config
        };
        nano_shutting_down = FALSE;
        BINDDEBUG(nanoprobe_main);
        
        hblistener_set_martian_callback(_real_martian_agent);
        cruftiness = initcrufty;
        g_source_ref(CASTTOCLASS(GSource, io));
        nanotransport = io;

        // Get our local switch discovery information.
        // To be really right, we probably ought to wait until we know our local network
        // configuration - and start it up on all interfaces assigned addresses of global scope.
        swdisc = switchdiscovery_new("switchdiscovery_eth0", "eth0", ENABLE_LLDP|ENABLE_CDP, G_PRIORITY_LOW
        ,                           g_main_context_default(), io, config, 0);
        obeycollective = authlistener_new(0, collective_obeylist, config, TRUE);
        obeycollective->baseclass.associate(&obeycollective->baseclass, io);
        // Initiate the startup process
        g_idle_add(nano_startupidle, &cruftiness);
}
WINEXPORT void
nano_shutdown(gboolean report)
{
        if (report) {
                NetIOstats*     ts = &nanotransport->_netio->stats;
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of heartbeats:", nano_hbstats.heartbeat_count);
                g_info("%-35s %8d", "Count of deadtimes:", nano_hbstats.dead_count);
                g_info("%-35s %8d", "Count of warntimes:", nano_hbstats.warntime_count);
                g_info("%-35s %8d", "Count of comealives:", nano_hbstats.comealive_count);
                g_info("%-35s %8d", "Count of martians:", nano_hbstats.martian_count);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of LLDP/CDP pkts sent:", swdisc->baseclass.reportcount);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of LLDP/CDP pkts received:", swdisc->baseclass.discovercount);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of LLDP/CDP pkts received:", swdisc->baseclass.discovercount);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of recvfrom calls:", ts->recvcalls);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of pkts read:", ts->pktsread);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of framesets read:", ts->fsreads);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of sendto calls:", ts->sendcalls);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of pkts written:", ts->pktswritten);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of framesets written:", ts->fswritten);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of reliable framesets sent:", ts->reliablesends);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of reliable framesets recvd:", ts->reliablereads);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of ACKs sent:", ts->ackssent);
                g_info("%-35s %8"G_GINT64_MODIFIER"d", "Count of ACKs recvd:", ts->acksrecvd);
        }
        hbsender_stopallsenders();
        hblistener_shutdown();
        UNREF2(swdisc);
        if (RscQ) {
                RscQ->cancelall(RscQ);
                UNREF(RscQ);
        }
        if (nanofailreportaddr) {
                UNREF(nanofailreportaddr);
        }
        if (nanotransport) {
                g_source_destroy(CASTTOCLASS(GSource, nanotransport));
                g_source_unref(CASTTOCLASS(GSource, nanotransport));
                nanotransport = NULL;
        }
        // Free packet decoder
        if (decoder) {
                UNREF(decoder);
        }
        // Unregister all discovery modules.
        discovery_unregister_all();
        obeycollective->baseclass.dissociate(&obeycollective->baseclass);
        UNREF2(obeycollective);
}

WINEXPORT gboolean
nano_initiate_shutdown(void)
{

        if (nano_connected) {
                FsProtocol*     proto = CASTTOCLASS(ReliableUDP, nanotransport->_netio)->_protocol;
                char *  sysname;
                DEBUGMSG("Sending HBSHUTDOWN to CMA");
                sysname = proj_get_sysname();
                nanoprobe_report_upstream(FRAMESETTYPE_HBSHUTDOWN, NULL, sysname, 0);
                g_free(sysname); sysname = NULL;
                // Initiate connection shutdown.
                // This process will wait for all our output to be ACKed.
                // It also has an ACK timer, so it won't wait forever...
                proto->closeall(proto);
                idle_shutdown_gsource = g_idle_add(shutdown_when_outdone, NULL);
                nano_shutting_down = TRUE;
        }else{
                nano_shutting_down = TRUE;
                g_warning("%s: Never connected to CMA - cannot send shutdown message.", procname);
                ++errcount;  // Trigger non-zero exit code...
                _nano_final_shutdown(NULL);
                return TRUE;
        }
        return FALSE;
}
FSTATIC gboolean
shutdown_when_outdone(gpointer unused)
{
        ReliableUDP*    t = CASTTOCLASS(ReliableUDP, nanotransport->_netio);
        FsProtocol*     proto = CASTTOCLASS(FsProtocol, t->_protocol);
        (void)unused;
        // Wait for all our connections to be shut down
        if (proto->activeconncount(proto) == 0){
                DEBUGMSG("%s.%d: Shutting down - all connections closed."
                ,       __FUNCTION__, __LINE__);
                g_main_quit(mainloop);
                return FALSE;
        }
        return TRUE;
}
// Final Shutdown -- a contingency timer to make sure we eventually shut down
FSTATIC gboolean
_nano_final_shutdown(gpointer unused)
{
        (void)unused;
        DEBUGMSG("Initiating final shutdown");
        if (nano_connected && nanotransport->_netio->outputpending(nanotransport->_netio)){
                g_warning("Shutting down with unACKed output.");
                DUMP("Transport info", &nanotransport->_netio->baseclass, NULL);
        }
        if (idle_shutdown_gsource) {
                g_source_remove(idle_shutdown_gsource);
                idle_shutdown_gsource = 0;
        }
        g_main_quit(mainloop);
        return FALSE;
}