---

control.c

/* DICElib (DIstributed CAVE Engine library) 
 * Copyright (c) 2001 Bruno Barberi Gnecco <brunobg@lsi.usp.br>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to
 * deal in the Software without restriction, including without limitation the
 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
 * sell copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies of the Software, its documentation and marketing & publicity
 * materials, and acknowledgment shall be given in the documentation, materials
 * and software packages that this Software was used.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
 * THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
 * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

#include "_dicelib.h"
#include <string.h>
#include <errno.h>
#ifndef WINDOWS
#include <signal.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <netdb.h>
#else
#include <winsock.h>
#endif

static int                      syncnum = 0;    /* # of clients that sent SYNC */
static int                      readynum = 0;   /* # of clients that sent READY */
static int                      fdmax;          /* maximum socket number */
static fd_set                   clients;        /* list of sockets */
static int                      *hierarchy;
int                             packetfrom;     /* socket to the received socket */

#define FUNCTION                "dice_broadcast"

int dice_broadcast ( dice_packet *packet, int except ) {
        int                     i;

        _dice_debug(3, dice_printf(FUNCTION, "(%p, %d)\n", packet, except);)
        for ( i = 0; i <= fdmax; i++ ) {
                if ( FD_ISSET(i, &clients) && i != except )
                         if ( dice_packet_send(packet, i) == -1 )
                                return -1;
        }
        return 0;
}

void dice_control_byebye ( int signum ) {
        dice_packet             packet;
        int                     i;

        _dice_debug(3, dice_printf("dice_control_byebye", "(%d)\n", signum);)
        if ( signum == SIGTERM )
                _dice_debug(3, dice_printf(NULL, "Control: got SIGTERM\n");)

        /* broadcast a goodbye packet */
        packet.packtype = GOODBYE;
        dice_broadcast(&packet, (signum == -1 ? readynum : -1));

#ifdef USE_ASYNC
        dice_async_end();
#endif
        dice_sync_end();
        if ( dice_data.clients ) {
                for ( i = 0; i < dice_data.clientnum; i++ ) {
                        dice_free(dice_data.clients[i].address);
                        dice_free(dice_data.clients[i].path);
                        dice_free(dice_data.clients[i].function);
                }
/*              dice_free(dice_data.clients); segfaults, why?*/
        }
        dice_free(dice_data.color[0]);
        dice_free(dice_data.color[1]);
        dice_free(hierarchy);
/*      dice_type_free_all(); segfaults, why? */

#ifndef WINDOWS
        signal(SIGTERM, SIG_DFL);
#endif
        exit(0);
}


int dice_control_hierarchy ( int a, int b ) {
        int                             i;
        
        if ( a == b )
                return 1;
        for ( i = 0; i < dice_data.clientnum; i++ ) {
                if ( hierarchy[i] == a )
                        return 0;
                else if ( hierarchy[i] == b ) 
                        return 1;
        }
        return -1;
}


#undef FUNCTION
#define FUNCTION                "dice_control_process"

