NerlMonitor

.gitignore

@@ -37,4 +37,5 @@ examples/.ipynb_checkpoints/simple_run-checkpoint.ipynb
 /Results
 arch.json
 conn.json
-/inputDataDir
+/inputDataDir
+/NerlNetGraph.png

NerlnetMonitor.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+
+MONITOR_PATH="src_erl/NerlMonitor"
+GUI_PATH="src_erl/NerlMonitor/src"
+
+echo "NerlnetMonitor Activated"
+
+
+cd $MONITOR_PATH
+rebar3 shell --name [email protected] --setcookie COOKIE 
+
+cd ../../

README.md

@@ -76,4 +76,4 @@ Minimum Python version: 3.8
 4. Run Jupyter notebook with ```jupyter-notebook``` and create a new notebook in the created dir from step 3. 
 5. Follow the example: https://github.com/leondavi/NErlNet/blob/master/examples/example_run.ipynb 
 
-Contact Email: [email protected]
+Contact Email: [email protected]

inputJsonFiles/Architecture/arch_1PCSIM6WorkerSynth.json

@@ -6,13 +6,13 @@
   },
   "devices": [
     {
-      "host": "192.168.0.108",
+      "host": "192.168.64.7",
       "entities": "mainServer,c1,c2,c3,c4,c5,c6,s1,r1,r2,r3,r4,r5,r6,apiServer"
     }
   ],
   "apiServer": 
     {
-      "host": "192.168.0.108",
+      "host": "192.168.64.7",
       "port": "8095",
       "args": ""
     }
@@ -26,7 +26,7 @@
   ,
   "mainServer": 
     {
-      "host": "192.168.0.108",
+      "host": "192.168.64.7",
       "port": "8080",
       "args": ""
     }

inputJsonFiles/experimentsFlow/exp_6Workers1SourceMNist.json

