acq2xx_server.cpp

Go to the documentation of this file.

/* ------------------------------------------------------------------------- */
/* file acq2xx_server.cpp                                                    */
/* ------------------------------------------------------------------------- */
/*   Copyright (C) 2006 Peter Milne, D-TACQ Solutions Ltd
 *                      <Peter dot Milne at D hyphen TACQ dot com>

    This program is free software; you can redistribute it and/or modify
    it under the terms of Version 2 of the GNU General Public License
    as published by the Free Software Foundation;

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.                */
/* ------------------------------------------------------------------------- */

/** @file acq2xx_server.cpp server WS implementation.
 *
 * usage: acq2xx_server [port]
 *
 * Server is multi-threaded, and stays resident by default
 */
#include "soapH.h"
#include "acq2xx.nsmap"

#ifdef __arm
#include "local.h"
#else
#include <stdio.h>
#endif

#ifndef __arm
#define info    printf
#endif
#define MAXBUF 16384

using namespace std;

#include <assert.h>
#include <iostream>
#include <iomanip>
#include <string>

#define ACQCMD "/usr/local/bin/acqcmd"

#define NOT_FOR_SERVER  0

static int verbose = 0;
static int logcommands = 0;

#include <string.h>


class Actor {
public:
        virtual ~Actor() {}
        virtual int spawn_read(char* lcommand, char* buf, int maxbuf);
};

class ActorStub : public Actor {
public:
        virtual int spawn_read(char* lcommand, char* buf, int maxbuf);  
};


static Actor* actor = new Actor();


#include "soapH.h"
#include <pthread.h>
#define BACKLOG (100) // Max. request backlog
#define MAX_THR (10) // Max. threads to serve requests

static int cgi_main(soap* soap);
static int standalone_main(soap* soap, int port);

int main(int argc, char **argv)
{
        struct soap soap;
        soap_init(&soap);

#ifdef __arm
        info("acq2xx_server $Revision: 1.23 $ " ACQ2XX_H_VERSION);
#endif
        if (getenv("ACQ2XX_SERVER_STUB")){
                actor = new ActorStub();               
        }
        if (getenv("ACQ2XX_LOG")){
                logcommands = atoi(getenv("ACQ2XX_LOG"));
                info("ACQ2XX_LOG set logcommands %d", logcommands);
        }
        if (getenv("ACQ2XX_WS_VERBOSE")){
                verbose = atoi(getenv("ACQ2XX_WS_VERBOSE"));
                info("ACQ2XX_WS_VERBOSE set verbose %d", verbose);
        }

        
        if (argc < 2){
                return cgi_main(&soap);
        }else{
                return standalone_main(&soap, atoi(argv[1]));
        }
}


static int cgi_main(soap* soap)
{
        soap_serve(soap); // serve request, one thread, CGI style
        soap_destroy(soap); // dealloc C++ data
        soap_end(soap); // dealloc data and clean up
        return 0;
}

