---

driver.c

Go to the documentation of this file.

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>

#include <stdlib.h>
#include <stdio.h>
#include <math.h>
#include <unistd.h>
#include <stdint.h>
#include <string.h>
#include <signal.h>
#include <time.h>
#include <pthread.h>
#include <getopt.h>

#include "nsds_util.h"
#include "flog.h"

/* Globals */
const int SRV_PORT = 42420;

const int QUEUE_LENGTH = 1;

const int NUM_THREADS = 2;

const size_t DATA_READ_SIZE = 256;

const time_t RATE_UPDATE_PERIOD = 10;

bool control_break;

bool streaming_active;

typedef struct 
{
    int daq_data;  
    int nsds_data; 
} data_thread_arg;


// ---------------------------------------------------------------------
void sighandler(int signal)
{
    char    me[] = "sighandler";

    flog_usr(FLOG_NOTICE, 0, me, "Got signal %d, exiting", signal);
    
    // New 10/17/02, let pipe errors propagate into socket layer
    if(signal == SIGPIPE)
    {
    flog_usr(FLOG_ERROR, FL_ERR_TRANSCEIVE, me,
         "Got SIGPIPE, ignoring");
    return;
    }
    
    // Signal must have been a break
    control_break = true;

    return;
}

// ---------------------------------------------------------------------
int tcp_send_driverids(const uint16_t control_socket, 
               const uint16_t data_socket, 
               const char *daq_name)
{
    char               me[] = "tcp_send_driverids";
    int                rc1, rc2;
    char               driver_uniq[128] = "";
    char               snd_len = 0;
    
    // Build unique driver id
    sprintf(driver_uniq, "driverid drv-lv-%s-ctr", daq_name);
    snd_len = strlen(driver_uniq);
    
    // Send driver id and so forth
    rc1 = tcp_nl_write(control_socket, driver_uniq);

    sprintf(driver_uniq, "driverid drv-lv-%s-data", daq_name);
    snd_len += strlen(driver_uniq);
    
    rc2 = tcp_nl_write(data_socket, driver_uniq);

    // Did it work?
    if((rc1 + rc2) == snd_len)
    return(0);
    else
    return(1);

    flog_usr(FLOG_NOLOG, 0, me, "Done");
}

// ---------------------------------------------------------------------
void *data_thread_main(void *arg)
{
    char              me[] = "data_thread_main";
    data_thread_arg   *s_args = arg;
    char              rcv_buf[MAX_CMD_LEN];
    int32_t           byte_count = 0;
    int               rc;
    data_thread_arg   l_args;
    int               rcv_count = 0;
    time_t            t_start = time(NULL);
    time_t            t_delta = 0;
    time_t            t_last = time(NULL);
    
    
    if(s_args == NULL)
    {
    flog_usr(FLOG_ERROR, FL_ERR_NULL_PTR, me,
         "Error - void thread argument pointer!");
    return(NULL);
    }

    // Save arguments, may be on the stack
    memcpy(&l_args, s_args, sizeof(l_args));

    // Blocking calls, so no delays are needed
    while((streaming_active) && (!control_break))
    {
    // Read a byte, write a byte.
    /* This is pretty inefficient - if we need to, we can
       use fdopen() and get line buffering from the OS.
       Test then decide.
    */
    rcv_count = read(l_args.daq_data, rcv_buf, DATA_READ_SIZE);
    if(rcv_count <= 0)
    {
        flog_usr(FLOG_ERROR, FL_ERR_TRANSCEIVE, me,
             "Error reading data from DAQ");
        goto thread_bail;
    }

    // Write data out to NSDS
    rc = write(l_args.nsds_data, rcv_buf, rcv_count);
    if(rc != rcv_count)
    {
        flog_usr(FLOG_ERROR, FL_ERR_TRANSCEIVE, me,
             "Error sending data to NSDS");
        goto thread_bail;
    }
    
    byte_count += rcv_count;

    // DDT Statistics
    t_delta = time(NULL) - t_last;
    if(t_delta >= RATE_UPDATE_PERIOD)
    {
        t_last = time(NULL);
        flog_usr(FLOG_NOTICE, 0, me, 
             "Average data rate so far: %f for %d kb",
             ((float) (byte_count >> 10) / (float) (time(NULL) - t_start)),
             (byte_count >> 10));
    }
    }

  thread_bail:
    // See if can appx the rate
    t_delta = time(NULL) - t_start;

    // If less than one second, accept the error in return for not /0
    if(t_delta == 0)
    t_delta = 1;
    
    flog_usr(FLOG_NOTICE, 0, me, 
         "Thread exiting, %d bytes transferred, appx %f kb/sec", 
         byte_count, ((float) (byte_count >> 10) / (float) t_delta));

    // Tell main we're offline
    streaming_active = false;
    
    return(NULL);
}
    