@@ -1,7 +1,7 @@
 {
    "Features": 784,
    "Labels": ["0", "1", "2", "3", "4", "5", "6", "7", "8", "9"],
-   "CSV path": "mnist",
+   "CSV path": "mnist-o",
    "Batches per source": 
      {
         "Training": 10000,

src_erl/Communication_Layer/http_Nerlserver/src/Client/clientStateHandler.erl

@@ -30,7 +30,8 @@ init(Req0, [Action,Client_StateM_Pid]) ->
     idle        -> gen_statem:cast(Client_StateM_Pid,{idle});
     training    -> gen_statem:cast(Client_StateM_Pid,{training});
     predict     -> gen_statem:cast(Client_StateM_Pid,{predict});
-    statistics  -> gen_statem:cast(Client_StateM_Pid,{statistics})
+    statistics  -> gen_statem:cast(Client_StateM_Pid,{statistics});
+    worker_kill -> gen_statem:cast(Client_StateM_Pid,{worker_kill , Body})
   end,
 
   %% reply ACKnowledge to main server for initiating, later send finished initiating http_request from client_stateM

src_erl/Communication_Layer/http_Nerlserver/src/MainServer/actionHandler.erl

@@ -25,7 +25,9 @@ init(Req0, [Action, Main_genserver_Pid]) ->
     predictRes ->  gen_statem:cast(Main_genserver_Pid, {predictRes,Body});
     statistics ->  gen_statem:cast(Main_genserver_Pid, {statistics,Body});
     startCasting ->  gen_statem:cast(Main_genserver_Pid, {startCasting,Body});
-    stopCasting ->  gen_statem:cast(Main_genserver_Pid, {stopCasting,Body})
+    stopCasting ->  gen_statem:cast(Main_genserver_Pid, {stopCasting,Body});
+    %monitor
+    worker_down ->  gen_statem:cast(Main_genserver_Pid, {worker_down,Body})
   end,
   Reply = io_lib:format("Body Received: ~p ~n ", [Body]),
   Req = cowboy_req:reply(200,

src_erl/Communication_Layer/http_Nerlserver/src/MainServer/mainGenserver.erl

@@ -22,7 +22,7 @@
 -define(SERVER, ?MODULE).
 
 
--record(main_genserver_state, {statisticsCounter = 0, myName, state, workersMap, clients, nerlnetGraph, sourcesCastingList = [], sourcesWaitingList = [], clientsWaitingList = [], statisticsMap, msgCounter = 0, batchSize}).
+-record(main_genserver_state, {statisticsCounter = 0, myName, state, workersMap, clients, nerlnetGraph, sourcesCastingList = [], sourcesWaitingList = [], clientsWaitingList = [], statisticsMap, msgCounter = 0, batchSize , etsRef}).
 
 %%%===============================================================
 %%% API
@@ -56,10 +56,10 @@ init({MyName,Clients,BatchSize,WorkersMap,NerlnetGraph}) ->
   ?LOG_NOTICE("Main Server is connected to: ~p~n",[ConnectedEntities]),
   put(nerlnetGraph, NerlnetGraph),
     % nerl_tools:start_connection([digraph:vertex(NerlnetGraph,Vertex) || Vertex <- digraph:out_neighbours(NerlnetGraph,MyName)]),
-
+  EtsRef = ets:new(main_server_data , [set]) , %% ! In some point the whole record will replaced by ets
   NewStatisticsMap = getNewStatisticsMap([digraph:vertex(NerlnetGraph,Vertex) || Vertex <- digraph:vertices(NerlnetGraph)--?LIST_OF_SPECIAL_SERVERS]),
   % io:format("New StatisticsMap = ~p~n",[NewStatisticsMap]),
-  {ok, #main_genserver_state{myName = MyNameStr, workersMap = WorkersMap, batchSize = BatchSize, state=idle, clients = Clients, nerlnetGraph = NerlnetGraph, msgCounter = 1,statisticsMap = NewStatisticsMap}}.
+  {ok, #main_genserver_state{myName = MyNameStr, workersMap = WorkersMap, batchSize = BatchSize, state=idle, clients = Clients, nerlnetGraph = NerlnetGraph, msgCounter = 1,statisticsMap = NewStatisticsMap , etsRef = EtsRef}}.
 
 %% @private
 %% @doc Handling call messages
@@ -76,10 +76,11 @@ init({MyName,Clients,BatchSize,WorkersMap,NerlnetGraph}) ->
 handle_call(getGraph, _From, State) ->
   NerlGraph = State#main_genserver_state.nerlnetGraph,
   FullNodes = [digraph:vertex(NerlGraph,Vertex) || Vertex <- digraph:vertices(NerlGraph)],
-  %io:format("Full graph is: ~p~n", [FullNodes]),
-  NodesList = [Entity++","++IP++","++integer_to_list(Port)++"#"||{Entity, {IP, Port}} <- FullNodes],
+  %%io:format("Full graph is: ~p~n", [FullNodes]),
+  NodesList = [atom_to_list(Entity)++","++binary_to_list(IP)++","++integer_to_list(Port)++"#"||{Entity, {IP, Port}} <- FullNodes],
+  %%io:format("graph nodes are: ~p~n", [NodesList]),
   EdgesList = [digraph:edge(NerlGraph,Edge) || Edge <- digraph:edges(NerlGraph)],
-  %io:format("graph edges are: ~p~n", [EdgesList]),
+  %%io:format("graph edges are: ~p~n", [EdgesList]),
   Nodes = nodeString(NodesList),
   Edges = edgeString(EdgesList),
 
@@ -100,11 +101,11 @@ nodeString([Node |NodeList]) -> nodeString(NodeList, Node).
 nodeString([], Str) -> Str;
 nodeString([Node |NodeList], Str)-> nodeString(NodeList, Node++Str).
 
-edgeString([Edge |EdgesList])-> {_ID, V1, V2, _Label} = Edge, edgeString(EdgesList, V1++"-"++V2).
+edgeString([Edge |EdgesList])-> {_ID, V1, V2, _Label} = Edge, edgeString(EdgesList, atom_to_list(V1)++"-"++atom_to_list(V2)).
 edgeString([], Str)-> Str;
 edgeString([Edge |EdgesList], Str)->
   {_ID, V1, V2, _Label} = Edge,
-  edgeString(EdgesList, V1++"-"++V2++","++Str).
+  edgeString(EdgesList, atom_to_list(V1)++"-"++atom_to_list(V2)++","++Str).
 
 %% @private
 %% @doc Handling cast messages
@@ -113,6 +114,11 @@ edgeString([Edge |EdgesList], Str)->
 {noreply, NewState :: #main_genserver_state{}, timeout() | hibernate} |
 {stop, Reason :: term(), NewState :: #main_genserver_state{}}).
 
+handle_cast({saveUtility , Body} , State = #main_genserver_state{etsRef = EtsRef , msgCounter = MsgCounter}) -> 
+  %gets called after a tool tried to connect, saves its addres in the ets.
+  ets:insert(EtsRef , Body),
+  {noreply, State#main_genserver_state{msgCounter = MsgCounter+1}};
+
 
 handle_cast({initCSV, Source,SourceData}, State = #main_genserver_state{state = idle, myName = MyName, sourcesWaitingList = SourcesWaitingList,nerlnetGraph = NerlnetGraph,msgCounter = MsgCounter}) ->
 %%  send router http request, to rout this message to all sensors
@@ -158,33 +164,60 @@ handle_cast({clientsIdle}, State = #main_genserver_state{state = idle, myName =
   {noreply, State#main_genserver_state{clientsWaitingList = ListOfClients,msgCounter = MsgCounter+1}};
 
 %%%get Statistics from all Entities in the network
-handle_cast({statistics,Body}, State = #main_genserver_state{myName = MyName, statisticsCounter = StatisticsCounter, nerlnetGraph = NerlnetGraph,statisticsMap = StatisticsMap,msgCounter = MsgCounter}) ->
+handle_cast({statistics,Body}, State = #main_genserver_state{myName = MyName, statisticsCounter = StatisticsCounter, nerlnetGraph = NerlnetGraph,statisticsMap = StatisticsMap,msgCounter = MsgCounter , etsRef = EtsRef , state = CurrState}) ->
 
     if Body == <<"getStatistics">> ->   %% initial message from APIServer, get stats from entities
-
           NewStatisticsMap = getNewStatisticsMap([digraph:vertex(NerlnetGraph,Vertex) || Vertex <- digraph:vertices(NerlnetGraph)--?LIST_OF_SPECIAL_SERVERS]),
-          [findroutAndsendStatistics(MyName, Name)||{Name,_Counter}<-maps:to_list(StatisticsMap)],
-          NewState = State#main_genserver_state{msgCounter = MsgCounter+1,statisticsMap = NewStatisticsMap, statisticsCounter = length(maps:to_list(StatisticsMap))};
-
+          [findroutAndsendStatistics(MyName, Name)||{Name,_Counter}<-maps:to_list(NewStatisticsMap)],
+          NewState = State#main_genserver_state{msgCounter = MsgCounter+1,statisticsMap = NewStatisticsMap, statisticsCounter = length(maps:to_list(NewStatisticsMap))};
       Body == <<>> ->  io:format("in Statistcs, State has: StatsCount=~p, MsgCount=~p~n", [StatisticsCounter, MsgCounter]), NewState = State;
 
       true ->
           %%      statistics arrived from Entity
-          [From|[NewCounter]] = re:split(binary_to_list(Body), ":", [{return, list}]),
-
-          NewStatisticsMap = maps:put(From,NewCounter,StatisticsMap),
+          CheckForDead=re:split(binary_to_list(Body), "#", [{return, list}]),
+          case CheckForDead of
+            [_]-> 
+              Data=CheckForDead,
+              UpdateStatMap=StatisticsMap; %entity is not a client
+            [Data,[]]->
+              UpdateStatMap=StatisticsMap;%entity is client without dead workers
+            [Data,DeadWorkers]->                %client with dead workers
+              case maps:is_key("Dead workers",StatisticsMap) of
+                true->
+                  MapWithKey=StatisticsMap; %key was already put in
+                false->
+                  MapWithKey=maps:put("Dead workers","",StatisticsMap)
+                end,
+                UpdateStatMap=maps:put("Dead workers",maps:get("Dead workers",MapWithKey)++DeadWorkers++" ",MapWithKey)
+          end,
+          [From|[NewCounter]] = re:split(Data, ":", [{return, list}]),
+
+          NewStatisticsMap = maps:put(From,NewCounter,UpdateStatMap),
           NewState = State#main_genserver_state{msgCounter = MsgCounter+1,statisticsMap = NewStatisticsMap,statisticsCounter = StatisticsCounter-1},
 
           if StatisticsCounter == 1 ->  %% got stats from all entities
               Statistics = maps:to_list(NewStatisticsMap),
               S = mapToString(Statistics,[]) ,
-              ?LOG_NOTICE("Sending stats: ~p~n",[S]),
-              {RouterHost,RouterPort} = nerl_tools:getShortPath(MyName,?API_SERVER_ATOM,NerlnetGraph),
-              nerl_tools:http_request(RouterHost,RouterPort,"statistics", S ++ "|mainServer:" ++integer_to_list(MsgCounter));
-
-          %% wait for more stats
-          true -> pass end
-      end,
+              %?LOG_NOTICE("Sending stats: ~p~n",[S]),
+              case CurrState of %statistics gets called in idle state by api at the end of the experiment or nerlMonitorin casting at the start of predict
+                idle ->
+                    case ets:member(EtsRef,nerlMonitor) of
+                      true->
+                        [{nerlMonitor , IP , Port}] = ets:lookup(EtsRef , nerlMonitor),
+                        nerl_tools:http_request(IP,list_to_integer(Port),"stats", S ++ "|mainServer:" ++integer_to_list(MsgCounter));
+                      false -> ok
+                    end,
+                    {RouterHost,RouterPort} = nerl_tools:getShortPath(MyName,?API_SERVER_ATOM,NerlnetGraph),
+                    nerl_tools:http_request(RouterHost,RouterPort,"statistics", S ++ "|mainServer:" ++integer_to_list(MsgCounter));
+                casting ->    
+                    [{_ , MonitorIP , MonitorPort}] = ets:lookup(EtsRef, nerlMonitor),
+                    nerl_tools:http_request(MonitorIP,list_to_integer(MonitorPort),"stats", S ++ "|mainServer:" ++integer_to_list(MsgCounter))
+              end;
+
+            %% wait for more stats
+            true -> pass 
+          end
+    end,
     {noreply, NewState};
 
 %%handle_cast({startPredicting}, State = #main_genserver_state{clients = ListOfClients, nerlnetGraph = NerlnetGraph}) ->
@@ -223,10 +256,17 @@ handle_cast({sourceAck,Body}, State = #main_genserver_state{sourcesWaitingList =
   {noreply, State#main_genserver_state{sourcesWaitingList = NewWaitingList,msgCounter = MsgCounter+1}};
 
 
-handle_cast({clientAck,Body}, State = #main_genserver_state{clientsWaitingList = WaitingList,msgCounter = MsgCounter}) ->
-  NewWaitingList = WaitingList--[list_to_atom(binary_to_list(Body))],
-  % io:format("new Waiting List: ~p ~n",[NewWaitingList]),
-  if length(NewWaitingList) == 0 -> ack();
+handle_cast({clientAck,Body}, State = #main_genserver_state{clientsWaitingList = WaitingList,msgCounter = MsgCounter , etsRef = EtsRef}) ->
+  ClientName = list_to_atom(binary_to_list(Body)),
+  NewWaitingList = WaitingList--[ClientName],
+  if length(NewWaitingList) == 0 -> 
+                case ets:member(EtsRef , nerlMonitor) of      %if NerlMonitor is up, initiate mid experiment statistics
+                  true -> 
+                    ?LOG_INFO(?LOG_HEADER ++ "Sending statistics request"),
+                    gen_server:cast(self(),{statistics , list_to_binary("getStatistics")});
+                  _ -> ok
+                end,
+                ack();
   true-> ok end,
   {noreply, State#main_genserver_state{clientsWaitingList = NewWaitingList, msgCounter = MsgCounter+1}};
 
@@ -310,6 +350,26 @@ handle_cast({predictRes,Body}, State = #main_genserver_state{batchSize = BatchSi
   end,
   {noreply, State#main_genserver_state{msgCounter = MsgCounter+1}};
 
+handle_cast({worker_down,Body}, State = #main_genserver_state{msgCounter = MsgCounter,etsRef = EtsRef}) ->
+  % some worker terminated, print to log and notify NerlMonitor if it is up
+  case ets:member(EtsRef,nerlMonitor) of
+    true->
+      [{nerlMonitor , IP , Port}] = ets:lookup(EtsRef , nerlMonitor),
+      URL = "http://" ++ IP ++ ":" ++ Port ++ "/utilInfo",
+      httpc:request(post,{URL, [],"application/x-www-form-urlencoded",Body}, [], []);
+    false -> ok
+  end, 
+  ?LOG_WARNING(?LOG_HEADER++"Worker down , ~p disconneted~n",[binary_to_list(Body)]),
+  {noreply, State#main_genserver_state{msgCounter = MsgCounter+1,etsRef=EtsRef}};
+
+handle_cast({worker_kill , WorkerName} , State = #main_genserver_state{workersMap = WorkersMap , msgCounter = MsgCounter}) ->
+  %got a user kill command for a worker from NerlMonitor
+  ?LOG_WARNING(?LOG_HEADER++"Killing worker ~p ~n",[WorkerName]),
+  WorkerNameAtom=binary_to_term(WorkerName),
+  ClientName = maps:get(WorkerNameAtom,WorkersMap),
+  Body = term_to_binary({ClientName, WorkerNameAtom}),
+  nerl_tools:sendHTTP(?MAIN_SERVER_ATOM , ClientName , atom_to_list(worker_kill) , Body),
+  {noreply, State#main_genserver_state{workersMap = WorkersMap , msgCounter = MsgCounter+1}};
 
 handle_cast(Request, State = #main_genserver_state{}) ->
   io:format("main server cast ignored: ~p~n",[Request]),
@@ -377,9 +437,9 @@ getNewStatisticsMap([{Name,{_Host, _Port}}|Tail],StatisticsMap) ->
   getNewStatisticsMap(Tail,maps:put(atom_to_list(Name), 0, StatisticsMap)).
 
 
-startCasting([],_NumOfSampleToSend,_MyName, _NerlnetGraph)->done;
+startCasting([],_NumOfSampleToSend,_MyName, _NerlnetGraph) -> done;
 startCasting([SourceName|SourceNames],NumOfSampleToSend, MyName, NerlnetGraph)->
-  ?LOG_NOTICE("~p sending start casting command to: ~p",[MyName, SourceName]),
+  ?LOG_NOTICE("~p sending start casting command to: ~p~n",[MyName, SourceName]),
   {RouterHost,RouterPort} = nerl_tools:getShortPath(MyName,SourceName,NerlnetGraph),
   nerl_tools:http_request(RouterHost,RouterPort,"startCasting", SourceName++[","]++NumOfSampleToSend),
   startCasting(SourceNames,NumOfSampleToSend, MyName, NerlnetGraph).

src_erl/Communication_Layer/http_Nerlserver/src/Router/routerGenserver.erl

@@ -176,6 +176,18 @@ handle_cast({getStats,_Body}, State  = #router_genserver_state{myName = MyName,
     nerl_tools:http_request(Host,Port,"routerStats",Mes),
     {noreply, State#router_genserver_state{msgCounter = MsgCounter+1}};
 
+handle_cast({worker_kill , Body} , State = #router_genserver_state{msgCounter = MsgCounter, myName = MyName}) ->
+    io:format("Body is ~p~n",[binary_to_term(Body)]),
+    {ClientName , _} = binary_to_term(Body),
+    nerl_tools:sendHTTP(MyName, ClientName, "worker_kill", Body),
+    {noreply, State#router_genserver_state{msgCounter = MsgCounter+1}};
+
+
+%monitor
+handle_cast({worker_down,Body}, State = #router_genserver_state{myName = MyName, msgCounter = MsgCounter, nerlnetGraph = NerlnetGraph}) ->
+    nerl_tools:sendHTTP(MyName, ?MAIN_SERVER_ATOM, "worker_down", Body),
+    {noreply, State#router_genserver_state{msgCounter = MsgCounter+1}};
+
 
 handle_cast(_Request, State = #router_genserver_state{msgCounter = MsgCounter }) ->
   {noreply, State#router_genserver_state{msgCounter = MsgCounter+1}}.

src_erl/Communication_Layer/http_Nerlserver/src/Router/routingHandler.erl

@@ -74,7 +74,10 @@ init(Req0, State) ->
     %   gen_server:cast(Router_genserver_Pid, {federatedWeights,Body});
 
     %%%%%%%%%%%%%%GUI actions
-    getStats -> gen_server:cast(Router_genserver_Pid, {getStats,Body})
+    getStats -> gen_server:cast(Router_genserver_Pid, {getStats,Body});
+    %monitor
+    worker_down -> gen_server:cast(Router_genserver_Pid, {worker_down,Body});
+    worker_kill -> gen_server:cast(Router_genserver_Pid, {worker_kill,Body})
 
   end,
   Reply = io_lib:format(" ", []),

src_erl/Communication_Layer/http_Nerlserver/src/nerlNetServer_app.erl

@@ -187,7 +187,8 @@ createClientsAndWorkers() ->
                 {"/clientTraining",clientStateHandler, [training,ClientStatemPid]},
                 {"/clientIdle",clientStateHandler, [idle,ClientStatemPid]},
                 {"/clientPredict",clientStateHandler, [predict,ClientStatemPid]},
-                {"/weightsVector",clientStateHandler, [vector,ClientStatemPid]}
+                {"/weightsVector",clientStateHandler, [vector,ClientStatemPid]},
+                {"/worker_kill" , clientStateHandler , [worker_kill , ClientStatemPid]}
             ]}
         ]),
         init_cowboy_start_clear(Client, {HostName, Port},NerlClientDispatch)
@@ -265,7 +266,10 @@ createRouters(MapOfRouters, HostName) ->
                 {"/federatedWeights",routingHandler, [federatedWeights,RouterGenServerPid]},
 
                 %%GUI actions
-                {"/getStats",routingHandler, [getStats,RouterGenServerPid]}
+                {"/getStats",routingHandler, [getStats,RouterGenServerPid]},
+                %monitor actions
+                {"/worker_down",routingHandler, [worker_down,RouterGenServerPid]},
+                {"/worker_kill" , routingHandler , [worker_kill , RouterGenServerPid]}
             ]}
         ]),
         %% cowboy:start_clear(Name, TransOpts, ProtoOpts) - an http_listener
@@ -304,7 +308,9 @@ createMainServer(true,BatchSize,HostName) ->
         %GUI actions
         {"/getGraph",[],guiHandler, [getGraph, MainGenServerPid]},
         {"/getStats",[],guiHandler, [getStats, MainGenServerPid]},
-
+        {"/toolConnectionReq" , [] , utilities_handler , [MainGenServerPid]} , %% Added with NerlMonitor Project
+        {"/worker_kill" , [] , utilities_handler , [worker_kill , MainGenServerPid]},
+        {"/worker_down",actionHandler, [worker_down,MainGenServerPid]},
         {"/[...]", [],noMatchingRouteHandler, [MainGenServerPid]}
         ]}
         ]),