fart/distrib/distrib.cc

#include "distrib.h"
#include <fstream>
#include <string>
#include <iostream>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <netdb.h>                  /* gethostbyname() */
#include <string.h>                 /* memset() */
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/ip.h>
using namespace std;

distrib::distrib()
{
    pthread_cond_init(&m_listen_cond, NULL);
    pthread_mutex_init(&m_listen_mutex, NULL);
    m_num_clients = 0;
    m_data = NULL;
    m_server = true;
}

int distrib::readHostFile(const char * filename)
{
    ifstream ifs(filename);

    if ( ! ifs.is_open() )
        return 1;

    string host;
    while ( ! ifs.eof() )
    {
        ifs >> host;
        if ( ifs.eof() )
            break;
        m_hosts.push_back(host);
    }

    ifs.close();
    return 0;
}

int distrib::startClients(const std::vector<std::string> & client_options)
{
    int ret = 0;
    for (int i = 0, sz = m_hosts.size(); i < sz; i++)
    {
        ret += clientConnect(m_hosts[i], client_options);
    }
    return ret;
}

int distrib::clientConnect(const string & host,
                     const std::vector<std::string> & client_options)
{
    int id = fork();
    if (id < 0)         /* check for fork() error */
    {
        cerr << "Error forking: " << id << endl;
        return 1;
    }
    else if (id > 0)    /* in the parent */
    {
        m_children.push_back(id);
        m_num_clients++;
    }
    else                /* in the child */
    {
        char server_port_str[15];
        sprintf(server_port_str, "%d", m_serverport);
        vector<string> args;
        args.push_back("ssh");
        args.push_back(host);
        args.push_back("fart");
        args.push_back("--host");
        args.push_back(m_servername);
        args.push_back("--port");
        args.push_back(server_port_str);
        for (int i = 0, sz = client_options.size(); i < sz; i++)
            args.push_back(client_options[i]);
        const char * char_star_args[args.size() + 1];
        for (int i = 0, sz = args.size(); i < sz; i++)
            char_star_args[i] = args[i].c_str();
        char_star_args[args.size()] = (char *) NULL;

#if 0
        /* debug */
        cout << "executing: 'ssh', ";
        for (int i = 0, sz = args.size(); i < sz; i++)
            cout << "'" << char_star_args[i] << "', ";
        cout << endl;
#endif

        execvp("ssh", (char * const *) char_star_args);

        /* we should not get here */
        cerr << "Error " << errno << " with execlp()!" << endl;
        exit(33);
    }

    return 0;
}

void distrib_server(distrib * the_distrib)
{
    char hostname[1000];
    gethostname(&hostname[0], 1000);
    the_distrib->m_servername = hostname;

    int listen_socket = socket(PF_INET, SOCK_STREAM, 0);
    if ( listen_socket == -1 )
    {
        cerr << "Error " << errno << " creating listen socket!" << endl;
        exit(39);
    }

    if ( listen(listen_socket, 5) == -1 )
    {
        cerr << "Error " << errno << " when trying to listen!" << endl;
        exit(40);
    }

    struct sockaddr_in addr;
    int addr_len = sizeof(struct sockaddr_in);
    getsockname(listen_socket,
                (struct sockaddr *) &addr,
                (socklen_t *) &addr_len);

    int ip_addr = ntohl(addr.sin_addr.s_addr);
    the_distrib->m_serverport = ntohs(addr.sin_port);
    cout << "Listening on "
         << (unsigned int) ((ip_addr >> 24) & 0xFF)
         << '.'
         << (unsigned int) ((ip_addr >> 16) & 0xFF)
         << '.'
         << (unsigned int) ((ip_addr >> 8) & 0xFF)
         << '.'
         << (unsigned int) (ip_addr & 0xFF)
         << ':'
         << the_distrib->m_serverport
         << endl;

    /* signal readiness of the listen thread */
    pthread_mutex_lock(&the_distrib->m_listen_mutex);
    pthread_cond_signal(&the_distrib->m_listen_cond);
    pthread_mutex_unlock(&the_distrib->m_listen_mutex);
}

int distrib::startServer()
{
    m_server = true;

    pthread_mutex_lock(&m_listen_mutex);

    /* start the listen thread */
    int ret = pthread_create(&m_server_thread,
                             NULL,
                             (void * (*)(void *)) distrib_server,
                             this);
    if (ret)
        return ret;

    /* wait for the listen thread to be running */
    pthread_cond_wait(&m_listen_cond, &m_listen_mutex);
    pthread_mutex_unlock(&m_listen_mutex);

    return ret;
}

int distrib::startClient(const char * server, int port)
{
    m_server = false;

    m_client_socket = socket(PF_INET, SOCK_STREAM, 0);
    if (m_client_socket < 0)
    {
        cerr << "Error creating client socket: " << errno << endl;
        return 1;
    }

    struct addrinfo hint;
    memset(&hint, 0, sizeof(hint));
    hint.ai_family = AF_INET;
    hint.ai_socktype = SOCK_STREAM;

    struct addrinfo * res;
    char portstr[15];
    sprintf(portstr, "%d", port);
    getaddrinfo(server, portstr, &hint, &res);

    if (connect(m_client_socket, res->ai_addr, res->ai_addrlen) == -1)
    {
        cerr << "Error connecting from client socket: " << errno << endl;
        return 2;
    }

    return 0;
}

int distrib::send_data(int task, unsigned char * data, int num_bytes)
{
    /* TODO: finish */
}

int distrib::getTask()
{
    /* TODO: finish */
    if (m_server)
    {
    }
    else
    {
    }
    return -1;
}