int dice_control_process ( dice_packet *packet, int from ) {
        dice_packet                     p;
        _dice_debug(3, dice_printf(FUNCTION, "(%p, %d); type %s\n", packet, from,
                        dice_packet_name(packet->packtype));)

        switch ( packet->packtype ) {
          case INVALID:
                return 0;
          case SYNC:
                /* get the updates */
                p = dice_packet_get(from);
                packetfrom = from;
                while ( p.packtype == VARIABLE_SYNC ||
                        p.packtype == VARIABLE_FREE_SYNC ) {

                        if ( dice_data.original_only && hierarchy[0] != from )
                                continue;

                        if ( p.packtype == VARIABLE_SYNC ) {
                                if ( DICE_sync_update(p.variable.name, 
                                                p.variable.value) == -2 )
                                        DICE_sync_new(p.variable.name, 
                                                p.variable.type, 
                                                p.variable.value );
                        }
                        else 
                                DICE_sync_delete(p.variable.name);
                        free(p.variable.name);
                        free(p.variable.type);
                        p = dice_packet_get(from);
                }
                if ( p.packtype != READY ) {
                        _dice_debug(1, dice_printf(FUNCTION, "did not get READY, got %s instead\n",
                                        dice_packet_name(p.packtype));)
                        return -1;
                }

                syncnum++;
                if ( syncnum == dice_data.clientnum ) { 
                        /* everybody sent a SYNC and is waiting. */
                        dice_sync_update_send();

                        p.packtype = SYNC;
                        dice_broadcast(&p, -1);
                        syncnum = 0;
                }
                return 0;
          case VARIABLE_ASYNC:
          case VARIABLE_FREE_ASYNC:
#ifdef USE_ASYNC
                if ( dice_data.original_only && hierarchy[0] != from ) {
                        DICE_async_update(packet->variable.name, packet->variable.value);
                        dice_broadcast(packet, from);
                }
#endif
                return 0;
          case VARIABLE_SYNC:
          case VARIABLE_FREE_SYNC:
                _dice_debug(1, dice_printf(FUNCTION, "got a lonely %s!\n", 
                                dice_packet_name(p.packtype));)
                return -1;
          case READY:
                readynum++;
                if ( readynum == dice_data.clientnum ) {
                        p.packtype = CONTINUE;
                        dice_broadcast(&p, -1);
                        readynum = 0;
                }
                return 0;
          case CONTINUE:
                _dice_debug(1, dice_printf(FUNCTION, "Control should never get CONTINUE!\n");)
                return -1;
          case GOODBYE:
                /* if a client said bye, quit all of them */
                readynum = from; /* except the one that sent the packet */
                dice_control_byebye(-1);
                return 0;
          case HELLO:
                _dice_debug(1, dice_printf(FUNCTION, "Control should never get HELLO!\n");)
                return -1;
        }
        return -1;
}

#undef FUNCTION
#define FUNCTION                "dice_control_loop"

