State bug fixed, CR configurable from args, finish frontend endpoint

Author: gferrate

frontend/app.py

@@ -14,6 +14,7 @@
     'clients': {},
     'iteration': -1,
     'started_training': False
+    'finished_training': False
 }
 
 def check_clients_ok():
@@ -25,6 +26,7 @@ def check_clients_ok():
         else:
             client['check_ok'] = True
 
+
 @app.route('/')
 def index():
     check_clients_ok()
@@ -38,10 +40,18 @@ def restart():
         'clients': {},
         'iteration': -1,
         'started_training': False
+        'finished_training': False
     }
     return jsonify(global_state)
 
 
+@app.route('/finish')
+def finish():
+    global global_state
+    global_state['finished_training'] = True
+    return jsonify(global_state)
+
+
 @app.route('/fake')
 def fake_data():
     global global_state

orchestrator/orchestrator.py

@@ -51,6 +51,16 @@ def send_iteration_to_frontend(i):
         logging.warning('Frontend may be down')
 
 
+def end_frontend():
+    logging.info('Sending end signal to frontend')
+    try:
+        requests.post(
+            url='http://{}:{}/finish'.format(hosts['frontend']['host'],
+                                             hosts['frontend']['port']))
+    except:
+        logging.warning('Frontend may be down')
+
+
 def restart_frontend():
     logging.info('Restarting frontend')
     try:
@@ -64,16 +74,14 @@ def restart_frontend():
         logging.warning('Frontend may be down')
 
 
-def main(op_mode):
-    # TODO: Configure epochs and everything from here
-    NUM_ITERATIONS = 50
+def main(op_mode, communication_rounds):
     all_results = []
     ch = client_handler.ClientHandler(clients=hosts['clients'],
                                       OPERATION_MODE=op_mode)
     #train_accs = {}
     start = time.time()
     # restart_frontend()
-    for i in range(NUM_ITERATIONS):
+    for i in range(communication_rounds):
         logging.info('Iteration {}...'.format(i))
         send_iteration_to_frontend(i)
 
@@ -131,6 +139,7 @@ def main(op_mode):
 
     logging.info('All results:')
     logging.info(all_results)
+    end_frontend()
 
 
 if __name__ == '__main__':
@@ -142,9 +151,15 @@ def main(op_mode):
                             'Operation mode. '
                             'Options: wait_all (default), n_firsts, timeout'
                         ))
+    parser.add_argument('-c',
+                        '--communication-rounds',
+                        type=int,
+                        required=False,
+                        default=50,
+                        help='Number of communication rounds. Default: 50)
     args = parser.parse_args()
     try:
-        main(op_mode=args.operation_mode)
+        main(op_mode=args.operation_mode, args.communication_rounds)
     except Exception:
         logging.error("Fatal error in main loop", exc_info=True)
 

shared/state.py

@@ -48,10 +48,11 @@ def current_state(self):
 
     @current_state.setter
     def current_state(self, state):
+        self._current_state = state
         payload = {
             'client_type': self.client_type,
             '_id': self._id,
-            'state': self.get_state_string(self.current_state),
+            'state': self.get_state_string(self._current_state),
             'port': self.port,
             'host': self.host
         }
@@ -66,8 +67,6 @@ def current_state(self, state):
         except Exception as e:
             logging.warning('Frontend not reachable.\n{}'.format(e))
 
-        self._current_state = state
-
     def idle(self):
         self.current_state = IDLE