agent_tunnel.cpp File Reference

#include "support/configCosmos.h"
#include <stdio.h>
#include "agent/agentclass.h"
#include "device/serial/serialclass.h"

Include dependency graph for agent_tunnel.cpp:

Macros
#define	MAXBUFFERSIZE   2560
#define	TUN_BUF_SIZE   2000

Functions
void	tcv_read_loop ()
void	tcv_write_loop ()
void	tun_read_loop ()
void	tun_write_loop ()
int	main (int argc, char *argv[])

Variables
static Agent *	agent
static std::queue< vector< uint8_t > >	tun_fifo
static std::queue< vector< uint8_t > >	tcv_fifo
static std::condition_variable	tcv_fifo_check
static std::condition_variable	tun_fifo_check
static int	tun_fd
static string	rxr_devname
static string	txr_devname
static Serial *	rxr_serial =nullptr
static Serial *	txr_serial =nullptr
static uint32_t	rxr_baud = 9600
static uint32_t	txr_baud = 9600

Macro Definition Documentation

#define MAXBUFFERSIZE   2560

#define TUN_BUF_SIZE   2000

Function Documentation

void tcv_read_loop

(

)

{
    std::vector<uint8_t> buffer;
    int32_t iretn;
    double lastin = currentmjd(0.);
    double lastbeacon = currentmjd(0.);

    while (agent->running())
    {
        // Read data from receiver port
        iretn = kpc9612p_recvframe(&rxr_handle);
        if (iretn > 0)
        { // Start of mutex for tun FIFO
            kpc9612p_unloadframe(&rxr_handle, buffer);
            //          uint16_t cs1 = kpc9612p_calc_fcs(&buffer[17], buffer.size()-17);
            //          uint16_t cs2 = *(uint16_t*)(astroin->payload+astroin->header.size-4);
            if (buffer[SOCKET_IP_BYTE_VERSION]>>4 == 4 && buffer.size() == buffer[SOCKET_IP_BYTE_LEN_HIGH]*256U+buffer[SOCKET_IP_BYTE_LEN_LOW])
            {
                if (buffer[SOCKET_IP_BYTE_PROTOCOL] != SOCKET_IP_PROTOCOL_UDP || (buffer[SOCKET_IP_BYTE_PROTOCOL] == SOCKET_IP_PROTOCOL_UDP && !socket_check_udp_checksum(buffer)))
                {
                    tun_fifo.push(buffer);
                    tun_fifo_check.notify_one();
                    printf("Buffer: [%u,%" PRIu64 "] %f\n", rxr_handle.frame.size, buffer.size(), 86400.*(currentmjd(0.)-lastin));
                    lastin = currentmjd(0.);
                    //                  for (uint16_t i=0; i<(rxr_handle.frame.size<150?rxr_handle.frame.size:150); ++i)
                    //                  {
                    //                      if (i == 17)
                    //                      {
                    //                          printf("[ ");
                    //                      }
                    //                      printf("%x ", rxr_handle.frame.full[i]);
                    //                      if (i == buffer.size()+16 || i == 149)
                    //                      {
                    //                          printf("] ");
                    //                      }
                    //                  }
                    //                  printf("\n");
                    fflush(stdout);
                }
                else
                {
                    printf("UDP checksum error:\n");
                }
            }
            else
            {
                printf("Beacon: [%d,%u,%" PRIu64 "] %f\n", iretn, rxr_handle.frame.size, buffer.size(), 86400.*(currentmjd(0.)-lastbeacon));
                lastbeacon = currentmjd(0.);
                //              std::string str(buffer.begin(), buffer.end());
                //              std::cout << "\t" << str << std::endl;
            }
        } // End of mutex for tun FIFO
    }
    tun_fifo_check.notify_all();

    kpc9612p_disconnect(&rxr_handle);
}

void tcv_write_loop

(

)

{
    std::vector<uint8_t> buffer;
    //  int32_t iretn;
    //  double lastout = currentmjd(0.);

    while (agent->running())
    {
        {   // Start of mutex for tcv FIFO
            std::unique_lock<std::mutex> locker(tcv_fifo_lock);

            while (tcv_fifo.empty())
            {
                tcv_fifo_check.wait(locker);
            }
        }   // End of mutex for tcv FIFO

        while (!tcv_fifo.empty())
        {
            // Get next packet from transceiver FIFO
            buffer = tcv_fifo.front();
            tcv_fifo.pop();
            // Write data to transmitter port
            kpc9612p_loadframe(&txr_handle, buffer);
            kpc9612p_sendframe(&txr_handle);
            //          printf("Out Radio: [%d,%u,%u] %f\n", iretn, txr_handle.frame.size, buffer.size(), 86400.*(currentmjd(0.)-lastout));
            //          lastout = currentmjd(0.);
            //          if (iretn >= 0)
            //          {
            //              for (uint16_t i=0; i<(txr_handle.frame.size<150?txr_handle.frame.size:150); ++i)
            //              {
            //                  if (i == 17)
            //                  {
            //                      printf("[ ");
            //                  }
            //                  if (i == buffer.size()+17)
            //                  {
            //                      printf("] ");
            //                  }
            //                  printf("%x ", txr_handle.frame.full[i]);
            //              }
            //              printf("\n");
            //          }
        }
    }

    kpc9612p_disconnect(&txr_handle);
}