int dice_control_loop ( int argc, char **argv ) {
        struct sockaddr_in      myaddr, remote;
        int                     sockfd, i;
        fd_set                  fdtemp;
        dice_packet             packet;

        _dice_debug(3, dice_printf(FUNCTION, "()\n");)

#ifndef WINDOWS
        /* first of all, handle SIGTERM */
        signal(SIGTERM, dice_control_byebye);
#endif

        FD_ZERO(&clients);
        FD_ZERO(&fdtemp);

        hierarchy = (int *)malloc(dice_data.clientnum*sizeof(int));
        if ( !hierarchy ) {
                _dice_debug(1, dice_printf(FUNCTION, "hierarchy: %s\n", strerror(errno));)
                return -1;
        }

        /* Opening socket */
        if ( (sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0 ) {
                _dice_debug(1, dice_printf(FUNCTION, "%s\n", strerror(errno));)
                return -1;
        }

        /* non blocking ? 
        fcntl(sockfd, F_SETFL, O_NOBLOCK); */

        /* get rid of "address already in use" messages */
        i = 1;
        setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &i, sizeof(int));
        
        /* TCP NODELAY */
        i = 1;
        setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &i, sizeof(int));

        /* Fill out the struct sockaddr_in structure */
        myaddr.sin_port = htons(dice_data.port);
        myaddr.sin_family = AF_INET;
        myaddr.sin_addr.s_addr = INADDR_ANY; /* Use my IP address */
        memset((void *)&myaddr.sin_zero, 0, 8);

        /* Bind to the port */
        if ( bind(sockfd, (struct sockaddr *)&myaddr, sizeof(myaddr)) < 0 ) {
                _dice_debug(1, dice_printf(FUNCTION, ": %s\n", strerror(errno));)
                return -1;
        }

        /* Limit connections */
        if ( listen(sockfd, dice_data.max_connections) < 0 ) {
                _dice_debug(1, dice_printf(FUNCTION, "%s\n", strerror(errno));)
                return -1;
        }

        /* accept connections from everybody. todo: timeout. */
        _dice_debug(3, dice_printf(FUNCTION, "waiting for %d clients.\n", dice_data.clientnum);)
        for ( i = 0; i < dice_data.clientnum; i++ ) {
                int len = sizeof(remote), fdnew, pid;

                /* distribute around the network */
                if ( dice_data.spawn != NONE ) {
                        _dice_debug(3, dice_printf(FUNCTION, "Spawning %s...\n",
                                        dice_data.clients[i].address);)

                        argv[0] = dice_data.clients[i].address;
                        argv[1] = ( dice_data.clients[i].path == NULL ?
                                defaultpath : dice_data.clients[i].path );
                        argv[3] = dice_data.controlname;
                        pid = fork();
                        if ( pid == 0 ) { /* child */
                                switch ( dice_data.spawn ) {
                                  case RSH:
                                        execvp("rsh", argv);
                                        break;
                                  case SSH:
                                        execvp("ssh", argv);
                                        break;
                                  case BPROC:
                                        _dice_debug(1, dice_printf(FUNCTION, 
                                                        "BPROC not implemented yet");)
                                        /* execvp("bproc", argv); */
                                        break;
                                  default:
                                        break;
                                }
                                _dice_debug(1, dice_printf(FUNCTION, "exec: %s\n",
                                                strerror(errno));)
                                _exit(1);
                        }
                        if ( pid < 0 ) {
                                _dice_debug(1, dice_printf(FUNCTION, "fork: %s\n",
                                                strerror(errno));)
                                exit(1); /*todo: should be nicer */
                        }

                }

                /* now wait for connection. todo: timeout. perhaps catching SIGCHILD? */
                if ( (fdnew = accept(sockfd, (struct sockaddr *)&remote, &len)) == -1 ) {
                        _dice_debug(1, dice_printf(FUNCTION, 
                                "error connecting: %s\n", strerror(errno));)
                }
                else {
                        struct hostent          *host;
                        int                     len2 = sizeof(remote.sin_addr);
                        char                    *name;

                        FD_SET(fdnew, &clients);
                        if ( fdnew > fdmax )
                                fdmax = fdnew;

                        /* hierarchy is done here! */
                        hierarchy[i] = fdnew;
                        
                        /* Get client name */
                        len2 = sizeof(remote.sin_addr);
                        name = inet_ntoa(remote.sin_addr);
                        host = gethostbyaddr((char *)&remote.sin_addr, len2, AF_INET);
                        if ( !host ) {
                                _dice_debug(1, dice_printf(FUNCTION, 
                                        "couldn't get host name for %s: %s\n"
                                        "This may lead to problems in hierarchy.\n", 
                                        inet_ntoa(remote.sin_addr), strerror(errno));)
                                continue;
                        } 
                        _dice_debug(2, dice_printf(FUNCTION, 
                                "Connection from socket %d by %s(%s)\n", fdnew,
                                host->h_name, inet_ntoa(remote.sin_addr));)
                                        
                        /* and send a HELLO packet */
                        packet.packtype = HELLO;
                        packet.hello.id = (unsigned long)i;
                        packet.hello.function = dice_data.clients[i].function;
                        dice_packet_send(&packet, fdnew);
                }
        }

        _dice_debug(2, dice_printf(FUNCTION, "All connected");)

        while ( 1 ) {
                fdtemp = clients; /* copy */
                if ( select(fdmax+1, &fdtemp, NULL, NULL, NULL) == -1 ) {
                        _dice_debug(1, dice_printf(FUNCTION, "%s\n", strerror(errno));)
                }

                for ( i = 0; i <= fdmax; i++ ) {
                        if ( FD_ISSET(i, &fdtemp) && i != sockfd ) { /* no more connections */
                                /* handle */
                                packet = dice_packet_get(i);
                                dice_control_process(&packet, i);
                        }
                }
        }
        return -1; /* so the compiler won't complain */
}

---