// ---------------------------------------------------------------------
int data_thread_startup(data_thread_arg d_arg)
{
    char        me[] = "data_thread_startup";
    int         rc;
    pthread_t   data_thread;

    // Just in case, this should restart thread
    streaming_active = false;
    sleep(1);
    
    streaming_active = true;
        
    rc = pthread_create(&data_thread, NULL, data_thread_main,
            (void *) &d_arg);
    if(rc != 0)
    {
    flog_usr(FLOG_ERROR, FL_ERR_SYSTEM, me,
         "Error on thread creation");
    return(rc);
    }
    
    // Let 'er rip
    pthread_detach(data_thread);
    pthread_setconcurrency(NUM_THREADS);

    // Give it time to start
    usleep(1000);
    
    flog_usr(FLOG_NOTICE, 0, me, "Data thread started OK");

    return(0);
}

// ---------------------------------------------------------------------
int cmd_handler(const uint16_t nsds_s_ctrl, 
        const uint16_t nsds_s_data,
        const uint16_t daq_control,
        const uint16_t daq_data, 
        const char     *daq_name)
{
    char               me[] = "cmd_handler";
    char               cmd_buf[MAX_CMD_LEN] = "";
    int                bytes_read = 0;
    int                rc;
    const time_t       normal_timeout = 30;
    bool               got_cmd = false;

    /* Loop until:
       1) Get a command from the NSDS
       2) Get an error on control or data sockets
       3) User aborts
    */
    while(!got_cmd)
    {
       flog_usr(FLOG_PROGRESS, 0, me, "Waiting for command...");
        
       // Get command from NSDS control channel, wait 30 secs
       bytes_read = tcp_nl_read(nsds_s_ctrl, cmd_buf, normal_timeout);
       if(bytes_read < 0)
       {
       flog_usr(FLOG_WARNING, FL_ERR_TRANSCEIVE, me,
            "Error reading command from control channel");
       return(1);
       }
       else if(bytes_read == 0)
       {
       flog_usr(FLOG_L3BUG, 0, me,
            "No command before timeout, checking status");
       // Check for dead streaming thread
       if(streaming_active == false)
       {
           flog_usr(FLOG_WARNING, FL_ERR_BAD_VALUE, me,
                "Data thread has died! Remaking connections.");
           return(2);
       }
       }
       else // Got command!
       {
       flog_usr(FLOG_PROGRESS, 0, me, "Command received OK from NSDS");
       got_cmd = true;
       }
    }

    // Here we intercept the welcome command and return our compiled-in
    // string (plus cmd-line modifier if present). For this we have 1KLOC of C.
    if(strstr(cmd_buf, "Welcome") != NULL)
    {
    flog_usr(FLOG_L2BUG, 0, me,
         "Got welcome string '%s'", cmd_buf);
    flog_usr(FLOG_CL2BUG, 0, me,
         "Sending driver ident to control and data channels...");

    rc = tcp_send_driverids(nsds_s_ctrl, nsds_s_data, daq_name);
    if(rc == 0)
    {
        flog_usr(FLOG_PROGRESS, 0, me,
             "Driver IDs sent OK, command complete");
    }
    else
    {
        flog_usr(FLOG_ERROR, FL_ERR_TRANSCEIVE, me,
             "Error sending driver IDs");
    }
    
    return(rc);
    }

    // If it ain't 'Welcome', we forward it as-is, ditto for the reply. We R Netcat.

    flog_usr(FLOG_L2BUG, 0, me, 
         "Got request '%s', forwarding to DAQ", cmd_buf);
    
    rc = tcp_nl_write(daq_control, cmd_buf);
    if(rc != strlen(cmd_buf))
    {
    flog_usr(FLOG_ERROR, FL_ERR_TRANSCEIVE, me,
         "Error sending command to DAQ");
    return(1);
    }
    
    flog_usr(FLOG_L2BUG, 0, me,"Reading reply from DAQ");
    
    // Get reply from DAQ
    bytes_read = tcp_nl_read(daq_control, cmd_buf, normal_timeout);
    if(bytes_read <= 0)
    {
    flog_usr(FLOG_ERROR, FL_ERR_TRANSCEIVE, me,
         "Error reading DAQ reply");
    return(1);
    }
    
    // DDT 
    flog_usr(FLOG_L2BUG, 0, me, 
         "Got reply '%s', forwarding to NSDS", cmd_buf);
    
    // Forward to NSDS
    rc = tcp_nl_write(nsds_s_ctrl, cmd_buf);
    if(rc != bytes_read)
    {
    flog_usr(FLOG_ERROR, FL_ERR_TRANSCEIVE, me,
         "Error writing to NSDS control channel");
    return(1);
    }
    
    flog_usr(FLOG_L2BUG, 0, me,"Query completed OK");
    flog_usr(FLOG_CL2BUG, 0, me, "Done with cmd_handler");
    
    return(0);
}

