DistRen Changeset - 27eb6c6418f6 · protofusion repositories

Changeset - 27eb6c6418f6

Parent rev.

Child rev.

[Not reviewed]

default

0 9 0

Nathan Brink (binki) - 15 years ago 2010-07-21 23:03:26
ohnobinki@ohnopublishing.net

Handle resolving hostnames gracefully.
Fix support for hostname:port in remoteio.
Added timeout parameter to multiio_poll().
Fix stupid mistakes in distrenslave's read handler.
The server and client may not sit and play table tennis at last :-D.

9 files changed with 57 insertions and 36 deletions:

src/common/multiio.c

src/common/multiio.h

src/common/protocol.c

src/common/remoteio.c

src/common/request.c

src/server/distrend.c

src/server/listen.c

src/server/slave.c

src/server/tabletennis.c

0 comments (0 inline, 0 general)

src/common/multiio.c

➞

Show inline comments

@@ @@ -81,200 +81,200 @@ struct multiio_socket_type_handler_info @@
 };
 /**
    Finds the index in the context->pollfds array of a particular socket.
    @param context the multiio context
    @param fd the socket to search for
    @param index a pointer which will be filled with the index of the socket if found
    @return 0 if the pollfds entry is found, 1 if the entry is not found
  */
 int multiio_pollfd_index_by_fd(const multiio_context_t context, int fd, size_t *index);
 multiio_context_t multiio_context_new()
+{
   struct multiio_context *context;
   context = malloc(sizeof(struct multiio_context));
   if(!context)
     return NULL;
   context->pollfds = malloc(sizeof(struct pollfd) * 2);
   if(!context->pollfds)
+    {
       free(context);
       return NULL;
+    }
   context->pollfds_len = 2;
   context->nfds = 0;
   context->socket_infos = malloc(sizeof(struct multiio_socket_info) * 2);
   context->num_socket_types = 0;
   context->socket_types = NULL;
   if(!context->pollfds
      || !context->socket_infos)
+    {
       free(context->pollfds);
       free(context->socket_infos);
       free(context);
       return NULL;
+    }
   return context;
+}
 int multiio_context_free(multiio_context_t context)
+{
   size_t counter;
   /*
     clean up pollfds and socket_infos
    */
   free(context->pollfds);
   free(context->socket_infos);
   /*
     wipe up the socket_types
    */
   for(counter = 0; counter < context->num_socket_types; counter ++)
     list_free(context->socket_types[counter].type_handlers, LIST_DEALLOC);
   free(context->socket_types);
   /*
     bye!
    */
   free(context);
   return 0;
+}
 struct multiio_poll_travinfo
+{
   multiio_context_t multiio;
   struct multiio_socket_info *socket_info;
   struct pollfd *pollfd;
 };
 /**
    list traverser for multiio_poll()
  */
 int multiio_poll_invoke_handlers(struct multiio_poll_travinfo *travinfo, struct multiio_socket_type_handler_info *handler_info)
+{
   short event;
   event = travinfo->pollfd->revents & handler_info->event;
   if(!event)
     return TRUE;
   handler_info->handler(travinfo->multiio,
 			travinfo->pollfd->fd,
 			event,
 			handler_info->handler_data,
 			travinfo->socket_info->socket_data);
   return TRUE;
+}
 int multiio_poll(multiio_context_t context)
+int multiio_poll(multiio_context_t context, int timeout)
+{
   size_t counter;
   struct multiio_poll_travinfo travinfo;
   int ret;
-  ret = poll(context->pollfds, context->nfds, -1);
+  ret = poll(context->pollfds, context->nfds, timeout);
   if(ret == -1)
+    {
       perror("poll");
+    }
   for(counter = 0; counter < context->nfds; counter ++)
     if(context->pollfds[counter].revents)
+      {
 	travinfo.multiio = context;
 	travinfo.socket_info = &context->socket_infos[counter];
 	travinfo.pollfd = &context->pollfds[counter];
 	list_traverse(context->socket_types[context->socket_infos[counter].socket_type].type_handlers,
 		      &travinfo,
 		      (list_traverse_func_t)&multiio_poll_invoke_handlers,
 		      LIST_FRNT | LIST_SAVE);
 	context->pollfds[counter].revents = 0;
+      }
   return 0;
+}
 int multiio_socket_type_register(multiio_context_t context, multiio_socket_type_t *new_type)
+{
   struct multiio_socket_type_info *new_socket_types;
   new_socket_types = malloc(sizeof(struct multiio_socket_type_info) * (context->num_socket_types + 1));
   if(!new_socket_types)
     return 1;
   *new_type = context->num_socket_types;
   new_socket_types[*new_type].type_handlers = list_init();
   if(!new_socket_types[*new_type].type_handlers)
+    {
       free(new_socket_types);
       return 2;
+    }
   if(context->num_socket_types)
+    {
       memcpy(new_socket_types, context->socket_types, sizeof(struct multiio_socket_type_info) * context->num_socket_types);
       free(context->socket_types);
+    }
   context->socket_types = new_socket_types;
   context->num_socket_types ++;
   return 0;
+}
 int multiio_event_handler_register(multiio_context_t context, multiio_socket_type_t socket_type, short event, multiio_event_handler_func_t handler_func, void *handler_data)
+{
   struct multiio_socket_type_handler_info handler_info;
   if(socket_type >= context->num_socket_types)
     return 1;
   if(!event)
     return 2;
   handler_info.event = event;
   handler_info.handler = handler_func;
   handler_info.handler_data = handler_data;
   list_insert_after(context->socket_types[socket_type].type_handlers, &handler_info, sizeof(struct multiio_socket_type_handler_info));
   return 0;
+}
 int multiio_socket_add(multiio_context_t context, int fd, multiio_socket_type_t socket_type, void *socket_data, short events)
