network throughput

svn path=/trunk/boinc/; revision=929

client/client_state.C

@@ -375,10 +375,14 @@ static void print_log(char* p) {
     }
 }
 
-// do_something is where all the action happens.
-// Each of the client's finite-state machine layers is polled,
+int CLIENT_STATE::net_sleep(double x) {
+    return net_xfers->net_sleep(x);
+}
+
+// do_something polls each of the client's finite-state machine layers,
 // possibly triggering state transitions.
 // Returns true if something happened
+// (in which case should call this again immediately)
 //
 bool CLIENT_STATE::do_something() {
     int nbytes=0;
@@ -390,14 +394,19 @@ bool CLIENT_STATE::do_something() {
 
     print_log("Polling; active layers:\n");
     if (activities_suspended) {
-    print_log("None (suspended)\n");
+        print_log("None (suspended)\n");
     } else {
         // Call these functions in bottom to top order with
         // respect to the FSM hierarchy
 
-        if (max_bytes > 0)
+        if (max_bytes > 0) {
             net_xfers->poll(max_bytes, nbytes);
-        if (nbytes) { max_bytes -= nbytes; action=true; print_log("net_xfers\n"); }
+        }
+        if (nbytes) {
+            max_bytes -= nbytes;
+            action=true;
+            print_log("net_xfers\n");
+        }
 
         x = http_ops->poll();
         if (x) {action=true; print_log("http_ops\n"); }

client/client_state.h

@@ -63,6 +63,13 @@ public:
     int restart_tasks();
     int cleanup_and_exit();
     bool do_something();
+        // Initiates and completes actions (file transfers, process executions)
+        // Never blocks.
+        // Returns true if it actually did something,
+        // in which case it should be called again immediately.
+    int net_sleep(double dt);
+        // sleeps until either dt seconds have elapsed,
+        // or until there's network activity.
     void parse_cmdline(int argc, char** argv);
     void parse_env_vars();
     bool time_to_exit();

client/main.C

@@ -86,8 +86,11 @@ int main(int argc, char** argv) {
     if (retval) exit(retval);
     while (1) {
         if (!gstate.do_something()) {
-            boinc_sleep(1);
-            if (log_flags.time_debug) printf("SLEEP 1 SECOND\n");
+            double x;
+            gstate.net_sleep(1.);
+            x = 1;
+            //boinc_sleep(1);
+            if (log_flags.time_debug) printf("SLEPT %f SECONDS\n", x);
             fflush(stdout);
         }
 

client/net_xfer.C

@@ -191,8 +191,9 @@ int NET_XFER_SET::remove(NET_XFER* nxp) {
     return 1;
 }
 
-// transfer data to/from a list of active streams
-// transfer at most max_bytes bytes.
+// Transfer data to/from active streams
+// Nonblocking; keep doing I/O until would block.
+// Transfer at most max_bytes bytes.
 // TODO: implement other bandwidth constraints (ul/dl ratio, time of day)
 //
 int NET_XFER_SET::poll(int max_bytes, int& bytes_transferred) {
@@ -202,9 +203,6 @@ int NET_XFER_SET::poll(int max_bytes, int& bytes_transferred) {
     bytes_transferred = 0;
     while (1) {
 		timeout.tv_sec = timeout.tv_usec = 0;
-#ifndef _WIN32
-        timeout.tv_sec = 1;
-#endif
         retval = do_select(max_bytes, n, timeout);
         if (retval) return retval;
         if (n == 0) break;
@@ -215,9 +213,24 @@ int NET_XFER_SET::poll(int max_bytes, int& bytes_transferred) {
     return 0;
 }
 
+// Wait at most x seconds for network I/O to become possible,
+// then do some of it.
+//
+int NET_XFER_SET::net_sleep(double x) {
+    int n, retval;
+    struct timeval timeout;
+
+    timeout.tv_sec = (int)x;
+    timeout.tv_usec = (int)(1000000*(x - (int)x));
+    retval = do_select(100000000, n, timeout);
+    return retval;
+}
+
 // do a select and do I/O on as many sockets as possible.
 // 
-int NET_XFER_SET::do_select(int max_bytes, int& bytes_transferred, struct timeval timeout) {
+int NET_XFER_SET::do_select(
+    int max_bytes, int& bytes_transferred, timeval& timeout
+) {
     struct timeval zeros;
     int n, fd, retval;
     socklen_t i;
@@ -253,7 +266,8 @@ int NET_XFER_SET::do_select(int max_bytes, int& bytes_transferred, struct timeva
         }
         FD_SET(net_xfers[i]->socket, &error_fds);
     }
-    n = select(FD_SETSIZE, &read_fds, &write_fds, &error_fds, &zeros);
+    //n = select(FD_SETSIZE, &read_fds, &write_fds, &error_fds, &zeros);
+    n = select(FD_SETSIZE, &read_fds, &write_fds, &error_fds, &timeout);
     if (log_flags.net_xfer_debug) printf("select returned %d\n", n);
     if (n == 0) return 0;
     if (n < 0) return ERR_SELECT;

client/net_xfer.h

@@ -38,12 +38,13 @@ public:
     bool want_download;     // at most one should be true
     bool want_upload;
     bool do_file_io;
-    // whether poll() should transfer data to/from file
-    // (in which case "file" and blocksize are relevant)
-    // or just set io_ready
+        // whether poll() should transfer data to/from file
+        // (in which case "file" and blocksize are relevant)
+        // or just set io_ready
     bool io_done;           // got error or EOF
     FILE* file;
     bool io_ready;
+        // can read or write socket now (used if !do_file_io)
     int error;
     char hostname[256];
     int port;
@@ -63,7 +64,8 @@ public:
     int insert(NET_XFER*);
     int remove(NET_XFER*);
     int poll(int max_bytes, int& bytes_transferred);
-    int do_select(int max_bytes, int& bytes_transferred, struct timeval timeout);
+    int net_sleep(double);
+    int do_select(int max_bytes, int& bytes_transferred, timeval& timeout);
     NET_XFER* lookup_fd(int);   // lookup by fd
 };