void tun_read_loop

(

)

{
    std::vector<uint8_t> buffer;
    int32_t nbytes;

    while (agent->running())
    {

        buffer.resize(TUN_BUF_SIZE);
        nbytes = read(tun_fd, &buffer[0], TUN_BUF_SIZE);
        if (nbytes > 0)
        {
            buffer.resize(nbytes);

            // Add UDP checksum if necessary
            if (buffer[SOCKET_IP_BYTE_PROTOCOL] == SOCKET_IP_PROTOCOL_UDP)
            {
                socket_set_udp_checksum(buffer);
            }

            tcv_fifo.push(buffer);
            tcv_fifo_check.notify_one();
        }
    }
    tcv_fifo_check.notify_all();
}

void tun_write_loop

(

)

{
    std::vector<uint8_t> buffer;
    int32_t nbytes;

    while (agent->running())
    {
        {   // Start of mutex for tun fifo
            std::unique_lock<std::mutex> locker(tun_fifo_lock);

            while (tun_fifo.empty())
            {
                tun_fifo_check.wait(locker);
            }   // End of mutex for tun fifo
        }

        while (!tun_fifo.empty())
        {
            buffer = tun_fifo.front();
            nbytes = agent->post(Agent::AgentMessage::COMM, buffer);
            if (open_tunnel)
            {
                nbytes = write(tun_fd, &buffer[0], buffer.size());
            }
            if (nbytes > 0)
            {
                //              printf("Out TUN: [%u,%u] ", nbytes, buffer.size());
                //              for (uint16_t i=0; i<(buffer.size()<150?buffer.size():150); ++i)
                //              {
                //                  if (i == 0)
                //                  {
                //                      printf("[ ");
                //                  }
                //                  printf("%x ", buffer[i]);
                //                  if (i+1 == buffer.size() || i == 149)
                //                  {
                //                      printf("] ");
                //                  }
                //              }
                //              printf("\n");
                tun_fifo.pop();
            }
        }
    }
}

int main	(	int	argc,
		char *	argv[]
	)

