Merge pull request #7084 from BOINC/dpa_app_network

Client, Manager, API, docker_wrapper: support network access from apps

Author: AenBleidd

api/boinc_api.cpp

@@ -190,9 +190,6 @@ static volatile int running_interrupt_count = 0;
 static volatile bool finishing;
     // used for worker/timer synch during boinc_finish();
 static int want_network = 0;
-static int have_network = 1;
-static double bytes_sent = 0;
-static double bytes_received = 0;
 bool boinc_disable_timer_thread = false;
     // simulate unresponsive app by setting to true (debugging)
 static FUNC_PTR timer_callback = 0;
@@ -205,6 +202,7 @@ int app_min_checkpoint_period = 0;
 static volatile SPORADIC_AC_STATE ac_state;
 static volatile int ac_fd, ca_fd;
 static volatile bool do_sporadic_files;
+volatile bool got_heartbeat_message = false;
 
 #define TIMER_PERIOD 0.1
     // Sleep interval for timer thread;
@@ -425,26 +423,17 @@ static bool update_app_progress(double cpu_t, double cp_cpu_t) {
 
     snprintf(msg_buf, sizeof(msg_buf),
         "<current_cpu_time>%e</current_cpu_time>\n"
-        "<checkpoint_cpu_time>%e</checkpoint_cpu_time>\n",
-        cpu_t, cp_cpu_t
+        "<checkpoint_cpu_time>%e</checkpoint_cpu_time>\n"
+        "<want_network>%d</want_network>\n",
+        cpu_t, cp_cpu_t,
+        want_network?1:0
     );
-    if (want_network) {
-        strlcat(msg_buf, "<want_network>1</want_network>\n", sizeof(msg_buf));
-    }
     if (fraction_done >= 0) {
         double range = aid.fraction_done_end - aid.fraction_done_start;
         double fdone = aid.fraction_done_start + fraction_done*range;
         snprintf(buf, sizeof(buf), "<fraction_done>%e</fraction_done>\n", fdone);
         strlcat(msg_buf, buf, sizeof(msg_buf));
     }
-    if (bytes_sent) {
-        snprintf(buf, sizeof(buf), "<bytes_sent>%f</bytes_sent>\n", bytes_sent);
-        strlcat(msg_buf, buf, sizeof(msg_buf));
-    }
-    if (bytes_received) {
-        snprintf(buf, sizeof(buf), "<bytes_received>%f</bytes_received>\n", bytes_received);
-        strlcat(msg_buf, buf, sizeof(msg_buf));
-    }
     if (ac_state) {
         sprintf(buf, "<sporadic_ac>%d</sporadic_ac>\n", ac_state);
         strlcat(msg_buf, buf, sizeof(msg_buf));
@@ -480,12 +469,13 @@ static void handle_heartbeat_msg() {
     if (parse_double(buf, "<max_wss>", dtemp)) {
         boinc_status.max_working_set_size = dtemp;
     }
-    if (parse_bool(buf, "suspend_network", btemp)) {
+    if (parse_bool(buf, "network_suspended", btemp)) {
         boinc_status.network_suspended = btemp;
     }
     if (parse_int(buf, "<sporadic_ca>", i)) {
         boinc_status.ca_state = (SPORADIC_CA_STATE)i;
     }
+    got_heartbeat_message = true;
 }
 
 // called in timer thread
@@ -970,11 +960,6 @@ void boinc_exit(int status) {
 #endif
 }
 
-void boinc_network_usage(double sent, double received) {
-    bytes_sent = sent;
-    bytes_received = received;
-}
-
 int boinc_is_standalone() {
     if (standalone) return 1;
     return 0;
@@ -1087,8 +1072,8 @@ int boinc_report_app_status_aux(
     double checkpoint_cpu_time,
     double _fraction_done,
     int other_pid,
-    double _bytes_sent,
-    double _bytes_received,
+    double /* _bytes_sent*/,
+    double /* _bytes_received*/,
     double wss
 ) {
     char msg_buf[MSG_CHANNEL_SIZE], buf[1024];
@@ -1106,14 +1091,6 @@ int boinc_report_app_status_aux(
         snprintf(buf, sizeof(buf), "<other_pid>%d</other_pid>\n", other_pid);
         safe_strcat(msg_buf, buf);
     }
-    if (_bytes_sent) {
-        snprintf(buf, sizeof(buf), "<bytes_sent>%f</bytes_sent>\n", _bytes_sent);
-        safe_strcat(msg_buf, buf);
-    }
-    if (_bytes_received) {
-        snprintf(buf, sizeof(buf), "<bytes_received>%f</bytes_received>\n", _bytes_received);
-        safe_strcat(msg_buf, buf);
-    }
     if (ac_state) {
         sprintf(buf, "<sporadic_ac>%d</sporadic_ac>\n", ac_state);
         strlcat(msg_buf, buf, sizeof(msg_buf));
@@ -1122,6 +1099,9 @@ int boinc_report_app_status_aux(
         sprintf(buf, "<wss>%f</wss>\n", wss);
         strlcat(msg_buf, buf, sizeof(msg_buf));
     }
+    if (want_network) {
+        strlcat(msg_buf, "<want_network>1</want_network>\n", sizeof(msg_buf));
+    }
 #ifdef MSGS_FROM_FILE
     if (fout) {
         fputs(msg_buf, fout);
@@ -1371,9 +1351,6 @@ static void handle_process_control_msg() {
     if (match_tag(buf, "<reread_app_info/>")) {
         boinc_status.reread_init_data_file = true;
     }
-    if (match_tag(buf, "<network_available/>")) {
-        have_network = 1;
-    }
 #ifdef ANDROID
     // Trigger call to worker_signal_handler() in the worker thread
     //
@@ -1794,17 +1771,8 @@ int boinc_upload_status(std::string& name) {
     return ERR_NOT_FOUND;
 }
 
-void boinc_need_network() {
-    want_network = 1;
-    have_network = 0;
-}
-
-int boinc_network_poll() {
-    return have_network?0:1;
-}
-
-void boinc_network_done() {
-    want_network = 0;
+void boinc_waiting_for_network(bool x) {
+    want_network = x;
 }
 
 #ifndef _WIN32

api/boinc_api.h

@@ -63,7 +63,7 @@ typedef struct BOINC_OPTIONS {
         // set this if application creates subprocesses.
 } BOINC_OPTIONS;
 
-// info passed from client to app in heartbeat message
+// info passed from client to app in heartbeat and process control messages
 //
 typedef struct BOINC_STATUS {
     int no_heartbeat;
@@ -100,10 +100,6 @@ extern int boinc_report_app_status(
 extern int boinc_time_to_checkpoint(void);
 extern void boinc_begin_critical_section(void);
 extern void boinc_end_critical_section(void);
-extern void boinc_need_network(void);
-extern int boinc_network_poll(void);
-extern void boinc_network_done(void);
-extern void boinc_network_usage(double sent, double received);
 extern int boinc_is_standalone(void);
 extern void boinc_ops_per_cpu_sec(double fp, double integer);
 extern void boinc_ops_cumulative(double fp, double integer);
@@ -132,6 +128,7 @@ extern int setMacIcon(char *filename, char *iconData, long iconSize);
 #include <string>
 #include "app_ipc.h"
 
+extern void boinc_waiting_for_network(bool);
 extern int boinc_resolve_filename_s(const char*, std::string&);
 extern int boinc_get_init_data(APP_INIT_DATA&);
 extern int boinc_wu_cpu_time(double&);
@@ -166,6 +163,7 @@ extern HANDLE worker_thread_handle;
 extern int boinc_init_options_general(BOINC_OPTIONS& opt);
 extern int start_timer_thread(void);
 extern bool boinc_disable_timer_thread;
+extern volatile bool got_heartbeat_message;
 
 inline void boinc_options_defaults(BOINC_OPTIONS& b) {
     b.main_program = 1;
@@ -178,7 +176,6 @@ inline void boinc_options_defaults(BOINC_OPTIONS& b) {
     b.multi_process = 0;
 }
 
-
 /////////// IMPLEMENTATION STUFF ENDS HERE
 
 #endif // C++ part

client/app.cpp

@@ -121,10 +121,6 @@ ACTIVE_TASK::ACTIVE_TASK() {
     run_interval_start_wall_time = gstate.now;
     checkpoint_wall_time = 0;
     elapsed_time = 0;
-    bytes_sent_episode = 0;
-    bytes_received_episode = 0;
-    bytes_sent = 0;
-    bytes_received = 0;
     safe_strcpy(slot_dir, "");
     safe_strcpy(slot_path, "");
     max_elapsed_time = 0;
@@ -795,9 +791,7 @@ int ACTIVE_TASK::write(MIOFILE& fout) {
         "    <swap_size>%f</swap_size>\n"
         "    <working_set_size>%f</working_set_size>\n"
         "    <working_set_size_smoothed>%f</working_set_size_smoothed>\n"
-        "    <page_fault_rate>%f</page_fault_rate>\n"
-        "    <bytes_sent>%f</bytes_sent>\n"
-        "    <bytes_received>%f</bytes_received>\n",
+        "    <page_fault_rate>%f</page_fault_rate>\n",
         result->project->master_url,
         result->name,
         task_state(),
@@ -812,9 +806,7 @@ int ACTIVE_TASK::write(MIOFILE& fout) {
         procinfo.swap_size,
         procinfo.working_set_size,
         procinfo.working_set_size_smoothed,
-        procinfo.page_fault_rate,
-        bytes_sent,
-        bytes_received
+        procinfo.page_fault_rate
     );
     fout.printf("</active_task>\n");
     return 0;
@@ -848,10 +840,7 @@ int ACTIVE_TASK::write_gui(MIOFILE& fout) {
         "    <working_set_size>%f</working_set_size>\n"
         "    <working_set_size_smoothed>%f</working_set_size_smoothed>\n"
         "    <page_fault_rate>%f</page_fault_rate>\n"
-        "    <bytes_sent>%f</bytes_sent>\n"
-        "    <bytes_received>%f</bytes_received>\n"
-        "%s"
-        "%s",
+        "%s%s%s",
         task_state(),
         app_version->version_num,
         slot,
@@ -865,10 +854,9 @@ int ACTIVE_TASK::write_gui(MIOFILE& fout) {
         procinfo.working_set_size,
         procinfo.working_set_size_smoothed,
         procinfo.page_fault_rate,
-        bytes_sent,
-        bytes_received,
         too_large?"   <too_large/>\n":"",
-        needs_shmem?"   <needs_shmem/>\n":""
+        needs_shmem?"   <needs_shmem/>\n":"",
+        want_network?"   <want_network/>\n":""
     );
     if (elapsed_time > first_fraction_done_elapsed_time) {
         fout.printf(
@@ -1008,8 +996,6 @@ int ACTIVE_TASK::parse(XML_PARSER& xp) {
         else if (xp.parse_double("working_set_size_smoothed", procinfo.working_set_size_smoothed)) continue;
         else if (xp.parse_double("page_fault_rate", procinfo.page_fault_rate)) continue;
         else if (xp.parse_double("current_cpu_time", x)) continue;
-        else if (xp.parse_double("bytes_sent", bytes_sent)) continue;
-        else if (xp.parse_double("bytes_received", bytes_received)) continue;
         else {
             if (log_flags.unparsed_xml) {
                 msg_printf(project, MSG_INFO,
@@ -1220,23 +1206,13 @@ void ACTIVE_TASK_SET::handle_upload_files() {
     }
 }
 
-bool ACTIVE_TASK_SET::want_network() {
+bool ACTIVE_TASK_SET::some_task_wants_network() {
     for (ACTIVE_TASK* atp: active_tasks) {
         if (atp->want_network) return true;
     }
     return false;
 }
 
-void ACTIVE_TASK_SET::network_available() {
-#ifndef SIM
-    for (ACTIVE_TASK* atp: active_tasks) {
-        if (atp->want_network) {
-            atp->send_network_available();
-        }
-    }
-#endif
-}
-
 void ACTIVE_TASK::upload_notify_app(const FILE_INFO* fip, const FILE_REF* frp) {
     char path[MAXPATHLEN];
     snprintf(path, sizeof(path),

client/app.h

@@ -127,13 +127,6 @@ struct ACTIVE_TASK {
         // wall time at the last checkpoint
     double elapsed_time;
         // current total running time, adjusted for CPU throttling
-    double bytes_sent_episode;
-        // bytes sent in current episode of job,
-        // as (optionally) reported by boinc_network_usage()
-    double bytes_received_episode;
-    double bytes_sent;
-        // bytes in all episodes
-    double bytes_received;
     char slot_dir[256];
         // directory where process runs (relative)
     char slot_path[MAXPATHLEN];
@@ -159,8 +152,9 @@ struct ACTIVE_TASK {
     bool needs_shmem;
         // waiting for a free shared memory segment
     int want_network;
-        // This task wants to do network comm (for F@h)
-        // this is passed via share-memory message (app_status channel)
+        // This task is waiting for the network
+        // (physical connection or no suspension)
+        // This is passed via share-memory message (app_status channel)
     double abort_time;
         // when we sent an abort message to this app
         // kill it 5 seconds later if it doesn't exit
@@ -300,7 +294,6 @@ struct ACTIVE_TASK {
     int preempt(PREEMPT_TYPE preempt_type, int reason=0);
         // preempt (via suspend or quit) a running task
     int resume_or_start(bool);
-    void send_network_available();
 #ifdef _WIN32
     void handle_exited_app(unsigned long);
 #else
@@ -363,8 +356,7 @@ class ACTIVE_TASK_SET {
     void report_overdue();
     void handle_upload_files();
     void upload_notify_app(FILE_INFO*);
-    bool want_network();    // does any task want network?
-    void network_available();   // notify tasks that network is available
+    bool some_task_wants_network();
     void free_mem();
     bool slot_taken(int);
     void get_memory_usage();