static void print_signon(soap* soap, int i, int s)
{
#ifdef __arm
        info(
#else
        fprintf(stderr, 
#endif
                "Thread %d accepts socket %d connection from IP %d.%d.%d.%d\n", 
                i, s, (soap->ip >> 24)&0xFF, (soap->ip >> 16)&0xFF, 
                (soap->ip >> 8)&0xFF, soap->ip&0xFF);
}


static int standalone_main(soap* soap, int port)
{
        // each thread needs a runtime environment
        struct soap *soap_thr[MAX_THR]; 
        pthread_t tid[MAX_THR];
        SOAP_SOCKET m, s;
        int i;

        m = soap_bind(soap, NULL, port, BACKLOG);
        if (!soap_valid_socket(m)){
                exit(1);
        }
        fprintf(stderr, "Socket connection successful %d\n", m);
        for (i = 0; i < MAX_THR; i++){
                soap_thr[i] = NULL;
        }
        for (;;){
                for (i = 0; i < MAX_THR; i++){
                        s = soap_accept(soap);
                        if (!soap_valid_socket(s)){
                                if (soap->errnum){
                                        soap_print_fault(soap, stderr);
                                        continue; // retry
                                }else{
#ifdef __arm
                                        err(
#else
                                        fprintf(stderr, 
#endif
                                                        "Server timed out\n");
                                        break;
                                }
                        }
                        print_signon(soap, i, s);

                        if (!soap_thr[i]){ // first time around
                                soap_thr[i] = soap_copy(soap);
                                if (!soap_thr[i]){
                                        exit(1); // could not allocate
                                }
                        }else{  // recycle soap environment
                                pthread_join(tid[i], NULL);
#ifdef __arm
                                info(
#else
                                fprintf(stderr, 
#endif       
                                                "Thread %d completed\n", i);
                                // deallocate C++ data of old thread
                                soap_destroy(soap_thr[i]); 
                                // deallocate data of old thread
                                soap_end(soap_thr[i]); 
                        }
                        soap_thr[i]->socket = s; // new socket fd
                        pthread_create(
                                &tid[i], NULL, (void*(*)(void*))soap_serve, 
                                (void*)soap_thr[i]);
                }
        }
        for (i = 0; i < MAX_THR; i++){
                if (soap_thr[i]){
                        soap_done(soap_thr[i]); // detach context
                        free(soap_thr[i]); // free up
                }
        }

        return 0;
}


int Actor::spawn_read(char* lcommand, char* buf, int maxbuf)
{
        if (logcommands){
                info("cmd:%s", lcommand);
        }
        FILE* fp = popen(lcommand, "r");

        buf[0] = '\0';

        for (char* bp = buf; bp - buf < MAXBUF; ){
                char *nl = fgets(bp, 256, fp);
                if (nl){
                        bp += strlen(nl);
                }else{
                        break;
                }
        }
        int rc = pclose(fp);    

        if (logcommands > 1){
                info("[%d] %s", rc, buf);
        }

        return rc;
}

int ActorStub::spawn_read(char* lcommand, char* buf, int maxbuf)
{
        char *stubcommand = new char[strlen(lcommand)+10];
        
        cout << lcommand << endl;

        sprintf(stubcommand, "echo %s", lcommand);
        return Actor::spawn_read(stubcommand, buf, maxbuf);
}


static int soap_command(
        struct soap* soap, char* lcommand, char* command, char*& response)
{
        char* buf = (char*)soap_malloc(soap, MAXBUF);
        
        sprintf(buf, "%s \'%s\'", lcommand, command);
        
        actor->spawn_read(buf, buf, MAXBUF);

        response = buf;
        return SOAP_OK;
}


int acq2xx__acqcmd(struct soap *soap, char* command, char*& response)
{
        return soap_command(soap, ACQCMD, command, response);
}

int acq2xx__acq2sh(struct soap *soap, char* command, char*& response)
{
        return soap_command(soap, "sh -c", command, response);
}

int acq2xx__getVersion(
        struct soap *soap, 
        char* client_version, 
        char*& server_version)
{
        char* buf = (char*)soap_malloc(soap, strlen(ACQ2XX_H_VERSION)+1);

        strcpy(buf, ACQ2XX_H_VERSION);
        server_version = buf;

        if (strcmp(client_version, server_version) != 0){
                info("WARNING: version mismatch client: %s server %s",
                     client_version, server_version);
        }

        return SOAP_OK;
}

static int acqcmd(const char* command, char* buf, int maxbuf)
{
        char lcommand[128];
        
        sprintf(lcommand, "%s \'%s\'", ACQCMD, command);
        actor->spawn_read(lcommand, buf, maxbuf);
        return 0;
}

int acq2xx__getAcqStatus(struct soap *soap, struct acq2xx__AcqStatus& status)
{
        char buf[256];
        acq2xx__AcqStatus lbuf;
        char* state_string = (char*)soap_malloc(soap, 32);
        lbuf.nchan = 5;
        lbuf.state_string = "Hello World";

        acqcmd("getNumSamples", buf, 256);
        sscanf(buf, "ACQ32:getNumSamples=%d pre=%d post=%d elapsed=%d",
               &lbuf.numsamples, &lbuf.prelen, 
               &lbuf.postlen, &lbuf.elapsed);
        
        acqcmd("getState", buf, 256);
        sscanf(buf, "ACQ32:%d %s", &lbuf.state, state_string);
        lbuf.state_string = state_string;

        acqcmd("getNumChannels", buf, 256);
        sscanf(buf, "ACQ32:getNumChannels=%d", &lbuf.nchan);
        status = lbuf;


        return SOAP_OK;
}


int acq2xx__setArm(
        struct soap* soap, 
        int block, 
        struct acq2xx__AcqStatus& status)
{
        char buf[256];

        acqcmd("setArm", buf, 256);

        if (block){
                acq2xx__AcqStatus lbuf;
                int scanned;

                do{
                        acqcmd("getState", buf, 256);
                        scanned = sscanf(buf, "ACQ32:%d", &lbuf.state);
                } while(scanned == 1 && lbuf.state != ST_STOP);
                
        }

        return acq2xx__getAcqStatus(soap, status);
}


int acq2xx__waitStop(
        struct soap* soap, 
        int timeout_sec, 
        struct acq2xx__AcqStatus& status)
{
        if (timeout_sec < 1) timeout_sec = 1;
        if (timeout_sec > 10) timeout_sec = 10;

        char command[80];
        char buf[80];   
        buf[0] = '\0';
        
        sprintf(command, "acqcmd -t %d --until ST_STOP", timeout_sec);

        actor->spawn_read(command, buf, 80);
        return acq2xx__getAcqStatus(soap, status);                      
}


static const char* toString(enum acq2xx__DIx dix)
{
        switch(dix){
        case DI0:       
                return "DI0";
        case DI1:
                return "DI1";
        case DI2:
                return "DI2";
        case DI3:
                return "DI3";
        case DI4:
                return "DI4";
        case DI5:
                return "DI5";
        case DI_NONE:
        default:
                return "none";
        }
}

static const char* toString(enum acq2xx__DOx dox)
{
        switch(dox){
        case DO0:       
                return "DO0";
        case DO1:
                return "DO1";
        case DO2:
                return "DO2";
        case DO3:
                return "DO3";
        case DO4:
                return "DO4";
        case DO5:
                return "DO5";
        case DO_NONE:
        default:
                return "none";
        }
}

static const char* toString(enum acq2xx__EDGE edge){
        return edge==EDGE_FALLING? "falling": "rising";
}



int acq2xx__acq2script(
        struct soap* soap,
        acq2xx__Transaction *in, 
        struct acq2xx__acq2scriptResponse &result)
{
        if (in == 0){
                cerr << "ERROR: null transaction" << endl;
                return -1;
        }

        in->exec(soap);
        result.out = in;
        if (verbose){
                cout << "acq2xx__acq2script" << " output: "; result.out->print();
        }
        return SOAP_OK;
}



void acq2xx__Transaction::exec(soap *soap)
{
        status = run(soap);
        complete = 1;

        if (verbose > 1){
                cout << "exec() done complete: " << complete 
                     << "status: " <<status ; print();
        }
}

int acq2xx__Transaction::run(soap *soap)
{
        cout << "Transaction::run";  print();
        return 37;
}




int acq2xx__TransactionList::run(soap *soap)
{
        for (int i = 0; i < __size; i++){
                acq2xx__Transaction* t = __ptr[i];

                t->exec(soap);
                if (t->status != 0 && t->abort_on_error){
                        return t->status;
                }
        }

        return 0;
}




int acq2xx__ShellTransaction::run(soap *soap)
{

        char* buf = (char*)malloc(MAXBUF);      
        buf[0] = '\0';
        int rc = actor->spawn_read(command, buf, MAXBUF);
        
        char *_result = (char*)soap_malloc(soap, strlen(buf)+1);
        strcpy(_result, buf);
        result = _result;

        if (verbose > 1){
                cout << "acq2xx__ShellTransaction::run() command:" << command << endl <<
                        "result:" << result << " rc:" << rc << endl;
        }

        free(buf);
        return rc;      
}


acq2xx__ModeSetTransaction::acq2xx__ModeSetTransaction(
        enum acq2xx__MODE mode,
        int prelen,
        int postlen             
        )
{
        assert(NOT_FOR_SERVER);
}

int acq2xx__ModeSetTransaction::run(struct soap* soap)
{
        char _command[80];

        cout << 
        "acq2xx__ModeSetTransaction::run() "<< endl <<
                "mode:" << mode << " pre:" << prelen << 
                " post:" <<  postlen << endl;   

        switch(mode){
        case MODE_SOFT_TRANSIENT:
        case MODE_SOFT_CONTINUOUS:
                sprintf(_command, "acqcmd setMode %s %d",
                        mode==MODE_SOFT_TRANSIENT?
                        "SOFT_TRANSIENT": "SOFT_CONTINUOUS",
                        mode==MODE_SOFT_TRANSIENT? postlen: prelen);
                break;
        case MODE_GATED_TRANSIENT:
                sprintf(_command, "acqcmd setMode GATED_TRANSIENT %d",
                        postlen);
                break;
        case MODE_TRIGGERED_CONTINUOUS:
                sprintf(_command, "acqcmd setModeTriggeredContinuous %d %d",
                        prelen, postlen);                       
                break;
        default:
                assert(0);
        }


        initCommand(_command);

        if (verbose > 1){
                cout << "acq2xx__ModeSetTransaction::run() "<<
                        _command << endl <<
                        "command:" << command << endl;
        }

        return acq2xx__ShellTransaction::run(soap);
}


acq2xx__ClockSetTransaction::acq2xx__ClockSetTransaction(char *command) :
        acq2xx__ShellTransaction(command)
{
        assert(NOT_FOR_SERVER);
}

acq2xx__InternalClockSetTransaction::acq2xx__InternalClockSetTransaction(
        int _hz,
        int div,
        enum acq2xx__DOx dox)
{
        assert(NOT_FOR_SERVER);
}

int acq2xx__InternalClockSetTransaction::run(struct soap* soap)
{
        char _command[80];
        int cursor = sprintf(_command, "acqcmd setInternalClock %d", hz);
        if (div > 1 || dox != DO_NONE){
                cursor += sprintf(_command+cursor, " %d", div);

                if (dox != DO_NONE){
                        cursor += sprintf(_command+cursor, " %s", 
                                                        toString(dox));
                }
        }
        initCommand(_command);
        return acq2xx__ClockSetTransaction::run(soap);
}

acq2xx__ExternalClockSetTransaction::acq2xx__ExternalClockSetTransaction(
        enum acq2xx__DIx dix,
        int div,
        enum acq2xx__DOx dox
        )
{
        assert(NOT_FOR_SERVER);
}

int acq2xx__ExternalClockSetTransaction::run(struct soap* soap)
{
        char _command[80];
        
        int cursor = sprintf(_command, "acqcmd setExternalClock %s", 
                                toString(dix));
        if (div > 1 || dox != DO_NONE){
                cursor += sprintf(_command+cursor, " %d", div);

                if (dox != DO_NONE){
                        cursor += sprintf(_command+cursor, " %s", 
                                                        toString(dox));
                }
        }
        initCommand(_command);
        return acq2xx__ClockSetTransaction::run(soap);
}

acq2xx__SignalSetTransaction::acq2xx__SignalSetTransaction(
        char *name,
        enum acq2xx__DIx dix,
        enum acq2xx__EDGE edge)
{
        assert(NOT_FOR_SERVER);
}

int acq2xx__SignalSetTransaction::run(soap* soap)
{
        char command[80];
        sprintf(command, "set.%s %s %s", signal, 
                        toString(dix), dix != DI_NONE? toString(edge): "");
        initCommand(command);
        return acq2xx__ShellTransaction::run(soap);
}

acq2xx__AcqControlTransaction::acq2xx__AcqControlTransaction(
        acq2xx__ModeSetTransaction* _modeSet,
        acq2xx__ClockSetTransaction* _clockSet,
        acq2xx__SignalSetTransaction* _trigger,
        acq2xx__SignalSetTransaction* _event0
        )
{
        assert(NOT_FOR_SERVER);
}

acq2xx__AcqControlTransaction::acq2xx__AcqControlTransaction()
{
}

int acq2xx__GetAvailableChannelsTransaction::run(soap* soap)
{
        int rc = acq2xx__ShellTransaction::run(soap);
        if (rc == 0){
                return sscanf(result, 
                                "ACQ32:getAvailableChannels AI=%d AO=%d",
                              &AI, &AO) != 2;
        }else{
                return rc;
        }
}


static int parseVR(VRange* ranges, int channels, char* result)
{
        char *rbuf = (char*)malloc(strlen(result)+1);

        strcpy(rbuf, result);
        char *str = rbuf;
        char *tok;

        for (int ic = 0; ic < channels; ++ic){
                if ((tok = strtok(str, ","))){
                        ranges[ic].min = atof(tok);
                }else{
                        return -1;
                }
                str = 0;
                if ((tok = strtok(str, ","))){
                        ranges[ic].max = atof(tok);
                }else{
                        return -1;
                }               
        }
        free(rbuf);
        return 0;
}
int acq2xx__GetVRangeTransaction::run(soap* soap)
{
        channels = soap_new_acq2xx__GetAvailableChannelsTransaction(soap, -1);

        int rc = channels->run(soap);
        if (rc == 0){
                rc = acq2xx__ShellTransaction::run(soap);
                if (rc == 0){
                        ranges.__ptr = soap_new_VRange(soap, channels->AI);
                        ranges.__size = channels->AI;
                        return parseVR(ranges.__ptr, channels->AI, result);
                }
        }

        return rc;
}

---