{
#if defined(COSMOS_LINUX_OS)

    int32_t iretn;
    char tunnel_ip[20];
    vector<uint8_t> buffer;

    switch (argc)
    {
    case 4:
        {
            rxr_devname = argv[3];
            size_t tloc = rxr_devname.find(":");
            if (tloc != string::npos)
            {
                rxr_baud = atol(rxr_devname.substr(tloc+1, rxr_devname.size()-(tloc+1)).c_str());
                rxr_devname = rxr_devname.substr(0, tloc);
            }
            // Open receiver port
            rxr_serial = new Serial(rxr_devname, rxr_baud);
            if (rxr_serial->get_error() < 0)
            {
                printf("Error opening %s as receiver - %s\n",rxr_devname.c_str(), cosmos_error_string(rxr_serial->get_error()).c_str());
                exit (-1);
            }
        }
    case 3:
        {
            txr_devname = argv[2];
            size_t tloc = txr_devname.find(":");
            if (tloc != string::npos)
            {
                txr_baud = atol(txr_devname.substr(tloc+1, txr_devname.size()-tloc+1).c_str());
                txr_devname = txr_devname.substr(0, tloc);
            }
            // Open receiver port
            txr_serial = new Serial(txr_devname, txr_baud);
            txr_serial->set_timeout(1.);
            if (txr_serial->get_error() < 0)
            {
                printf("Error opening %s as receiver - %s\n",txr_devname.c_str(), cosmos_error_string(txr_serial->get_error()).c_str());
                exit (-1);
            }
            // Copy transmitter to  receiver port if not already open (Duplex)
            if (rxr_serial == nullptr)
                rxr_serial = txr_serial;
            // Get address for tunnel
            strcpy(tunnel_ip,argv[1]);
        }
        break;
    default:
        printf("Usage: agent_tunnel ip_address transmit_device [receive_device]\n");
        exit (-1);
    }

    // Initialize the Agent
    agent = new Agent("", "tunnel", 1., MAXBUFFERSIZE, true);
    if ((iretn = agent->wait()) < 0)
    {
        fprintf(agent->get_debug_fd(), "%16.10f %s Failed to start Agent %s on Node %s Dated %s : %s\n",currentmjd(), mjd2iso8601(currentmjd()).c_str(), agent->getAgent().c_str(), agent->getNode().c_str(), utc2iso8601(data_ctime(argv[0])).c_str(), cosmos_error_string(iretn).c_str());
        exit(iretn);
    }
    else
    {
        fprintf(agent->get_debug_fd(), "%16.10f %s Started Agent %s on Node %s Dated %s\n",currentmjd(), mjd2iso8601(currentmjd()).c_str(), agent->getAgent().c_str(), agent->getNode().c_str(), utc2iso8601(data_ctime(argv[0])).c_str());
    }


    // Start serial threads
    thread tcv_read_thread(tcv_read_loop);
    thread tcv_write_thread(tcv_write_loop);

    // Open tunnel device
    int tunnel_sock;
    struct ifreq ifr1, ifr2;
    struct sockaddr_in *addr = (struct sockaddr_in *)&ifr2.ifr_addr;

    if ((tun_fd=open("/dev/net/tun", O_RDWR)) < 0)
    {
        perror("Error opening tunnel device");
        exit (-1);
    }

    memset(&ifr1, 0, sizeof(ifr1));
    ifr1.ifr_flags = IFF_TUN | IFF_NO_PI;
    strncpy(ifr1.ifr_name, agent->cinfo->agent[0].beat.proc, IFNAMSIZ);
    if (ioctl(tun_fd, TUNSETIFF, static_cast<void *>(&ifr1)) < 0)
    {
        perror("Error setting tunnel interface");
        exit (-1);
    }

    if((tunnel_sock=socket(AF_INET, SOCK_DGRAM, IPPROTO_IP)) < 0)
    {
        perror("Error opening tunnel socket");
        exit (-1);
    }

    // Get ready to set things
    strncpy(ifr2.ifr_name, agent->cinfo->agent[0].beat.proc, IFNAMSIZ);
    ifr2.ifr_addr.sa_family = AF_INET;

    // Set interface address

    inet_pton(AF_INET, tunnel_ip, &addr->sin_addr);
    if (ioctl(tunnel_sock, SIOCSIFADDR, &ifr2) < 0 )
    {
        perror("Error setting tunnel address");
        exit (-1);
    }

    // Set interface netmask
    inet_pton(AF_INET, static_cast<const char *>("255.255.255.0"), &addr->sin_addr);
    if (ioctl(tunnel_sock, SIOCSIFNETMASK, &ifr2) < 0 )
    {
        perror("Error setting tunnel netmask");
        exit (-1);
    }

    if (ioctl(tunnel_sock, SIOCGIFFLAGS, &ifr2) < 0 )
    {
        perror("Error getting tunnel interface flags");
        exit (-1);
    }

    // Bring interface up
    ifr2.ifr_flags |= (IFF_UP | IFF_RUNNING);
    if (ioctl(tunnel_sock, SIOCSIFFLAGS, &ifr2) < 0 )
    {
        perror("Error setting tunnel interface flags");
        exit (-1);
    }

    // Set interface MTU
    ifr2.ifr_mtu = 250;
    if (ioctl(tunnel_sock, SIOCSIFMTU, &ifr2) < 0 )
    {
        perror("Error setting tunnel interface MTU");
        exit (-1);
    }


    close(tunnel_sock);

    // Start tunnel threads
    thread tun_read_thread(tun_read_loop);
    thread tun_write_thread(tun_write_loop);

    double nmjd = currentmjd(0.);
    int32_t sleept;

    // Start performing the body of the agent
    while(agent->running())
    {
        // Set beginning of next cycle;
        nmjd += agent->cinfo->agent[0].aprd/86400.;

        sleept = (int32_t)((nmjd - currentmjd(0.))*86400000000.);
        if (sleept < 0)
        {
            sleept = 0;
        }
        COSMOS_USLEEP(sleept);
    }
#endif
    exit (0);
}

Variable Documentation

Agent* agent

static

std::queue<vector<uint8_t> > tun_fifo

static

std::queue<vector<uint8_t> > tcv_fifo

static

std::condition_variable tcv_fifo_check

static

std::condition_variable tun_fifo_check

static

int tun_fd

static

string rxr_devname

static

string txr_devname

static

Serial* rxr_serial =nullptr

static

Serial* txr_serial =nullptr

static

uint32_t rxr_baud = 9600

static

uint32_t txr_baud = 9600

static