[API-SERVER] New DB inherritance approach

src_erl/NerlnetApp/src/Client/clientStatem.erl

@@ -89,7 +89,7 @@ init({MyName,NerlnetGraph, ClientWorkers , WorkerShaMap , WorkerToClientMap , Sh
   clientWorkersFunctions:create_workers(MyName , EtsRef , ShaToModelArgsMap , EtsStats),
   %% send pre_idle signal to workers
   WorkersNames = clientWorkersFunctions:get_workers_names(EtsRef),
-  [gen_statem:cast(clientWorkersFunctions:get_worker_pid(EtsRef , WorkerName), {pre_idle}) || WorkerName <- clientWorkersFunctions:get_workers_names(EtsRef)],
+  [gen_statem:cast(clientWorkersFunctions:get_worker_pid(EtsRef , WorkerName), {pre_idle}) || WorkerName <- WorkersNames],
 
   % update dictionary
   WorkersEts = ets:lookup_element(EtsRef , workers_ets , ?DATA_IDX),
@@ -133,11 +133,11 @@ waitforWorkers(cast, In = {NewState}, State = #client_statem_state{myName = _MyN
   cast_message_to_workers(EtsRef, {NewState}), %% This function increments the number of sent messages in stats ets
   {next_state, waitforWorkers, State#client_statem_state{nextState = NewState, waitforWorkers = Workers}};
 
-waitforWorkers(cast, EventContent, State = #client_statem_state{myName = _MyName}) ->
+waitforWorkers(cast, EventContent, State = #client_statem_state{myName = MyName}) ->
   ClientStatsEts = get(client_stats_ets),
   stats:increment_messages_received(ClientStatsEts),
   stats:increment_bytes_received(ClientStatsEts , nerl_tools:calculate_size(EventContent)),
-  ?LOG_WARNING("client ~p waitforWorkers ignored!!!:  ~p ~n",[myName, EventContent]),
+  ?LOG_WARNING("client ~p waitforWorkers ignored!!!:  ~p ~n",[MyName, EventContent]),
   {next_state, waitforWorkers, State}.
 
 
@@ -167,17 +167,19 @@ idle(cast, _In = {statistics}, State = #client_statem_state{ myName = MyName, et
   EtsStats = get(ets_stats),
   ClientStatsEts = get(client_stats_ets),
   ClientStatsEncStr = stats:encode_ets_to_http_bin_str(ClientStatsEts),
-  ClientStatsToSend = atom_to_list(MyName) ++ ?API_SERVER_WITHIN_ENTITY_SEPERATOR ++ ClientStatsEncStr ++ ?API_SERVER_ENTITY_SEPERATOR,
+  %ClientStatsToSend = atom_to_list(MyName) ++ ?API_SERVER_WITHIN_ENTITY_SEPERATOR ++ ClientStatsEncStr ++ ?API_SERVER_ENTITY_SEPERATOR,
   stats:increment_messages_received(ClientStatsEts),
-  ListStatsEts = ets:tab2list(EtsStats) -- [{MyName , ClientStatsEts}],
+  ListStatsEts = ets:tab2list(EtsStats) -- [{MyName , ClientStatsEts}], 
   WorkersStatsEncStr = create_encoded_stats_str(ListStatsEts),
-  StatsBody = {MyName , ClientStatsToSend ++ WorkersStatsEncStr},
+  DataToSend = ClientStatsEncStr ++ WorkersStatsEncStr,
+  io:format("DataToSend: ~p~n",[DataToSend]),
+  StatsBody = {MyName , DataToSend},
   {RouterHost,RouterPort} = ets:lookup_element(EtsRef, my_router, ?DATA_IDX),
   nerl_tools:http_router_request(RouterHost, RouterPort, [?MAIN_SERVER_ATOM], atom_to_list(statistics), StatsBody),
   stats:increment_messages_sent(ClientStatsEts),
   {next_state, idle, State};
 
-idle(cast, In = {training}, State = #client_statem_state{myName = MyName, etsRef = EtsRef}) ->
+idle(cast, In = {training}, State = #client_statem_state{myName = _MyName, etsRef = EtsRef}) ->
   ClientStatsEts = get(client_stats_ets),
   stats:increment_messages_received(ClientStatsEts),
   stats:increment_bytes_received(ClientStatsEts , nerl_tools:calculate_size(In)),  MessageToCast = {training},
@@ -408,7 +410,8 @@ cast_message_to_workers(EtsRef, Msg) ->
 
 create_encoded_stats_str(ListStatsEts) ->
   Func = fun({WorkerName , StatsEts}) ->
-    WorkerEncStatsStr = stats:encode_ets_to_http_bin_str(StatsEts),
-    WorkerName ++ ?API_SERVER_WITHIN_ENTITY_SEPERATOR ++ WorkerEncStatsStr ++ ?API_SERVER_ENTITY_SEPERATOR
+    WorkerEncStatsStr = stats:encode_workers_ets_to_http_bin_str(StatsEts),
+    %% w1&bytes_sent:6.0:float#bad_messages:0:int....|
+    atom_to_list(WorkerName) ++ ?WORKER_SEPERATOR ++ WorkerEncStatsStr
     end,
   lists:flatten(lists:map(Func , ListStatsEts)).

src_erl/NerlnetApp/src/MainServer/mainGenserver.erl

@@ -64,8 +64,9 @@ init({MyName,ClientsNames,BatchSize,WorkersMap,NerlnetGraph}) ->
   EtsStats = ets:new(stats , [set]),
   put(etsStats, EtsStats), %% All entities including mainServer ets tables statistics
   Entities = [digraph:vertex(NerlnetGraph,Vertex) || Vertex <- digraph:vertices(NerlnetGraph)--[?API_SERVER_ATOM]],
-  generate_stats_ets_tables(Entities),
-  ets:insert(MainServerEts , {entities_names_list , Entities -- [?MAIN_SERVER_ATOM]}),
+  EntitiesNames = [Name || {Name, _CommTuple} <- Entities],
+  generate_stats_ets_tables(EntitiesNames),
+  ets:insert(MainServerEts , {entities_names_list , EntitiesNames -- [?MAIN_SERVER_ATOM]}),
   ets:insert(MainServerEts , {batch_size , BatchSize}),
   ets:insert(MainServerEts , {workers_map , WorkersMap}),
   ets:insert(MainServerEts , {clients_names_list , ClientsNames}),
@@ -153,11 +154,12 @@ handle_cast({statistics,Body}, State = #main_genserver_state{myName = MyName}) -
       true ->
           %% TODO - Guy here you should get the the encoded statistics from entities and decode it use it the function you should implement
           %%      statistics arrived from Entity
-          {From, StatsEtsEncStr} = Body,
-          EntityName = binary_to_atom(From), %TODO Guy
-          set_entity_stats_ets_str(EntityName, StatsEtsEncStr),
+          {From, StatsEtsEncStr} = binary_to_term(Body),
+          %% EntityName = binary_to_atom(From), %TODO Guy / NO NEED
+          set_entity_stats_ets_str(From, StatsEtsEncStr),
           % TODO increase counter_received_stats ets by 1
-          ets:update_counter(StatsEts, counter_received_stats, 1),
+          ets:update_counter(get(main_server_ets), counter_received_stats, 1),
+          stats:increment_messages_received(StatsEts),
           % [From|[NewCounter]] = re:split(binary_to_list(Body), ":", [{return, list}]),
 
           % NewStatisticsMap = maps:put(From,NewCounter,StatisticsMap),
@@ -166,17 +168,17 @@ handle_cast({statistics,Body}, State = #main_genserver_state{myName = MyName}) -
           ReceivedCounterStatsValue = ets:lookup_element(get(main_server_ets), counter_received_stats, ?DATA_IDX),
           EntitiesNamesList = ets:lookup_element(get(main_server_ets), entities_names_list, ?DATA_IDX),
           TotalNumOfEntities = length(EntitiesNamesList), % without MainServer!
-
           if ReceivedCounterStatsValue == TotalNumOfEntities ->  %% got stats from all entities
             Func = fun(Entity) ->
-              EntityStatsEncStr = get_entity_stats_ets_str(EntityName),
-              Entity ++ ?API_SERVER_WITHIN_ENTITY_SEPERATOR ++ EntityStatsEncStr ++ ?API_SERVER_ENTITY_SEPERATOR
+              EntityStatsEncStr = get_entity_stats_ets_str(Entity),
+              atom_to_list(Entity) ++ ?API_SERVER_WITHIN_ENTITY_SEPERATOR ++ EntityStatsEncStr ++ ?API_SERVER_ENTITY_SEPERATOR
             end,
             MainServerEncStatsEts = stats:encode_ets_to_http_bin_str(get_entity_stats_ets_str(?MAIN_SERVER_ATOM)),
             MainServerStr = atom_to_list(?MAIN_SERVER_ATOM) ++ ?API_SERVER_WITHIN_ENTITY_SEPERATOR ++ MainServerEncStatsEts ++ ?API_SERVER_ENTITY_SEPERATOR,
             StatsToSend = lists:flatten([Func(Entity) || Entity <- EntitiesNamesList] ++ MainServerStr), % add main server to the list
             {RouterHost,RouterPort} = ets:lookup_element(get(main_server_ets), my_router, ?DATA_IDX),
             ActionStr = atom_to_list(statistics),
+            io:format("STATS TO SEND TO API SERVER: ~p~n",[StatsToSend]),
             nerl_tools:http_router_request(RouterHost,RouterPort, [?API_SERVER_ATOM], ActionStr, list_to_binary(StatsToSend)), % update the source with its data
             ets:update_element(StatsEts, counter_received_stats, {?STATS_KEYVAL_VAL_IDX, 0});
 
@@ -194,6 +196,7 @@ handle_cast({sourceDone,Body}, State = #main_genserver_state{myName = MyName, so
   UpdatedSourcesCastingList = SourcesCastingList--[SourceName],
   case UpdatedSourcesCastingList of
     [] -> % the list is empty - all sources were done casting their batches
+      ?LOG_NOTICE("[MAIN-Server] All sources finished casting"),
       PhaseAtom = clientIdle,
       update_clients_phase(PhaseAtom, MyName),
       ListOfClients = ets:lookup_element(get(main_server_ets), clients_names_list, ?DATA_IDX),
@@ -349,20 +352,21 @@ update_clients_phase(PhaseAtom, MessageBody) when is_atom(PhaseAtom) ->
 % Sends requests for statisics from all entities excludes main server
 statistics_requests_to_entities() ->
   ListOfEntities = ets:lookup_element(get(main_server_ets), entities_names_list, ?DATA_IDX),
+  % remove mainServer from the list
+  %DestinationsList = lists:map(fun({Entity , CommTuple}) when Entity =/= ?MAIN_SERVER_ATOM -> {Entity, CommTuple} end, ListOfEntities),
   {RouterHost,RouterPort} = ets:lookup_element(get(main_server_ets), my_router, ?DATA_IDX),
   ActionStr = atom_to_list(statistics),
-  DestinationsList = ListOfEntities,
   MessageBody = "", % there is no need for body in statistics request
-  nerl_tools:http_router_request(RouterHost, RouterPort, DestinationsList, ActionStr, MessageBody).
+  nerl_tools:http_router_request(RouterHost, RouterPort, ListOfEntities, ActionStr, MessageBody).
 
-generate_stats_ets_tables(VerticesList) ->
+generate_stats_ets_tables(EntitiesNamesList) ->
   MainServerEtsStats = get(etsStats),
   Func = 
-    fun({Name, {_Host, _Port, _DeviceName}}) ->
+    fun(Name) ->
         EntityStatsEts = stats:generate_stats_ets(),
         ets:insert(MainServerEtsStats, {Name, EntityStatsEts})
     end,
-  lists:foreach(Func, VerticesList).
+  lists:foreach(Func, EntitiesNamesList).
 
 get_entity_stats_ets(EntityName) ->
   MainServerEtsStats = get(etsStats),

src_erl/NerlnetApp/src/Router/routerGenserver.erl

@@ -54,15 +54,16 @@ init({MyName , _Policy , NerlnetGraph}) -> %% TODO : Add policy to router
   {ok, #router_genserver_state{msgCounter = 1, myName = MyName, etsRef=RoutingTableEtsRef}}.
 
 handle_cast({statistics , _Body} , State=#router_genserver_state{etsRef = Routing_table}) ->
-
+  io:format("~p Got statistics request ~n" , [get(myName)]),
   RouterStatsEts = get(router_stats_ets),
   stats:increment_messages_received(RouterStatsEts),
 
   MyName = get(myName),
   StatsEtsStr = stats:encode_ets_to_http_bin_str(RouterStatsEts),
-  StatisticsBody = {term_to_binary(MyName) , list_to_binary(StatsEtsStr)}, % old data
-  [{_Dest,{_Name , RouterHost , RouterPort}}] = ets:lookup(Routing_table , ?MAIN_SERVER_ATOM),
-  nerl_tools:http_router_request(RouterHost, RouterPort, [?MAIN_SERVER_ATOM], atom_to_list(statistics), StatisticsBody),
+  StatisticsBody = {MyName , StatsEtsStr},
+  [{_Dest,{_Name , MyRouterAddress , MyRouterPort}}] = ets:lookup(Routing_table , MyName), % Router is always owned by itself
+  io:format("~p is sending statistics~n" , [get(myName)]),
+  nerl_tools:http_router_request(MyRouterAddress, MyRouterPort, [?MAIN_SERVER_ATOM], atom_to_list(statistics), StatisticsBody),
   stats:increment_messages_sent(RouterStatsEts),
   {noreply , State};
 
@@ -91,18 +92,20 @@ handle_cast({unicast,{Dest,Body}}, State = #router_genserver_state{msgCounter =
   stats:increment_messages_sent(RouterStatsEts),
   {noreply, State#router_genserver_state{msgCounter = MsgCounter+1,etsRef=Routing_table }};
 
-handle_cast({broadcast,{DestList,Body}}, State = #router_genserver_state{etsRef=Routing_table }) ->
+handle_cast({broadcast,{DestList,Body}}, State = #router_genserver_state{etsRef=Routing_table , myName = _MyName}) ->
+  io:format("@ROUTER BOROADCAST  , Body: ~p~n" , [Body]),
   RouterStatsEts = get(router_stats_ets),
+  %%Destinations = [Dest || Dest <- DestList, Dest =/= get(myName)],
   stats:increment_messages_received(RouterStatsEts),
   MapFunc=fun(Dest,Acc)->
     %make a map when keys are addreses to send a message to, and values are lists of destination of the message that go throu key addres
-    [{Dest,{Name,Host,Port}}]=ets:lookup(Routing_table,Dest),
+    [{Dest,{Name,Host,Port}}] = ets:lookup(Routing_table,Dest),
     case maps:is_key({Name,Host,Port},Acc) of
       true->
-        %addres alread in, append Dest to exsisting value 
+        %address already in, append Dest to existing value 
         NewVal=maps:get({Name,Host,Port},Acc)++[Dest];
       false->
-        %addres not in yet, create new value for it
+        %address not in yet, create new value for it
         NewVal=[Dest]
     end,
     maps:put({Name,Host,Port},NewVal,Acc)

src_erl/NerlnetApp/src/Source/sourceStatem.erl

@@ -161,8 +161,8 @@ idle(cast, {startCasting,Body}, State = #source_statem_state{batchesList = Batch
   {next_state, castingData, State#source_statem_state{transmitter_pid = TransmitterPID}};
 
 idle(cast, {startCasting}, State) ->
+  ?LOG_ERROR("Should not get start casting in idle - start casting is only after setting the phase type"),
   EtsRef = get(source_ets),
-  % io:format("im not suppose to be here"),
   StatsEtsRef = get(source_stats_ets),
   stats:increment_messages_received(StatsEtsRef),
 
@@ -183,7 +183,7 @@ idle(cast, {statistics}, State) ->
 
   MyName = ets:lookup_element(EtsRef, my_name, ?DATA_IDX),
   StatsEtsStr = stats:encode_ets_to_http_bin_str(StatsEtsRef),
-  StatisticsBody = {term_to_binary(MyName) , list_to_binary(StatsEtsStr)}, % old data
+  StatisticsBody = {MyName , StatsEtsStr}, 
   {RouterHost,RouterPort} = ets:lookup_element(EtsRef, my_router, ?DATA_IDX),
   nerl_tools:http_router_request(RouterHost, RouterPort, [?MAIN_SERVER_ATOM], atom_to_list(statistics), StatisticsBody),
   stats:increment_messages_sent(StatsEtsRef),

src_erl/NerlnetApp/src/Stats/stats.erl

@@ -10,7 +10,7 @@
 -export([get_bytes_sent/1, increment_bytes_sent/2]).
 -export([get_bad_messages/1, increment_bad_messages/1]).
 -export([get_value/2, increment_by_value/3]).
--export([encode_ets_to_http_bin_str/1 , decode_http_bin_str_to_ets/1]).
+-export([encode_ets_to_http_bin_str/1 , decode_http_bin_str_to_ets/1 , encode_workers_ets_to_http_bin_str/1]).
 -export([update_workers_ets/4, increment_workers_ets/4 , generate_workers_stats_ets/0]).
 
 get_numeric_type(Value) ->
@@ -28,14 +28,27 @@ encode_ets_to_http_bin_str(StatsEts) ->
                 KeyStr = lists:flatten(io_lib:format("~p" , [Key])),
                 ValueStr = lists:flatten(io_lib:format("~p" , [Value])),
                 TypeStr = lists:flatten(io_lib:format("~p" , [Type])),
+                %% StatsKey:StatsValue:StatsType#StatsKey:StatsValue:StatsType#...
                 KeyStr ++ ?SEPERATOR_WITHIN_TRIPLET ++ ValueStr ++ ?SEPERATOR_WITHIN_TRIPLET ++ TypeStr ++ ?SEPERATOR_TRIPLETS
         end,
     lists:flatten(lists:map(Func , StatsList)).
 
+encode_workers_ets_to_http_bin_str(StatsEts) -> 
+    %% Takes value from ets and converts it to "<EntitiyName1>SEPERATOR<Value1 <EntityName2>SEPERATOR<Value2>" string.
+    StatsList = ets:tab2list(StatsEts),
+    Func = fun({Key , Value}) ->
+                Type = get_numeric_type(Value),
+                KeyStr = lists:flatten(io_lib:format("~p" , [Key])),
+                ValueStr = lists:flatten(io_lib:format("~p" , [Value])),
+                TypeStr = lists:flatten(io_lib:format("~p" , [Type])),
+                %% StatsKey:StatsValue:StatsType#StatsKey:StatsValue:StatsType#...
+                KeyStr ++ ?WORKER_SEPERATOR_WITHIN_TRIPLET ++ ValueStr ++ ?WORKER_SEPERATOR_WITHIN_TRIPLET ++ TypeStr ++ ?WORKER_SEPERATOR_TRIPLETS
+        end,
+    lists:flatten(lists:map(Func , StatsList)).
+
 decode_http_bin_str_to_ets(EncodedStr) -> 
     ReturnedEts = ets:new(ets_to_merge , [set]),
     KeyValTypeTokens = string:tokens(EncodedStr , ?SEPERATOR_TRIPLETS),
-    io:format("KeyValTypeTokens: ~p~n" , [KeyValTypeTokens]),
     Func = fun(Triplet) ->
                 [Key , ValueStr , Type] = string:tokens(Triplet , ?SEPERATOR_WITHIN_TRIPLET),
                 TypeAtom = list_to_atom(Type),
@@ -50,7 +63,7 @@ decode_http_bin_str_to_ets(EncodedStr) ->
     ReturnedEts.
 
 
-generate_stats_ets() ->
+generate_stats_ets() -> %% clients , routers , mainserver...
     StatsEts = ets:new(stats_ets , [set]),
     ets:insert(StatsEts, {messages_received , 0}),
     ets:insert(StatsEts, {messages_sent , 0}),
@@ -63,7 +76,7 @@ generate_stats_ets() ->
     ets:insert(StatsEts, {batches_sent , 0}), % related with source
     StatsEts.
 
-generate_workers_stats_ets() ->
+generate_workers_stats_ets() -> %% workers..
     WorkersStatsEts = ets:new(workers_ets , [set, public]),
     ets:insert(WorkersStatsEts, {bytes_received , 0}),
     ets:insert(WorkersStatsEts, {bytes_sent , 0}),

src_erl/NerlnetApp/src/Stats/stats.hrl

@@ -15,4 +15,8 @@
 -define(SEPERATOR_WITHIN_TRIPLET , ":").
 
 -define(API_SERVER_ENTITY_SEPERATOR , "|").
--define(API_SERVER_WITHIN_ENTITY_SEPERATOR , "&").
+-define(API_SERVER_WITHIN_ENTITY_SEPERATOR , "&").
+
+-define(WORKER_SEPERATOR , "^").
+-define(WORKER_SEPERATOR_TRIPLETS , "@").
+-define(WORKER_SEPERATOR_WITHIN_TRIPLET , "$").

src_erl/NerlnetApp/src/nerl_tools.erl

@@ -127,7 +127,7 @@ read_all_data(Req0, Got) ->
 
 deleteOldJson(FilePath) ->
   try   file:delete(FilePath)
-  catch {error, E} -> io:format("couldn't delete file ~p, beacuse ~p~n",[FilePath, E])
+  catch {error, E} -> ?LOG_ERROR("couldn't delete file ~p, beacuse ~p~n",[FilePath, E])
   end.
 
 % get this host ip 
@@ -208,13 +208,18 @@ calculate_size(List) when is_list(List) ->
 
 %% TODO: create create_body func for standard message passing
 
-make_routing_table(Ets,EntitiesList,Origin,NerlnetGraph)->
+%% Entities List must not contain ThisRouter
+%% NextHop is the next router for all entities that are not connected with ThisRouter
+%% Otherwise , NextHop is an entity of ThisRouter
+make_routing_table(Ets,EntitiesList,ThisRouter,NerlnetGraph)->
   GenerateTablesFunc = fun(Entity) -> 
-  case digraph:get_short_path(NerlnetGraph,Origin,Entity) of
-    false -> ok;
-    ShortPath -> NextHop = lists:nth(2,ShortPath),
-                 {Name , {Host , Port , _DeviceName}} = digraph:vertex(NerlnetGraph,NextHop),
-                 ets:insert(Ets,{Entity,{Name,Host,Port}})
-  end % case end
+    case digraph:get_short_path(NerlnetGraph,ThisRouter,Entity) of
+      false -> ok;
+      ShortPath -> NextHop = lists:nth(2,ShortPath),
+                  {Name , {Host , Port , _DeviceName}} = digraph:vertex(NerlnetGraph,NextHop),
+                  ets:insert(Ets,{Entity,{Name,Host,Port}})
+    end % case end
   end, % fun end
-  lists:foreach(GenerateTablesFunc, EntitiesList).
+  lists:foreach(GenerateTablesFunc, EntitiesList),
+  {ThisRouter , {RouterHost , RouterPort , _DeviceName}} = digraph:vertex(NerlnetGraph , ThisRouter),
+  ets:insert(Ets , {ThisRouter , {ThisRouter , RouterHost , RouterPort}}).