// ---------------------------------------------------------------------
void main_loop(const char *daq_machine, const uint16_t daq_port,
           const char *nsds_machine, const uint16_t nsds_port,
           const char *driver_name)
{
    char               me[] = "main_loop";
    int                nsds_s_data, nsds_s_ctrl, daq_s_ctrl, daq_s_data;
    int                init_count = 0;
    time_t             wait_time = 10;
    time_t             end_interval = 3 * wait_time;
    int                rc = 0;
    static data_thread_arg d_arg;
    
    
    // ------------------------
    // Main loop, runs until control-c
    while(control_break == false)
    {
    // If restarting, close all first
    if(init_count > 0)
    {
        flog_usr(FLOG_NOTICE, 0, me,
             "Re-init driver, closing connections");

        // Tell data thread to exit, if running
        streaming_active = false;

        // Shut down all TCP connections
        tcp_close(daq_s_ctrl);
        tcp_close(daq_s_data);
        
        tcp_close(nsds_s_ctrl);
        tcp_close(nsds_s_data);

        // Mark as invalid FDs
        daq_s_ctrl = daq_s_data = nsds_s_ctrl = nsds_s_data = -1;

        // Let labview finish. Do NOT REMOVE THIS. It's a hack to workaround
        // a really stupid LV bug. You Have Been Warned.
        sleep(2);
    }

    init_count++;

    flog_usr(FLOG_NOTICE, 0, me,
         "Beginning connect attempt #%d, DAQ first", init_count);
    
    // Connect to DAQ control channel, wait forever for same
    daq_s_ctrl = tcp_connect_retry(daq_machine, daq_port, 
                       wait_time, (time_t) 0,
                       "DAQ control port");
    // If error, start over
    if(daq_s_ctrl <= 0)
        continue;
    
    // Did they hit control-c?
    if(control_break == true)
        break;

    flog_usr(FLOG_CNOTICE, 0, me,
         "Connected to DAQ machine '%s' control channel", 
         tcp_peername(daq_s_ctrl));

    // DDT labview likes a bit of a delay between TCP connects
    sleep(1);

    // Connect to DAQ data channel, wait usual time
    daq_s_data = tcp_connect_retry(daq_machine, daq_port + 1, 
                       wait_time, (time(NULL) + end_interval),
                       "DAQ data port");
    // If error, start over
    if(daq_s_data <= 0)
        continue;
    
    // Did they hit control-c?
    if(control_break == true)
        break;

    flog_usr(FLOG_CNOTICE, 0, me,
         "Connected to DAQ machine '%s' data channel", 
         tcp_peername(daq_s_ctrl));

    // Connect to NSDS control, end at now + 30 secs
    nsds_s_ctrl = tcp_connect_retry(nsds_machine, nsds_port,
                    wait_time, 
                    (time(NULL) + end_interval),
                    "NSDS control port");
    if(nsds_s_ctrl <= 0)
    {
        flog_usr(FLOG_NOTICE, 0, me, "No NSDS control connection, restarting");
        continue;
    }
    if(control_break == true)
        break;

    flog_usr(FLOG_CNOTICE, 0, me,
         "Connected to NSDS control channel on %s",
         tcp_peername(nsds_s_ctrl));

    // ---------------
    // Second TCP connection to NSDS - data channel, wait 30 secs
    //  note use of adjacent TCP port for connection!
    nsds_s_data = tcp_connect_retry(nsds_machine, (nsds_port + 1),
                    wait_time, 
                    (time(NULL) + end_interval),
                    "NSDS data port");
    if(nsds_s_data <= 0)
    {
        flog_usr(FLOG_NOTICE, 0, me, 
             "No NSDS data connection, restarting");
        continue;
    }
    if(control_break == true)
        break;

    flog_usr(FLOG_CNOTICE, 0, me,
         "Connected to NSDS data channel on %s", 
         tcp_peername(nsds_s_data));

    flog_usr(FLOG_PROGRESS, 0, me,
         "All network connections in place, proceeding");
    
    // New 10/21/02, start data thread now and let it run
    flog_usr(FLOG_PROGRESS, 0, me, "Starting data thread");
    
    d_arg.nsds_data = nsds_s_data;
    d_arg.daq_data = daq_s_data;
    
    rc = data_thread_startup(d_arg);
    if(rc != 0)
    {
        flog_usr(FLOG_ERROR, FL_ERR_SYSTEM, me,
             "Error on data thread creation!");
        return;
    }
    
    // Do work (read commands and respond) until error
    do
    {
        rc = cmd_handler(nsds_s_ctrl, nsds_s_data, daq_s_ctrl, 
                 daq_s_data, driver_name);
    } while(rc == 0);
    }
    
    return;
}
    