+{
   struct pollfd *pollfds;
   struct multiio_socket_info *socket_infos;
   size_t new_pollfds_len;
   int fdstatus;
   int tmp;
   fdstatus = fcntl(fd, F_GETFL);
   fdstatus |= O_NONBLOCK;
   tmp = fcntl(fd, F_SETFL, fdstatus);
   if(tmp == -1)
+    {
       perror("fcntl");
       return 1;
+    }
   if(socket_type >= context->num_socket_types)
     return 1;
   /**
      extend sockfds array
    */
   if(context->nfds >= context->pollfds_len)
+    {

src/common/multiio.h

➞

Show inline comments

 /*
   Copyright 2010 Nathan Phillip Brink
   This file is a part of DistRen.
   DistRen is free software: you can redistribute it and/or modify
   it under the terms of the GNU Affero General Public License as published by
   the Free Software Foundation, either version 3 of the License, or
   (at your option) any later version.
   DistRen 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 Affero General Public License for more details.
   You should have received a copy of the GNU Affero General Public License
   along with DistRen.  If not, see <http://www.gnu.org/licenses/>.
 */
 #ifndef DISTREN_MULTIIO_H
 #define DISTREN_MULTIIO_H
 /**
    @file multiio provides the event-oriented interface to accessing
    sockets of different sorts. Its purpose is to provide a unified
    interface around poll() which distrend and distren (client) would
    both be able to use. This means supporting tracking file descriptors
    from sockets of different types that must be handled differently, etc.
    This interface will be as simple as possible to provide greatest
    flexibility and most ease of maintainence.
  */
 struct pollfd;
 /**
    A socket type or classification.
    Each socket type is associated with a set of socket event handlers.
    This allows, for example, a listen()/accept() socket to be distinguised
    and reacted to differently from a read()/write() socket. It would also
    allow one set of read()/write() sockets to be treated differently from
    another set of read()/write() sockets.
  */
 typedef int multiio_socket_type_t;
 struct multiio_context;
 /**
    The multiio_context a program is using.
    Each process should only use one multiio_context. Different types of events
    should be handled through the use of different multiio_socket_type_t values.
  */
 typedef struct multiio_context *multiio_context_t;
 /**
    Initializes and returns a multiio_context.
  */
 multiio_context_t multiio_context_new();
 /**
    Destroys and frees a multiio_context.
    @param context the context to destroy.
  */
 int multiio_context_free(multiio_context_t context);
 /**
    Call poll() on the registered sockets and react to events accordingly.
    @param context the context which sockets and handlers were registered with
  * Call poll() on the registered sockets and react to events accordingly.
+ *
  * @param context the context which sockets and handlers were registered with
+ *
  * @param timeout see poll(3p). The number of milliseconds to wait for
  * a request before returning. 0 for immediate return if no activity
  * and -1 for waiting until some activity happens before returning.
  */
 int multiio_poll(multiio_context_t context);
+int multiio_poll(multiio_context_t context, int timeout);
 /**
    Registers a new socket type/classification for use with this multiio_context.
    @param context the multiio context to register this type with.
    @param new_type a pointer to a type which this function will set to the value of
    the newly registered type. The caller should record this type and use it for
    future multiio_event_handler_register() calls, etc.
    @return 0 on success, other on error
  */
 int multiio_socket_type_register(multiio_context_t context, multiio_socket_type_t *new_type);
 /**
    The basic event handler prototype.
    @param multiio the multiio context which is required for many multiio API functions.
    @param fd the handle for which data was recieved.
    @param revent the event that this handler was configured to respond to.
    @param handler_data the same pointer named handler_data that was passed to multiio_event_handler_register(). This allows handlers to store general context information (i.e., pointer to a general settings struct)
    @param socket_data the same pointer named socket_data that was passed to multiio_socket_add(). This allows handlers to access socket-specific information.
  */
 typedef int(*multiio_event_handler_func_t)(multiio_context_t multiio, int fd, short revent, void *handler_data, void *socket_data);
 /**
    Register a socket event handler for a specified socket type/class.
    Don't forget to register handlers for POLLERR, POLLHUP, and POLLNVAL.
    You can probably just register one function to handle each of these
    categories and clean up the socket. (and then don't forget to call
    multiio_socket_del() :-p (or to free your per-socket pointers, etc.)).
    Handlers are called in the order they're registered.
    @param context the multiio_context
    @param socket_type the type of socket this handler should be registered for.
    @param event the poll() event on which this handler should be called.
    @param handler_func the function to be called when event is matched.
    @param handler_data a pointer that will be passed to handler_func whenever this event is matched for sockets of this socket_type.
  */
 int multiio_event_handler_register(multiio_context_t context, multiio_socket_type_t socket_type, short event, multiio_event_handler_func_t handler_func, void *handler_data);
 /**
    Register a socket of a specified type so that it may be checked for events
    and the event handlers for its socket_type may be called.
    This function will set the socket into nonblocking mode.
    @param context the multiio context.
    @param fd the socket.
    @param socket_type the type/classification of this socket
    @param socket_data a pointer to socket-specific data to pass to the multiio_event_handler_func_t.
    @param events the poll() events with which this socket should be set to listen for. Certain events may be added or removed later using multiio_socket_event_enable() and multiio_socket_event_disable().
  */
 int multiio_socket_add(multiio_context_t context, int fd, multiio_socket_type_t socket_type, void *socket_data, short events);
 /**
    Unregisters a socket from a multiio context
    @param context the context
    @param fd the socket to remove
  */
 int multiio_socket_del(multiio_context_t context, int fd);
 /**
    Enable checking for the specified event for the specified socket.
    The specified socket must have been beforehand registered using multiio_socket_add().
    @param context the multiio context
    @param fd the socket for which checking of an event should be enabled
    @param event the poll() event which should be enabled for this socket.
  */
 int multiio_socket_event_enable(multiio_context_t context, int fd, short event);
 /**
    Disable checking for the specified event for the specified socket.
    Disabling all events for a particular socket does not remove it from the
    multiio context.
    @param context the multiio context
    @param fd the socket for which checking of an event should be disabled
    @param event the poll() event which should be disabled on this socket.
  */
 int multiio_socket_event_disable(multiio_context_t context, int fd, short event);
 #endif /* DISTREN_MULTIIO_H */

src/common/protocol.c

➞

Show inline comments

 /*
   Copyright 2010 Nathan Phillip Brink
   This file is a part of DistRen.
   DistRen is free software: you can redistribute it and/or modify
   it under the terms of the GNU Affero General Public License as published by
   the Free Software Foundation, either version 3 of the License, or
   (at your option) any later version.
   DistRen 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 Affero General Public License for more details.
   You should have received a copy of the GNU Affero General Public License
   along with DistRen.  If not, see <http://www.gnu.org/licenses/>.
 */
 #include "protocol.h"
 #include "remoteio.h"
 #include <malloc.h>
 #include <stdio.h>
 #include <string.h>
 #define DISTREN_REQUEST_MAGIC ((uint32_t)0x32423434)
 int distren_request_new(struct distren_request **req, uint32_t len, enum distren_request_type type)
+{
   struct distren_request *newreq;
   newreq = malloc(sizeof(struct distren_request));
   if(!newreq)
+    {
       (*req) = NULL;
       return 1;
+    }
   newreq->magic = DISTREN_REQUEST_MAGIC;
   newreq->len = len;
   newreq->type = type;
   (*req) = newreq;
   return 0;
+}
 int distren_request_send(struct remoteio *rem, struct distren_request *req, void *data)
+{
   void *packet;
   size_t len;
   int write_err;
   if(req->magic != DISTREN_REQUEST_MAGIC)
     fprintf(stderr, "distren_request_send got a bad req\n");
   len = sizeof(struct distren_request) + req->len;
   packet = malloc(len);
   if(!packet)
+    {
       fprintf(stderr, "Error allocating memory for packet\n");
       return 1;
+    }
   memcpy(packet, req, sizeof(struct distren_request));
   memcpy(packet + sizeof(struct distren_request), data, req->len);
   write_err = remoteio_write(rem, packet, len);
   free(packet);
   return 0;
+}
 int distren_request_new_fromdata(struct distren_request **req, void *data, size_t len)
+{
   struct distren_request *newreq;
 #if 0
   size_t counter;
   uint32_t debugtmp;
 #endif
   if(len < sizeof(struct distren_request))
     return 1;
   if( ((struct distren_request *)data)->magic != DISTREN_REQUEST_MAGIC )
+    {
       fprintf(stderr, "packet doesn't match magic stuffs\n");
       fprintf(stderr, "packet doesn't match magic stuffs");
 #if 0
       fputs("\n\tmagic=`", stderr);
       debugtmp = DISTREN_REQUEST_MAGIC;
       for(counter = 0; counter < sizeof(uint32_t); counter ++)
 	putc(((char *)&debugtmp)[counter], stderr);
       fputs("'\n\t`", stderr);
       for(counter = 0; counter < sizeof(struct distren_request); counter ++)
 	putc(((char *)data)[counter], stderr);
       fputs("'\n", stderr);
 #endif
       return 1;
+    }
   newreq = malloc(sizeof(struct distren_request));
   if(!newreq)
+    {
       fprintf(stderr, "OOM\n");
       return 1;
+    }
   memcpy(newreq, data, sizeof(struct distren_request));
   (*req) = newreq;
   return 0;
+}
 int distren_request_free(struct distren_request *req)
+{
   free(req);
   return 0;
+}

src/common/remoteio.c

➞

Show inline comments

@@ @@ -646,163 +646,169 @@ int _remoteio_sock_open(struct remoteio @@
 #endif /* _WIN32 */
 int _remoteio_sock_close(struct remoteio *rem)
+{
   close(rem->sock);
   return 0;
+}
 int _remoteio_sock_read(struct remoteio *rem, void *buf, size_t len, size_t *bytesread)
+{
   ssize_t readrtn;
   *bytesread = 0;
   readrtn = read(rem->sock, buf, len);
   /*
     The following is valid for blocking sockets:
    */
   if(readrtn == -1)
+    {
       /*
 	in this case, we may have been interrupted by a signal and errno == EINTR
 	or the connection was reset and errno = ECONNRESET
 	Some of these are not error conditions:
        */
       perror("read");
       if(errno != EINTR)
+	{
 	  fprintf(stderr, "error reading socket in remoteio_sock_read\n");
 	  return 1;
+	}
       return 0;
+    }
   *bytesread = readrtn;
   if(!readrtn)
+    {
       /*
 	means EOF except when FD is in ``message-nondiscard'' or ``message-discard''
 	modes.
        */
       return 1;
+    }
   return 0;
+}
 int _remoteio_sock_write(struct remoteio *rem, void *buf, size_t len, size_t *byteswritten)
+{
   ssize_t writertn;
   writertn = write(rem->sock, buf, len);
   if(writertn == -1)
+    {
       perror("write");
       if(errno != EINTR)
+	{
 	  fprintf(stderr, "error writing to socket in remoteio_sock_write()\n");
 	  return 1;
+	}
+    }
   *byteswritten = writertn;
   if(!writertn)
+    {
       /*
 	should we consider this an error? I'm pretty
 	sure we should :-/
        */
       fprintf(stderr, "write() returned 0 in remoteio_sock_write()\n");
       return 1;
+    }
   return 0;
+}
 /**
    TCP, taking advantage of the two generic socks read/write functions:
  */
 int _remoteio_tcp_open(struct remoteio *rem, struct remoteio_server *server)
+{
   int tmp;
   int tmp2;
   int sock;
   char *hostname;
   char *port;
   struct addrinfo addrinfo_hints;
   struct addrinfo *addrinfo_res;
   static char *default_port = REMOTEIO_DEFAULT_PORT;
   /**
      only hostname should be free()-ed, not port,
      because both are from the same block of malloc()-ed
      memory
    */
   hostname = strdup(server->hostname);
   for(port = hostname;
       *port && *port != ':';
       port ++)
+    ;
   if(*port)
+    {
       *port = '\0';
       port ++;
+    }
   else
-    port = REMOTEIO_DEFAULT_PORT;
+    port = default_port;
   memset(&addrinfo_hints, '\0', sizeof(struct addrinfo));
   addrinfo_hints.ai_family = AF_UNSPEC;
 #ifdef _WIN32
   /* windows lacks stuff documented in POSIX, I guess :-( */
   addrinfo_hints.ai_flags = 0;
 #else
   addrinfo_hints.ai_flags = AI_V4MAPPED | AI_ADDRCONFIG;
 #endif
   addrinfo_hints.ai_socktype = SOCK_STREAM;
-  tmp = getaddrinfo(server->hostname, port, &addrinfo_hints, &addrinfo_res);
   tmp = getaddrinfo(hostname, port, &addrinfo_hints, &addrinfo_res);
   if(tmp)
     fprintf(stderr, "error resolving %s:%s: %s\n", server->hostname, port, gai_strerror(tmp));
+    {
       fprintf(stderr, "error resolving %s:%s: %s\n", server->hostname, port, gai_strerror(tmp));
       free(hostname);
       return 1;
+    }
   fprintf(stderr, "connecting to %s[%s]:%s\n", server->hostname, addrinfo_res->ai_canonname, port);
   free(hostname);
   sock = socket(addrinfo_res->ai_family, SOCK_STREAM, addrinfo_res->ai_protocol);
   if(sock == -1)
+    {
       perror("socket");
       freeaddrinfo(addrinfo_res);
       return 1;
+    }
   tmp = connect(sock, addrinfo_res->ai_addr, addrinfo_res->ai_addrlen);
   tmp2 = errno;
   freeaddrinfo(addrinfo_res);
   errno = tmp2;
   if(tmp == -1)
+    {
       perror("connect");
       close(sock);
       return 1;
+    }
   rem->sock = sock;
   return 0;
+}
 int _remoteio_tcp_close(struct remoteio *rem)
+{
   close(rem->sock);
   return 0;
+}

src/common/request.c

➞

Show inline comments

 /*
  * Copyright 2010 Nathan Phillip Brink
+ *
  * This file is a part of DistRen.
+ *
  * DistRen is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published by
  * the Free Software Foundation, either version 3 of the License, or
  * (at your option) any later version.
+ *
  * DistRen 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 Affero General Public License for more details.
+ *
  * You should have received a copy of the GNU Affero General Public License
  * along with DistRen.  If not, see <http://www.gnu.org/licenses/>.
  */
 #include "common/protocol.h"
 #include <stdlib.h>
 #include <string.h>
 int distren_request_free_with_data(struct distren_request *req, void *data)
+{
   free(data);
   return distren_request_free(req);
+}
-uint32_t distren_request_poing(struct distren_request **req, void **data, short is_ping, const void *poing_cookie, size_t poing_data_len)
+int distren_request_poing(struct distren_request **req, void **data, short is_ping, const void *poing_cookie, size_t poing_data_len)
+{
   enum distren_request_type type;
   if(is_ping)
     type = DISTREN_REQUEST_PING;
   else
     type = DISTREN_REQUEST_PONG;
   distren_request_new(req, poing_data_len, type);
   (*data) = malloc(poing_data_len);
   memcpy(*data, poing_cookie, poing_data_len);
-  return (uint32_t)poing_data_len;
+  return 0;
+}

src/server/distrend.c

➞

Show inline comments

@@ @@ -94,193 +94,193 @@ int distrend_handle_version(struct gener @@
 void update_general_info(struct general_info *geninfo);
 int import_general_info(struct general_info *general_info);
 int update_xml_joblist(struct general_info *geninfo);
 /* **************Test Functions**************** */
 int interactiveTest(int test, multiio_context_t multiio, struct general_info *general_info);
 /* **************** Main ********************* */
 int main(int argc, char *argv[])
+{
   /* Parse arguments */
   int counter;
   int test = 0; /*< Interactive mode if 1 */
   int tmp;
   struct general_info general_info;
   multiio_context_t multiio;
   enum clientstatus
+  {
     CLIENTSTATUS_UNINITIALIZED = 0,
     CLIENTSTATUS_BUSY = 1,
     CLIENTSTATUS_IDLE = 2
   } clientstatus;
   clientstatus = CLIENTSTATUS_UNINITIALIZED;
   // xmlinit();
   for(counter = 0; counter < argc; counter ++)
+    {
       if(strcmp(argv[counter], "-h") == 0)
+      {
     	  fprintf(stderr, "Usage: distrend [option] \nStarts the distrend server\n\t-h\tshow this help\n\t-t\tlaunches queue testing interface \n");
 	  return 2;
+      }
       else if(strcmp(argv[counter], "-t") == 0)
+      {
     	  fprintf(stderr, "Entering into test mode...\n\n");
     	  test = 1;
+      }
+    }
   multiio = multiio_context_new();
   if(distrend_do_config(argc, argv, &general_info.config, multiio))
     return 1;
   /** preset paths */
   _distren_asprintf(&general_info.files.geninfo, "%s/general_info.xml",
 		    general_info.config->datadir);
   /** MySQL Connection */
   fprintf(stderr,"Connecting to mysql...\n");
   if(mysqlConnect(&general_info.conn,
 		  general_info.config->mysql_user,
 		  general_info.config->mysql_host,
 		  general_info.config->mysql_pass,
 		  general_info.config->mysql_database) )
+    {
       fprintf(stderr, "%s:%d: mysqlConnect() failed\n", __FILE__, __LINE__);
       fprintf(stderr, "don't test mysql stuff\n");
       interactiveTest(test, multiio, &general_info);
       return 1;
+    }
   fprintf(stderr,"Finished connecting!\n");
   /** Execute test function */
   interactiveTest(test, multiio, &general_info);
   general_info.config->listens = distrend_listens_new(multiio, &general_info, general_info.config->options);
   if(!general_info.config->listens)
+    {
       fprintf(stderr, "error initializing listens\n");
       return 1;
+    }
   remoteio_generic_data_set(general_info.config->options->remoteio, general_info.config->listens);
   for(counter = 0; general_info.config->listen_ports[counter]; counter ++)
+    {
       tmp = distrend_listen_add(general_info.config->listens, general_info.config->listen_ports[counter]);
       if(tmp)
+	{
 	  fprintf(stderr, "Error listening on port %d\n", general_info.config->listen_ports[counter]);
 	  return 1;
+	}
+    }
   distrend_listen_handler_add(general_info.config->listens, DISTREN_REQUEST_VERSION, &distrend_handle_version);
   int slaveKey = 0; // Remotio should set me on a per-slave basis
   /* Main Loop */
   general_info.config->die = 0;
   while(!general_info.config->die)
+    {
       int clientrequest = 0; /*< temporary example variable, will be replaced when we can handle messages */
       multiio_poll(multiio);
+      multiio_poll(multiio, 15000);
       tabletennis_serve(general_info.config->listens->tabletennis);
       /* Run the watchdog, @TODO: like every 10 mins or something */
       frame_watchdog(general_info.conn);
       struct frameset frame;
       struct distrenjob *job;
       distrenjob_new(&job);
       memset(&frame, '\0', sizeof(frame));
       /** If client requests work */
       if(clientrequest == DISTREN_REQUEST_GETWORK)
+	{
 	  int returnnum = find_jobframe(general_info.conn, slaveKey, &job->jobnum, &frame.num); // Finds a frame to render @FIXME: Slavenum :D
 	  if(returnnum); /* No work on server */
 	  else
 	    remotio_send_to_client(frame.num, job->jobnum); // Pseudo-sends data to client
+	}
       /* If the client states that they finished the frame */
       else if(clientrequest == DISTREN_REQUEST_DONEFRAME)
+	{
 	  clientstatus = CLIENTSTATUS_IDLE; // Sets the client back to idle
 	  finish_frame(general_info.conn, 0, job->jobnum, frame.num); // @TODO: Make sure this actually works.
+	}
       distrenjob_free(&job);
+    }
   distrend_listen_free(general_info.config->listens);
   distrend_config_free(general_info.config);
   xmlcleanup();
   /** free() paths */
   free(general_info.files.geninfo);
   mysqlDisconnect(general_info.conn);
   return 0;
+}
 /* ********************** Functions ************************* */
 int distrend_handle_version(struct general_info *geninfo, struct distrend_client *client, size_t req_len, void *req_data)
+{
   char *tmp_str;
   char fixedbuf[32];
   struct distren_request *newreq;
   if(client->state != DISTREND_CLIENT_PREVERSION)
+    {
       distrend_send_disconnect(client, "You have already sent the VERSION command.");
+    }
   if(strlen(PACKAGE_STRING) == req_len
      && !strncmp(PACKAGE_STRING, req_data, req_len))
+    {
       /**
 	 The client and I claim to be of the same version of distren :-D
 	 Now we will mark the client as valid.
 	 We won't increment his time to live, though, because it shouldn't take
 	 him that long to auth.
       */
       client->state = DISTREND_CLIENT_PREAUTH;
       distren_request_new(&newreq, strlen(PACKAGE_STRING), DISTREN_REQUEST_VERSION);
       distrend_client_write_request(client, newreq, PACKAGE_STRING);
       distren_request_free(newreq);
+    }
   else
+    {
       /**
 	 The client claims to be of a different version of distren.
 	 Now we will just send a disconnect packet and disconnect the client.
       */
       strncpy(fixedbuf, req_data, 31);
       fixedbuf[31] = '\0';
       if(req_len < 31)
 	fixedbuf[req_len] = '\0';
       _distren_asprintf(&tmp_str, "You have tried to connect to a %s server when your client claims to be running %s. Bye ;-)\n", PACKAGE_STRING, fixedbuf);
       distrend_send_disconnect(client, tmp_str);
+    }
   return 0;
+}
 /**
    Performs command stored in a client's request. @TODO: Fill stub
 */
 int distrend_do()
+{
   return 0;
+}
@@ @@ -486,111 +486,111 @@ int update_xml_joblist(struct general_in @@
   /**
      free writer and save xml file to disk
   */
   xmlFreeTextWriter(writer);
   return 0;
+}
 /* ************************** Test Functions ************************* */
 /** Interactive test for the queuing system */
 /* @QUEUE: Test uses methods not present in C code using mysql web-based system */
 int interactiveTest(int test, multiio_context_t multiio, struct general_info *geninfo)
+{
   int command;
   int32_t slaveKey = 1;
   jobnum_t jobKey = 0;
   int32_t frameNum = 0;
   int32_t newPriority = 0;
   int tmp = 0;
   fprintf(stderr,"Hello!\n");
   while(test == 1)
+   {
      fprintf(stderr, "Welcome to DistRen Alpha Interactive Test Mode\n\n");
      fprintf(stderr, "\t1 \tGet a frame to render\n");
      fprintf(stderr, "\t2 \tChange job priority\n");
      fprintf(stderr, "\t3 \tSet frame finished\n");
      fprintf(stderr, "\t4 \tSet frame started\n");
      fprintf(stderr, "\t5 \tStart listener\n");
      fprintf(stderr, "\t0 \tQuit\n");
      scanf("%d", &command);
      switch(command)
+     {
      case 1:
        fprintf(stderr,"Slave Key: ");
        scanf("%d", &slaveKey);
        fprintf(stderr, "Got frame: ");
        if(find_jobframe(geninfo->conn, slaveKey, &jobKey, &frameNum))
          fprintf(stderr,"No frames available to render!\n");
        else if(jobKey == -1)
          fprintf(stderr,"Slave %d has no render power!", slaveKey);
        else
          fprintf(stderr, "jobKey: %d, frameNum: %d\n",jobKey,frameNum);
        break;
      case 2:
        fprintf(stderr,"Job key: ");
        scanf("%d", &tmp);
        jobKey = tmp;
        fprintf(stderr,"New priority: ");
        scanf("%d", &tmp);
        newPriority = tmp;
        change_job_priority(geninfo->conn, jobKey, newPriority);
        fprintf(stderr,"Changed!");
        break;
      case 3:
        fprintf(stderr,"Slave Key: ");
        scanf("%d", &tmp);
        slaveKey = tmp;
        fprintf(stderr,"Job Key: ");
        scanf("%d", &tmp);
        jobKey = tmp;
        fprintf(stderr,"Frame Number: ");
        scanf("%d", &tmp);
        frameNum = tmp;
        finish_frame(geninfo->conn, slaveKey, jobKey, frameNum);
        fprintf(stderr,"Finished Frame!\n");
        break;
      case 4:
        fprintf(stderr,"Slave Key: ");
        scanf("%d", &tmp);
        slaveKey = tmp;
        fprintf(stderr,"Job Key: ");
        scanf("%d", &tmp);
        jobKey = tmp;
        fprintf(stderr,"Frame Number: ");
        scanf("%d", &tmp);
        frameNum = tmp;
        start_frame(geninfo->conn, slaveKey, jobKey, frameNum);
        fprintf(stderr,"Started Frame!\n");
        break;
      case 5:
        while(1)
+	 {
 	   multiio_poll(multiio);
+	   multiio_poll(multiio, 15000);
 	   tabletennis_serve(geninfo->config->listens->tabletennis);
+	 }
        break;
      case 0:
        test = 0;
        break;
      default:
        fprintf(stderr, "Invalid input, please try again.\n");
        break;
+     }
+   }
   return 0;
+}

src/server/listen.c

➞

Show inline comments

 /*
   Copyright 2010 Nathan Phillip Brink
   This file is a part of DistRen.
   DistRen is free software: you can redistribute it and/or modify
   it under the terms of the GNU Affero General Public License as published by
   the Free Software Foundation, either version 3 of the License, or
   (at your option) any later version.
   DistRen 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 Affero General Public License for more details.
   You should have received a copy of the GNU Affero General Public License
   along with DistRen.  If not, see <http://www.gnu.org/licenses/>.
 */
 #include "listen.h"
 #include "common/protocol.h"
 #include "common/remoteio.h"
 #include <errno.h>
 #include <list.h>
 #include <malloc.h>
 #include <netinet/in.h>
 #include <stdio.h>
 #include <string.h>
 #include <sys/types.h>
 #include <poll.h>
 #include <sys/socket.h>
 #include <unistd.h>
 /* local */
 struct distrend_request_handler_info
+{
   enum distren_request_type request_type;
   distrend_handle_request_func_t handler;
 };
 struct distrend_client *distrend_client_new(struct distrend_listens *listens,
 					    enum distrend_client_state state,
 					    struct remoteio *rem);
 int distrend_client_free(struct distrend_client *client);
 int distrend_dispatch_request(struct distrend_listens *listens, struct remoteio *rem, struct distrend_client *client, struct distren_request *req, void *reqdata);
 struct distrend_polled_sock *distrend_polled_sock_get_by_offset(struct distrend_listens *listens, size_t pollfds_offset);
 static size_t distrend_listen_read_handle(struct remoteio *rem, struct distrend_listens *listens, void *buf, size_t len, struct distrend_client *client);
 static void distrend_listen_remoteio_handle_close(struct distrend_listens *listens, struct distrend_client *client);
 int listen_handle_accept(multiio_context_t multiio,
 			 int fd,
 			 short revent,
 			 struct distrend_listens *listens,
 			 int *port);
 int listen_handle_error(multiio_context_t multiio,
 			int fd,
 			short revent,
 			struct distrend_listens *listens,
 			int *port);
 /*** TO BE MOVED TO REMOTEIO */
 int listen_handle_read(multiio_context_t multiio,
 		       int fd,
 		       short revent,
 		       struct distrend_listens *listens,
 		       struct distrend_client *client);
 int listen_handle_existence(multiio_context_t multiio,
 			    int fd,
 			    short revent,
 			    struct distrend_listens *listens,
 			    struct distrend_client *client);
 struct distrend_listens *distrend_listens_new(multiio_context_t multiio,
 					      struct general_info *geninfo,
 					      struct options_common *opts)
+{
   struct distrend_listens *listens;
   listens = malloc(sizeof(struct distrend_listens));
   if(!listens)
     return NULL;
   listens->request_handlers = list_init();
   if(!listens->request_handlers)
+    {
       free(listens);
       return NULL;
+    }
   listens->options = opts;
   listens->geninfo = geninfo;
   /* multiio */
   listens->multiio = multiio;
   /* tabletennis */
   listens->tabletennis = tabletennis_new(listens, 32, 16);
   /* This type is used for accepting connections with accept() */
   multiio_socket_type_register(multiio, &listens->socket_type);
   multiio_event_handler_register(multiio,
 				 listens->socket_type,
 				 POLLERR | POLLHUP | POLLNVAL,
 				 (multiio_event_handler_func_t)&listen_handle_error,
 				 listens);
   multiio_event_handler_register(multiio,
 				 listens->socket_type,
 				 POLLRDNORM,
 				 (multiio_event_handler_func_t)&listen_handle_accept,
 				 listens);
   return listens;
+}
 int distrend_listen_add(struct distrend_listens *listens, int port)
+{
   int tmp;
   int fd;
   int *saved_port;
   struct sockaddr_in6 sockaddr =
+    {
       .sin6_family = AF_INET6,
       .sin6_port = 0,
       .sin6_flowinfo = 0,
       .sin6_addr = IN6ADDR_ANY_INIT,
       .sin6_scope_id = 0
     };
   sockaddr.sin6_port = htons(port);
   fd = socket(AF_INET6, SOCK_STREAM, 0);
   tmp = bind(fd, (struct sockaddr *)&sockaddr, sizeof(sockaddr));
   if(tmp == -1)
+    {
       perror("bind");
       close(fd);
       return 1;
+    }
   tmp = listen(fd, 1);
   if(tmp == -1)
+    {
       perror("listen");
       close(fd);
       return 1;
+    }
   saved_port = malloc(sizeof(int));
   if(!saved_port)
+    {
       perror("malloc");
       close(fd);
       return 1;
+    }
   *saved_port = port;
   tmp = multiio_socket_add(listens->multiio, fd, listens->socket_type, saved_port, POLLRDNORM);
   if(tmp)
+    {
       close(fd);
       return 1;
+    }
   /**
    * @todo perhaps we'll someday want the ability to unlisten()? :-p
    * Then we have to store the fd somewheres so that we can call
    * multiio_socket_del() on it. So far, however, that functionality
    * isn't needed.
    */
   return 0;
+}
 int listen_handle_error(multiio_context_t multiio,
 			int fd,
 			short revent,
 			struct distrend_listens *listens,
 			int *port)
+{
   fprintf(stderr, "Port %d experienced an error or is closed. Closing it.\n", *port);
   multiio_socket_del(listens->multiio, fd);
   close(fd);
   free(port);
   return 0;
+}
 //int listen_handle_read(struct distrend_listens *listens,
 //			struct distrend_client *client)
 /**
  * an important thing to handle
+ *
  * @deprecated to be replaced with table_tennis
  */
 int listen_handle_existence(multiio_context_t multiio,
 			    int fd,
 			    short revent,
 			    struct distrend_listens *listens,
 			    struct distrend_client *client)
+{
   /**
    * handle dead/closed sockets here!
    */
   fprintf(stderr, __FILE__ ":%d: handle dead/closed sockets here!\n", __LINE__);
   /**
    * Manage timing-out clients.
    */
   if(time(NULL) > client->cleanup_time)
     switch(client->state)
+      {
       case DISTREND_CLIENT_PREVERSION:
 	distrend_send_disconnect(client, "You have failed to present version information in a timely manner. Cya :-p");
 	break;
       case DISTREND_CLIENT_PREAUTH:
 	distrend_send_disconnect(client, "You have failed to present authentication information in a timely manner. Cya ;-)");
 	break;
       case DISTREND_CLIENT_GOOD:
 	/*
 	 * pings should be managed with two queues sharing this struct:
 	 *   struct pingpong_queue { struct distrend_client *client; time_t ping_time };
+	 *
 	 * - queue to_ping: contains queue of clients to send pings to
 	 *   with the client whose ping time is earliest at the front
+	 *
 	 * - queue to_be_ponged: contains a queue of clients to clean up
 	 *   if they haven't recieved pongs. client with earliest cleanup
 	 *   time is at the front. If a PONG packet is received in time, the
 	 *   cleanup_time is bumped but the queue is left alone. When the
 	 *   queue's element is encountered, cleanup_time is checked and then
 	 *   the client is readded to the to_ping queue.
+	 *
 	 * Each queue shall be eaten as time passes and continue forever
 	 * in circularity.
+	 *
 	 * data structures.
 	 * fun.
 	 * impossible to explain in text.
 	 * easy to think up.
 	 * impossible to determine the workings of existing ones.
 	 * and... when you need them, screenshot utilities just aren't available :-/.
 	 */
 	distrend_send_disconnect(client, "Ping timeout :-p");
 	break;
       case DISTREND_CLIENT_BAD:
 	fprintf(stderr, __FILE__ ":%d: aaarrrrgh!\n :-D\n", __LINE__);
 	break;
       default:
 	break;
+      }
   return 0;
+}
 struct distrend_accept_client_proc_data
+{
   struct distrend_listens *listens;
   time_t current_time;
 };
 /**
  * Handle new connections.
  */
 int listen_handle_accept(multiio_context_t multiio,
 			 int fd,
 			 short revent,
 			 struct distrend_listens *listens,
 			 int *port)
    //int distrend_accept(struct distrend_listens *listens)//, struct distrend_clientset *clients, distrend_handle_request_t handlereq, void *handlereqdata)
+ {
    struct distrend_client *newclient;
    int newclientsock;
    struct remoteio *rem;
    newclientsock = accept(fd, (struct sockaddr *)NULL, (socklen_t *)NULL);
    /* used to call int distrend_client_add(struct distrend_listens *listens, int sock, DISTREND_CLIENT_PREVERSION)*/
    newclient = distrend_client_new(listens, DISTREND_CLIENT_PREVERSION, NULL);
    if(remoteio_open_socket(&rem, listens->options->remoteio, (remoteio_read_handle_func_t)&distrend_listen_read_handle, newclient, (remoteio_close_handle_func_t)&distrend_listen_remoteio_handle_close, newclientsock))
+     {
        fprintf(stderr, "error allocating/adding client struct\n");
        return 1;
+     }
    newclient->rem = rem;
    fprintf(stderr, "accepted new connection; fd=%d\n", newclientsock);
    /* tabletennis */
    tabletennis_add_client(listens->tabletennis, newclient);
    /**
     * For those using netcat/telnet to debug their internets.
     */
  #ifndef PACKAGE_URL
  #define PACKAGE_URL "http://ohnopub.net/distren/"
  #endif
  #define DISTREN_GREETING PACKAGE_STRING " " PACKAGE_URL " : Nathan Phillip Brink && Ethan Michael Zonca\n"
    /* using sizeof() - 1 because the sizeof() includes a NULL byte we want to ignore. */
    remoteio_write(newclient->rem, DISTREN_GREETING, sizeof(DISTREN_GREETING) - 1);
    /*
    list_mvfront(listens->clients);
    newclient = list_curr(listens->clients);
    while(newclient)
+     {
        if(newclient->state == DISTREND_CLIENT_DEAD)
+	 {
 	   distrend_listen_poll_deletefd(listens, &newclient->sock);
 	   distrend_client_free(newclient);
 	   list_remove_curr(listens->clients);
 	   fprintf(stderr, "removed dead connection\n");
+	 }
        list_mvnext(listens->clients);
    *//* provide for termination of this loop *//*
        if(newclient == list_curr(listens->clients))
 	 newclient = NULL;
        else
 	 newclient = list_curr(listens->clients);
+     }
 					       */
    return 0;
+ }
  /**
   * Handle read events from remoteio, remoteio_read_handle_func_t.
+  *
   * This func requires that someone called remoteio_generic_data_set(remoteio_opts, listens);
+  *
   * @param client the client associated with this remoteio instance.
   */
 size_t distrend_listen_read_handle(struct remoteio *rem, struct distrend_listens *listens, void *buf, size_t len, struct distrend_client *client)
+ {
    struct distren_request *req;
    void *reqdata;
    size_t used_len;
    used_len = 0;
    /**
     * Manage input, etc.
     */
    if(client->expectlen == 0)
+     {
        /* search out header from input so far */
        if(len > sizeof(struct distren_request))
+	 {
 	   if(distren_request_new_fromdata(&req, buf, len))
+	     {
 	       fprintf(stderr, "Error handling data from client (magic likely did not match), closing connection\n");
 	       /*
 		* yes, this is safe and legal... because of hackishness
 		* in remoteio_close() ;-)
 		*/
 	      remoteio_close(rem);
 	      return 1;
+	    }
 	  client->expectlen = req->len + sizeof(struct distren_request);
 	  distren_request_free(req);
+	}
+    }
   if(client->expectlen
      && len >= client->expectlen)
+    {
       if(distren_request_new_fromdata(&req, buf, len))
+	{
 	  if(client->state == DISTREND_CLIENT_PREAUTH)
 	    remoteio_close(rem);
 	  else
 	    distrend_send_disconnect(client, "Protocol error.");
 	  return 1;
+	}
       /*
        * this really shouldn't happen... reparsing the same data with
        * distren_request_new_fromdata() a second time shouldn't yeild
        * a different req->len than it did before.
        */
       if(len - sizeof(struct distren_request) < req->len)
+	{
 	  fprintf(stderr, "Unexpected error handling some data from client\n");
 	  distren_request_free(req);
 	  /* but we should pay homage to W3C if the impossible happens */
 	  distrend_send_disconnect(client, "HTTP/1.1 503 Internal Server Error");
 	  return 1;
+	}
       reqdata = malloc(req->len);
       if(!reqdata)
+	{
 	  fprintf(stderr, "OOM\n");
 	  distren_request_free(req);
 	  return 1;
+	}
       memcpy(reqdata, ((void *)buf) + sizeof(struct distren_request), req->len);
       client->expectlen = 0;
       used_len = sizeof(struct distren_request) + req->len;
       distrend_dispatch_request(listens, rem, client, req, reqdata);
       free(reqdata);
       distren_request_free(req);
       /* I actually just used recursion in non-LISP code! :-D */
-      return req->len + distrend_listen_read_handle(rem, listens, buf + req->len, len - req->len, client);
+      return used_len + distrend_listen_read_handle(rem, listens, buf + req->len, len - req->len, client);
+    }
-  return 0;
+  return used_len;
+}
 /**
  * Handle cleaning up after remoteio_close() has been called. This includes cleaning up the struct distrend_client and stuffs
  */
 void distrend_listen_remoteio_handle_close(struct distrend_listens *listens, struct distrend_client *client)
+{
   /*
    * remoteio handles removing itself from multiio for us. We just
    * have to clean up tabletennis and the struct itself.
    */
   tabletennis_del_client(listens->tabletennis, client);
   free(client);
+}
 int distrend_listen_free(struct distrend_listens *listens)
+{
   tabletennis_free(listens->tabletennis);
   fprintf(stderr, "%s:%d: I am a stub that needn't be implemented 'til later\n", __FILE__, __LINE__);
   return 1;
+}
 /**
    This is probably just NOT a placeholder for remotio
 */
 void remotio_send_to_client(struct distrend_client *client, const char *msg, size_t len)
+{
     fprintf(stderr, "%s:%d: STUB I should queue data for writing to a client.... or should I? :-p\n", __FILE__, __LINE__);
+}
 /**
  * Allocates and initializes a struct distrend_client.
+ *
+ *
  */
 struct distrend_client *distrend_client_new(struct distrend_listens *listens, enum distrend_client_state state, struct remoteio *rem)
+{
   struct distrend_client *client;
   client = malloc(sizeof(struct distrend_client));
   if(!client)
+    {
       fprintf(stderr, "OOM\n");
       return NULL;
+    }
   client->state = state;
   client->cleanup_time = time(NULL) + DISTREND_LISTEN_AUTHTIME;
   client->inlen = 0;
   client->expectlen = 0;
   client->rem = rem;
   return client;
+}
 int distrend_client_write_request(struct distrend_client *client, const struct distren_request *req, const void *data)
+{
   char *towrite;
   int ret;
   size_t msglen;
   msglen = sizeof(struct distren_request) + req->len;
   towrite = malloc(msglen);
   if(!towrite)
+    {
       fprintf(stderr, "OOM\n");
       return 1;
+    }
   memcpy(towrite, req, sizeof(struct distren_request));
   memcpy(towrite + sizeof(struct distren_request), data, req->len);
   ret = remoteio_write(client->rem, towrite, msglen);
   free(towrite);
   return ret;
+}
 /*
 int distrend_client_read(struct distrend_client *client, char **toread, size_t *lenread)
+{
   struct distrend_packet *packet;
   *lenread = 0;
   *toread = NULL;
   if(q_empty(client->inmsgs))
     return 1;
   packet = q_dequeue(client->inmsgs);
   *lenread = packet->len;
   *toread = packet->data;
   free(packet);
   return 0;
+}
 */
 int distrend_send_disconnect(struct distrend_client *client, const char *quit_msg)
+{
   struct distren_request *req;
   distren_request_new(&req, strlen(quit_msg), DISTREN_REQUEST_DISCONNECT);
   distrend_client_write_request(client, req, quit_msg);
   distren_request_free(req);
   client->state = DISTREND_CLIENT_BAD;
   client->cleanup_time = time(NULL) + DISTREND_LISTEN_DISCONNECT_GRACE;
   return 0;
+}
 int distrend_listen_handler_add(struct distrend_listens *listens, enum distren_request_type type, distrend_handle_request_func_t handler)
+{
   struct distrend_request_handler_info *handler_info;
   handler_info = malloc(sizeof(struct distrend_request_handler_info));
   if(!handler_info)
     return 1;
   handler_info->request_type = type;
   handler_info->handler = handler;
   list_insert_after(listens->request_handlers, handler_info, 0);
   return 0;
+}
 struct distrend_dispatch_request_data
+{
   struct general_info *geninfo;
   struct distrend_client *client;
   struct distren_request *req;
   void *req_data;
 };
 /**
    traversal function for distrend_dispatch_request().
  */
 int _distrend_dispatch_request_trav(struct distrend_dispatch_request_data *data, struct distrend_request_handler_info *handler_info)
+{
   if(handler_info->request_type == data->req->type)
     (*handler_info->handler)(data->geninfo, data->client, data->req->len, data->req_data);
   return TRUE;
+}
 /**
-   helper for listen_handle_read() which looks up the correct
+   helper for distrend_listen_read_handle() which looks up the correct
    request handler and handles handing the the request to the
    handler. :-p
- */
 */
 int distrend_dispatch_request(struct distrend_listens *listens, struct remoteio *rem, struct distrend_client *client, struct distren_request *req, void *reqdata)
+{
   struct distrend_dispatch_request_data data;
   data.geninfo = listens->geninfo;
   data.client = client;
   data.req = req;
   data.req_data = reqdata;
   list_traverse(listens->request_handlers, &data, (list_traverse_func_t)&_distrend_dispatch_request_trav, LIST_FRNT | LIST_SAVE);
   return 0;
+}

src/server/slave.c

➞

Show inline comments

@@ @@ -137,363 +137,365 @@ int main(int argc, char *argv[]) @@
       return 1;
+    }
   if(!server)
+    {
       fprintf(stderr, "server not set\n");
       return 1;
+    }
   if(!username)
+    {
       fprintf(stderr, "username not set\n");
       return 1;
+    }
   if(!password)
+    {
       fprintf(stderr, "password not set\n");
       return 1;
+    }
   if(!hostname)
+    {
       fprintf(stderr, "hostname not set\n");
       return 1;
+    }
   /* Notifies the user if there no username in .conf */
   if(checkUsername(username))
     return 1;
   if(!strncmp(password, "!password",10))
+    {
       fprintf(stderr, "You haven't specified a password. Please edit distrenslave.conf!\n");
       return 1;
+    }
   // Variables needed for main loop
   int jobnum = 0;
   int framenum = 0;
   int slavekey = atoi(username); // @TODO: Make this more friendly
   char *urltoTar;      /* Full URL to the server-side location of job#.tgz */
   char *pathtoTar;     /* Full path to the location of the job#.tgz */
   char *pathtoTardir;
   char *urltoOutput;   /* Full URL where output is posted */
   char *pathtoOutput;  /* Full path to the output (rendered) file */
   char *pathtoOutdir;  /* Full path to output directory */
   char *pathtoRenderOutput;  /* Contains blender framenum placeholder */
   char *urltoJobfile; /* No longer used, url to .blend on server */
   char *pathtoJob;     /* Full path to job data folder */
   char *pathtoJobfile; /* Full path to the job's main file */
   char *outputExt = "jpg";     /* Output Extension (e.g., JPG) */
   int haveWork = 0;
   int benchmarkTime = 0;
   int renderPower = 0;
   slave_state.copyright_done = 0;
   slave_state.expectlen = 0;
   slave_state.quit = 0;
   fprintf(stderr, "Connecting to server...\n");
   if(remoteio_open_server(&slave_state.rem, commonopts->remoteio, &distren_slave_remoteio_read_handle, &slave_state, &distren_slave_remoteio_close_handle, server))
+    {
       fprintf(stderr, "Error connecting to server; exiting\n");
       return 1;
+    }
   greet_server(slave_state.rem);
   fprintf(stderr,"\nDistRen Slave Pre-Alpha %s\n- Experimental build: Use at your own risk!\n\n", PACKAGE_VERSION);
   // @TODO: add call to function to force recalc if $render_power == ""
   if(runBenchmark)
+    {
       if(slaveBenchmark(datadir, &benchmarkTime, &renderPower))
+        {
           fprintf(stderr,"Benchmark failed! Exiting.\n");
           return 1;
+        }
       else
+        {
           fprintf(stderr,"Benchmark successful, time taken was %d seconds, giving you a render power of %d.\n",
                   benchmarkTime, renderPower);
           _web_setrenderpower(slavekey, password, renderPower);
           return 0;
+        }
+    }
   if(!DEBUG)
     fprintf(stderr, "Running..");
   // Main loop
   while(!slave_state.quit)
+    {
       multiio_poll(multiio);
+      multiio_poll(multiio, 15000);
       if(slave_state.quit)
 	break;
     // request work
     fprintf(stderr, "Waiting...\n");
     haveWork = getwork(slave_state.rem, &jobnum, &framenum);
     /* If we got a frame */
     if(haveWork)
+      {
         fprintf(stderr,"Got work from server...\n");
         /* @TODO: Add remotio hooks */
         // jobnum = remoteio_read(jobnum); /* Set jobnum from remoteio (we could use info from struct, but we need this info to download the xmlfile */
         // framenum = remoteio_read(jobnum); /* Set framenum from remoteio */
         // outputExt = remotio)read(outputExt); /* Set output extension from remotio */
         if(DEBUG)
           fprintf(stderr, "Preparing to render frame %d in job %d\n", framenum, jobnum);
         prepareJobPaths(jobnum, framenum, outputExt, datadir, &urltoTar, &pathtoTar,&pathtoTardir,&pathtoJob, &pathtoJobfile, &urltoJobfile, &urltoOutput, &pathtoOutput, &pathtoRenderOutput, &pathtoOutdir);
         int dlret = downloadTar(urltoTar, pathtoTar);
         if(dlret == 0)
           fprintf(stderr,"Data retrieved successfully!\n");
         else if(dlret == 3){
           resetframe(slave_state.rem, jobnum, framenum);  // Unassign the frame on the server so other slaves can render it
           return 0; // ouput dir doesn't exist
+        }
         else
           if(DEBUG)
             fprintf(stderr,"Using existing tarball %s...\n", pathtoTar);
         // Decompress tarball
         struct stat jbuffer;
         int jstatus = stat(pathtoTar, &jbuffer);
         if(jstatus == -1){
           if(DEBUG)
             fprintf(stderr,"Main job file does not exist, extracting...\n");
           // If error unpacking tarball
           if(unpackJob(pathtoJob, pathtoTar)){
             resetframe(slave_state.rem, jobnum, framenum);  // Unassign the frame on the server so other slaves can render it
             fprintf(stderr,"Error decompressing tarball! Exiting.\n");
             return 1;
+          }
+        }
         /* ignore return because directory may exist already */
         if(DEBUG)
           fprintf(stderr,"Creating output directory %s\n", pathtoOutdir);
         mkdir(pathtoOutdir, 0700);
         if(DEBUG)
           fprintf(stderr,"Marking frame started on server... ");
         _web_startframe(slavekey, password, jobnum, framenum);
         /* Execute blender */
         if(DEBUG){
           fprintf(stderr,"Executing blender on file %s\n", pathtoJobfile);
           fprintf(stderr,"Directing output to file %s\n", pathtoOutput);
+        }
         /* Execute blender */
         if(exec_blender(pathtoJobfile, pathtoOutput, framenum))
+          {
             fprintf(stderr,"Error running Blender. Check your installation and/or your PATH.\n");
             resetframe(slave_state.rem, jobnum, framenum);  // Unassign the frame on the server so other slaves can render it
             return 1;
+          }
         free(pathtoJobfile);
         pathtoJobfile = NULL;
         struct stat buffer;
         int fstatus = stat(pathtoOutput, &buffer);
         if(fstatus == -1){
           fprintf(stderr,"*** %s doesn't exist! Scene may not have camera, or your blender installation is not working.\n", pathtoOutput);
           resetframe(slave_state.rem, jobnum, framenum);  // Unassign the frame on the server so other slaves can render it
           return 1;
+        }
         else{
           /* Post-execution */
           if(DEBUG)
             fprintf(stderr, "Finished frame %d in job %d, uploading...\n", framenum, jobnum);
           else
             fprintf(stderr,"Finished frame.\n");
           uploadOutput(pathtoOutput, urltoOutput, jobnum, framenum, slavekey); // @TODO: Handle return value
           free(urltoOutput);
           free(pathtoOutput);
           urltoOutput = NULL;
           pathtoOutput = NULL;
           // Tell the server that rendering and upload are complete of "jobjum.framenum"
           finishframe(slave_state.rem, jobnum, framenum);
   free(urltoTar);
   free(pathtoTar);
   free(pathtoTardir);
   free(pathtoJob);
   free(pathtoJobfile);
   free(urltoJobfile);
   free(urltoOutput);
   free(pathtoRenderOutput);
   free(pathtoOutdir);
+        }
+     }
     else{
       if(DEBUG)
         fprintf(stderr,"Nothing to do. Idling...\n");
       else
         fprintf(stderr,".");
       /**
 	 to prevent infinite loops from burning CPU, we just sleep(1) ;-)
       */
       sleep(1);
+    }
     /* @TODO: If the server says that every frame for the last jobnum is finished, OR if the data is getting old */
     if(1 == 0)
+      {
         // Note: individual frames are already deleted after uploading,
         // except for ones that couldn't be uploaded
         delete_jobdata(jobnum, datadir);
+      }
     sleep(5); // Poll 5 seconds. @TODO: Remove all polling
+  }
   fprintf(stderr, "Quitting...\n");
   free(my_cfg);
   free(datadir);
   fprintf(stderr, "Goodbye!\n");
   return 0;
+}
 static void distren_slave_remoteio_close_handle(void *blah, void *data)
+{
   struct slave_state *slave_state = (struct slave_state *)data;
   fprintf(stderr, "Lost connection to server\n");
   slave_state->quit = 1;
+}
 static size_t distren_slave_remoteio_read_handle(struct remoteio *rem,
 						 void *blah,
 						 void *buf,
 						 size_t len,
 						 void *data)
+{
   struct slave_state *slave_state = (struct slave_state *)data;
   struct distren_request *req, *my_req;
   void *req_data, *my_req_data;
   size_t to_return;
   size_t counter;
   fprintf(stderr, "expected to eat %d bytes\n", len);
   /* to_return shall record how much of buf we've eaten already */
   to_return = 0;
   if(!slave_state->copyright_done)
+    {
       putchar('\n');
       /* we have to flush through data until we hit a newline */
       for(counter = 0;
 	  counter + slave_state->expectlen < 256 && counter < len && ((char *)buf)[counter] != '\n';
 	  counter ++)
+	{
 	  putchar(((char *)buf)[counter]);
+	}
       slave_state->expectlen += counter - 1;
       if(slave_state->expectlen == 256)
+	{
 	  fprintf(stderr, "\nThe server's greeting is too long. Maybe it speaks a foreign language\n");
 	  slave_state->quit = 1;
+	}
       if(counter < len && ((char *)buf)[counter] == '\n')
+	{
 	  putchar('\n');
 	  counter ++;
 	  slave_state->expectlen = 0;
 	  slave_state->copyright_done = 1;
 	  to_return += counter;
 	  buf += counter;
 	  len -= counter;
+	}
       if(!slave_state->copyright_done)
 	return to_return;
+    }
   while(1)
+    {
       /* if we haven't read a full header yet: */
       if(!slave_state->expectlen)
+	{
 	  if(len < sizeof(struct distren_request))
-	    return 0;
+	    return to_return;
 	  /* figure out how much we need to read in before we can get anywhere */
 	  if(distren_request_new_fromdata(&req, buf, len))
+	    {
 	      fprintf(stderr, "Failing to interpret data from server, exiting\n");
 	      slave_state->quit = 1;
-	      return 0;
+	      return to_return;
+	    }
 	  slave_state->expectlen = sizeof(struct distren_request) + req->len;
 	  distren_request_free(req);
+	}
       if(slave_state->expectlen
 	    && slave_state->expectlen <= len)
+	{
 	  distren_request_new_fromdata(&req, buf, len);
 	  req_data = buf + sizeof(struct distren_request);
 	  switch((enum distren_request_type)req->type)
+	    {
 	    case DISTREN_REQUEST_VERSION:
 	      fprintf(stderr, "The server runs ");
 	      for(counter = 0; counter < req->len; counter ++)
 		putc(((char *)req_data)[counter], stderr);
 	      putc('\n', stderr);
 	      break;
 	    case DISTREN_REQUEST_PING:
 	      fprintf(stderr, "PONG ! :-D\n");
 	      distren_request_poing(&my_req, &my_req_data, 0, req_data, req->len);
 	      remoteio_write(slave_state->rem, &my_req, sizeof(struct distren_request));
 	      remoteio_write(slave_state->rem, &my_req_data, req->len);
 	      remoteio_write(slave_state->rem, my_req, sizeof(struct distren_request));
 	      remoteio_write(slave_state->rem, my_req_data, req->len);
 	      distren_request_free_with_data(my_req, my_req_data);
 	      break;
 	    case DISTREN_REQUEST_DISCONNECT:
 	      /* hopefully this ends up being a useful message... */
 	      printf("You have been disconnected: \"");
 	      for(counter = 0; counter < req->len; counter ++)
 		putchar(((char *)buf)[counter]);
 	      putchar('"');
 	      putchar('\n');
 	      break;
 	    default:
 	      fprintf(stderr, "something\n");
 	      break;
+	    }
 	  counter = req->len + sizeof(struct distren_request);
 	  distren_request_free(req);
 	  slave_state->expectlen = 0;
 	  counter = req->len + sizeof(struct distren_request);
 	  len -= counter;
 	  buf += counter;
 	  to_return += counter;
+	}
+    }
   fprintf(stderr, "ate %d bytes\n", to_return);
   return to_return;
+}

src/server/tabletennis.c

➞

Show inline comments

 /*
  * Copyright 2010 Nathan Phillip Brink
+ *
  * This file is a part of DistRen.
+ *
  * DistRen is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published by
  * the Free Software Foundation, either version 3 of the License, or
  * (at your option) any later version.
+ *
  * DistRen 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 Affero General Public License for more details.
+ *
  * You should have received a copy of the GNU Affero General Public License
  * along with DistRen.  If not, see <http://www.gnu.org/licenses/>.
  */
 #include "distrend.h"
 #include "listen.h"
 #include "tabletennis.h"
 #include "common/request.h"
 #include "common/protocol.h"
 #include <queue.h>
 #include <stdlib.h>
 #include <string.h>
 #include <time.h>
 struct tabletennis
+{
   unsigned int ping_interval;
   unsigned int pong_time;
   /* of type (struct distrend_client *) */
   queue_t clients_to_ping;
   /* of type (struct distrend_client *) */
   queue_t clients_need_pong;
   struct timespec time_last_check;
 };
 static int tabletennis_pong_request_handle(struct general_info *geninfo, struct distrend_client *client, size_t req_len, void *req_data);
 static int tabletennis_ping_request_handle(struct general_info *geninfo, struct distrend_client *client, size_t req_len, void *req_data);
 tabletennis_t tabletennis_new(struct distrend_listens *listens, unsigned int ping_interval, unsigned int pong_time)
+{
   tabletennis_t tabletennis;
   tabletennis = malloc(sizeof(struct tabletennis));
   tabletennis->ping_interval = ping_interval;
   tabletennis->pong_time = pong_time;
   tabletennis->clients_to_ping = q_init();
   tabletennis->clients_need_pong = q_init();
   clock_gettime(CLOCK_MONOTONIC, &tabletennis->time_last_check);
   distrend_listen_handler_add(listens, DISTREN_REQUEST_PING, &tabletennis_ping_request_handle);
   distrend_listen_handler_add(listens, DISTREN_REQUEST_PONG, &tabletennis_pong_request_handle);
   return tabletennis;
+}
 int tabletennis_add_client(tabletennis_t tabletennis, struct distrend_client *client)
+{
   client->tabletennis_client.state = TABLETENNIS_NEED_PING;
   client->tabletennis_client.time_next_check = tabletennis->time_last_check.tv_sec
     + tabletennis->ping_interval;
   q_enqueue(tabletennis->clients_to_ping, client, 0);
   return 0;
+}
 int tabletennis_serve(tabletennis_t tabletennis)
+{
   struct timespec time_now;
   struct distrend_client *client;
   time_t time_next_check;
   struct distren_request *req;
   void *req_data;
   size_t req_len;
   clock_gettime(CLOCK_MONOTONIC, &time_now);
   time_next_check = time_now.tv_sec + tabletennis->pong_time;
   for(client = q_front(tabletennis->clients_to_ping);
       client && client->tabletennis_client.time_next_check < time_now.tv_sec;
       client = q_front(tabletennis->clients_to_ping))
+    {
       q_dequeue(tabletennis->clients_to_ping);
       /* use time_next_check as the ping data */
-      req_len = distren_request_poing(&req, &req_data, 1, &time_next_check, sizeof(time_next_check));
       distren_request_poing(&req, &req_data, 1, &time_next_check, sizeof(time_next_check));
       distrend_client_write_request(client, req, req_data);
       distren_request_free_with_data(req, req_data);
       client->tabletennis_client.state = TABLETENNIS_NEED_PONG;
       client->tabletennis_client.time_next_check = time_next_check;
       q_enqueue(tabletennis->clients_need_pong, client, 0);
+    }
   time_next_check = time_now.tv_sec + tabletennis->ping_interval;
   for(client = q_front(tabletennis->clients_need_pong);
       client && client->tabletennis_client.time_next_check < time_now.tv_sec;
       client = q_front(tabletennis->clients_need_pong))
+    {
       q_dequeue(tabletennis->clients_need_pong);
       if(client->tabletennis_client.state == TABLETENNIS_NEED_PONG)
+	{
 	  distrend_send_disconnect(client, "You failed to respond to a PING packet after some seconds!");
 	  client->tabletennis_client.state = TABLETENNIS_DELINQUENT;
+	}
       else /* state must be TABLETENNIS_HAS_PONG */
+	{
 	  client->tabletennis_client.time_next_check = time_next_check;
 	  client->tabletennis_client.state = TABLETENNIS_NEED_PING;
 	  q_enqueue(tabletennis->clients_to_ping, client, 0);
+	}
+    }
   /* for tabletennis_add_client() */
   memcpy(&tabletennis->time_last_check, &time_now, sizeof(struct timespec));
   return 0;
+}
 /**
  * helper for tabletennis_del_client() which just looks for a
  * particular pointer in the queue (a.k.a., list).
  */
 static int tabletennis_del_client_traverse(void *trav_data, void *client)
+{
   if(trav_data == client)
     return FALSE;
   return TRUE;
+}
 int tabletennis_del_client(tabletennis_t tabletennis, struct distrend_client *client)
+{
   list_t thelist;
   thelist = NULL;
   switch(client->tabletennis_client.state)
+    {
     case TABLETENNIS_NEED_PING:
       thelist = tabletennis->clients_to_ping;
       break;
     case TABLETENNIS_NEED_PONG:
     case TABLETENNIS_HAS_PONG:
       thelist = tabletennis->clients_need_pong;
       break;
     case TABLETENNIS_DELINQUENT:
       thelist = NULL;
       break;
+    }
   if(!thelist)
     return 0;
   list_traverse(thelist, (void *)client, (list_traverse_func_t)&tabletennis_del_client_traverse,
 		LIST_FORW | LIST_FRNT | LIST_SAVE);
   if(list_curr(thelist) == client)
     list_remove_curr(thelist);
   return 0;
+}
 void tabletennis_free(tabletennis_t tabletennis)
+{
   q_free(tabletennis->clients_to_ping, LIST_NODEALLOC);
   q_free(tabletennis->clients_need_pong, LIST_NODEALLOC);
   free(tabletennis);
+}
 /* implementations of locals */
 /**
  * Handles a DISTREN_REQUEST_PING from a client.
+ *
  * @todo throttling?
  */

0 comments (0 inline, 0 general)