Merge pull request #400 from leondavi/source

improve compression of data in sources add notice of receiveing data …

Author: leondavi

src_erl/NerlnetApp/src/Source/castingHandler.erl

@@ -18,9 +18,8 @@ init(Req0, [Action,Source_StateM_Pid]) ->
 %%  io:format("casting handler got Body:~p~n",[Body]),
   case Action of
     csv ->  {_ , Body} = nerl_tools:read_all_data(Req0 , <<>>),
-            {WorkersList, Phase, NumOfBatches, NerlTensorType, Data} = binary_to_term(Body),
-            UncompressedData = binary_to_list(zlib:uncompress(Data)),
-            gen_statem:cast(Source_StateM_Pid, {batchList, WorkersList, list_to_atom(Phase), list_to_integer(NumOfBatches), NerlTensorType , UncompressedData});
+            {WorkersList, Phase, NumOfBatches, NerlTensorType, CompressedData} = binary_to_term(Body),
+            gen_statem:cast(Source_StateM_Pid, {batchList, WorkersList, list_to_atom(Phase), list_to_integer(NumOfBatches), NerlTensorType , CompressedData});
     startCasting  ->  {_,Body,_} = cowboy_req:read_body(Req0),
                       gen_statem:cast(Source_StateM_Pid, {startCasting,Body});
     statistics    ->  gen_statem:cast(Source_StateM_Pid, {statistics});

src_erl/NerlnetApp/src/Source/parser.erl

@@ -105,7 +105,8 @@ dataStrToNumericParallelLoop(PF, EtsTable, ListOfLinesOfData, ErlType, LastKey)
 dataStrToNumericData(ListOfLinesOfData , ErlType)->
   EtsTable = ets:new(data_str_to_numeric_data, [ordered_set, public]),
   dataStrToNumericParallelLoop(?PARALLELIZATION_FACTOR, EtsTable, ListOfLinesOfData, ErlType, 0),
-  [ element(?DATA_IDX, Attribute) || Attribute <- ets:tab2list(EtsTable)].
+  NumericDataList = [ element(?DATA_IDX, Attribute) || Attribute <- ets:tab2list(EtsTable)],
+  ets:delete(EtsTable), NumericDataList.
 
 generateListOfBatches(ListOfList, BatchSize) -> generateListOfBatches(ListOfList, BatchSize, []).
 

src_erl/NerlnetApp/src/Source/sourceStatem.erl

@@ -117,12 +117,14 @@ state_name(_EventType, _EventContent, State = #source_statem_state{}) ->
 
 
 %% This cast receive a list of samples to load to the records batchList
-idle(cast, {batchList, WorkersList, Phase, NumOfBatches, NerlTensorType, Data}, State) ->
+idle(cast, {batchList, WorkersList, Phase, NumOfBatches, NerlTensorType, CompressedData}, State) ->
   EtsRef = get(source_ets),
   StatsEtsRef = get(source_stats_ets),
   MyName = ets:lookup_element(EtsRef, my_name, ?DATA_IDX),
   BatchSize = ets:lookup_element(EtsRef, batch_size, ?DATA_IDX),
-  {NerlTensorBatchesList, SampleSize} = parser:parseCSV(MyName, BatchSize, NerlTensorType, Data), % TODO this is slow and heavy policy! pre parse in ETS a possible solution
+  ?LOG_NOTICE("Source ~p, Receiving and parsing data", [MyName]),
+  UncompressedData = binary_to_list(zlib:uncompress(CompressedData)),
+  {NerlTensorBatchesList, SampleSize} = parser:parseCSV(MyName, BatchSize, NerlTensorType, UncompressedData), % TODO this is slow and heavy policy! pre parse in ETS a possible solution
   ets:update_element(EtsRef, workers_list, [{?DATA_IDX, WorkersList}]),
   ets:update_element(EtsRef, num_of_batches, [{?DATA_IDX, NumOfBatches}]),
   ets:update_element(EtsRef, current_phase, [{?DATA_IDX, Phase}]),