// ---------------------------------------------------------------------
int main(int argc, char *argv[])
{
    char          me[] = "main";
    char          daq_machine[512] = "";
    char          nsds_machine[512] = "";
    const char    daq_predef[] = "localhost";
    const int     daq_preport = 55055;
    const char    nsds_predef[] = "localhost";
    const int     nsds_preport = 42420;
    char          driver_name[512] = "";
    int           rc, idx;
    int           help_flag = 0;
    uint16_t      daq_port = daq_preport;
    uint16_t      nsds_port = nsds_preport;
    struct option long_options[] = 
    {
        {"daq_port", required_argument, 0, 'd'},
        {"daq_machine", required_argument, 0, 'm'},
        {"nsds_port", required_argument, 0, 'n'},
        {"nsds_machine", required_argument, 0, 'l'},
        {"help", no_argument, &help_flag, 'h'},
        {"driver_name", required_argument, 0, 'x'},
        {0, 0, 0, 0}
    };
        

    // Need to set logging options before any printouts
    flog_set_report(FLOG_L3BUG, FLOG_QUIET, FLOG_QUIET);
    flog_set_style(0x2019, 0x0038, 0x6008);    

    // Set defaults
    strcpy(nsds_machine, nsds_predef);
    strcpy(daq_machine, daq_predef);
    
    daq_port = daq_preport;
    nsds_port = nsds_preport;
    

    // Check command line
    while(1)
    {
    rc = getopt_long(argc, argv, "d:m:n:l:x:h",
             long_options, &idx);
    if(rc == -1)
        break;
    
    switch(rc)
    {
      case 0: // Flag set, just bail
        break;
        
      case 'd': // daq_port
        daq_port = atoi(optarg);
        flog_usr(FLOG_NOTICE, 0, me, 
             "Setting DAQ port to %d", daq_port);
        break;

      case 'm': // Daq machine name
        strcpy(daq_machine, optarg);
        flog_usr(FLOG_NOTICE, 0, me, 
             "Set DAQ machine name to '%s'", daq_machine);
        break;
        
      case 'n': // NSDS port
        nsds_port = atoi(optarg);
        flog_usr(FLOG_NOTICE, 0, me, 
             "Set NSDS port to %d", nsds_port);
        break;
        
      case 'l': // NSDS machine name
        strcpy(nsds_machine, optarg);
        flog_usr(FLOG_NOTICE, 0, me,
             "Set NSDS machine name to '%s'", nsds_machine);
        break;
        
      case 'x': // Driver name
        strcpy(driver_name, optarg);
        flog_usr(FLOG_NOTICE, 0, me, 
             "Set driver name to '%s'", driver_name);
        break;
        
      default:
      case 'h': // Help!
        help_flag = 1;
        break;
    }
    }
    
    if(help_flag)
    {
    flog_usr(FLOG_NOTICE, 0, me, 
         "Options are --daq_port, --daq_machine,");
    flog_usr(FLOG_CNOTICE, 0, me, 
         "--nsds_port, --nsds_machine, --driver_name");
    flog_usr(FLOG_CNOTICE, 0, me,
         "All have arguments, either  hostname or TCP port number");
    flog_usr(FLOG_CNOTICE, 0, me,
         "driver_name is for multiple instances per host");
    exit(1);
    }
    
    flog_usr(FLOG_NOTICE, 0, me, "Installing signal handler.");

    control_break = false;
    signal(SIGINT, sighandler);
    signal(SIGPIPE, sighandler);

    // Set driver name for NSDS if necessary
    if(driver_name[0] == 0x00)
    strcpy(driver_name, daq_machine);

    // Call main routine
    main_loop(daq_machine, daq_port, nsds_machine, nsds_port, driver_name);
    
    flog_usr(FLOG_NOTICE, 0, me, "Done OK");
    
    return(0);
}

---