Refactor

src/derflow.erl

@@ -2,17 +2,17 @@
 -module(derflow).
 -behaviour(application).
 -export([start/2,
-	 stop/1,
-	 bind/2,
-	 bind/3,
-	 read/1,
-	 declare/0,
-	 thread/3,
-	 get_stream/1,
-	 byNeed/2,
-	 byNeed/3,
-	 waitNeeded/1,
-	 async_print_stream/1]).
+     stop/1,
+     bind/2,
+     bind/3,
+     read/1,
+     declare/0,
+     thread/3,
+     get_stream/1,
+     byNeed/2,
+     byNeed/3,
+     waitNeeded/1,
+     async_print_stream/1]).
 start(normal, _Args) ->
     derflow_sup:start_link().
 
@@ -49,17 +49,17 @@ get_stream(Stream)->
 async_print_stream(Stream)->
     io:format("Stream: ~w~n", [Stream]),
     case read(Stream) of
-	{nil, _} -> {ok, stream_read};
-	{Value, Next} -> 
-	    io:format("Value of stream: ~w~n",[Value]),
-	    async_print_stream(Next)
+    {nil, _} -> {ok, stream_read};
+    {Value, Next} -> 
+        io:format("Value of stream: ~w~n",[Value]),
+        async_print_stream(Next)
     end.
 
 %Internal functions
 
 internal_get_stream(Head, Output) ->
     case read(Head) of
-	{nil, _} -> Output;
-	{Value, Next} -> 
-	    internal_get_stream(Next, lists:append(Output, [Value]))
+    {nil, _} -> Output;
+    {Value, Next} -> 
+        internal_get_stream(Next, lists:append(Output, [Value]))
     end.

src/derflow_server.erl

@@ -8,40 +8,68 @@
 -record(state, {clock}).
 -record(dv, {value, next, waitingThreads = [], creator, lazy=false,bounded = false}). 
 
+
+%%% API functions
+
+
 start_link() ->
     io:format("Server running~n"),
     gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
 
 init(_Args) ->
-	io:format("Init called~n"),
-    	ets:new(dvstore, [set, named_table, public, {write_concurrency, true}]),
-    	{ok, #state{clock=0}}.
+    io:format("Init called~n"),
+        ets:new(dvstore, [set, named_table, public, {write_concurrency, true}]),
+        {ok, #state{clock=0}}.
+
 
+% declare a new variable
+% returns the new variable's Id, an Erlang term
 declare() ->
     gen_server:call(?MODULE, {declare}).
 
 
+% bind a function result to a variable
+% Id: the variable id provided by declare/0
+% F: the function to execute
+% Arg: function arguments (TODO: what should they look like? why not just put in a fun)
 bind(Id, F, Arg) ->
     gen_server:call(?MODULE, {bind, Id, F, Arg}).
 
+
+% bind the variable Id to Value
+% Id: the variable id provided by declare/0
+% Value: an erlang term
 bind(Id, Value) ->
     gen_server:call(?MODULE, {bind, Id, Value}).
 
+
+% registers the calling process as wanting a notification when variable Id is bound
+% Id: the variable id provided by declare/0
 waitNeeded(Id) ->
     gen_server:call(?MODULE, {waitNeeded, Id}).
 
-
+% TODO: what's this for
+% TODO: unimplemented?
 wait(X) ->
     gen_server:call(?MODULE, {wait, X}).
 
+
+% read dataflow variable X; block until bound if not yet bound
+% TODO: is X the same as Id from declare ? rename?
 read(X) ->
     gen_server:call(?MODULE, {read, X}).
 
+
+%%% end API functions
+
+
+
+
 handle_call({declare}, _From, State) ->
-    	Clock = State#state.clock +1,
-    	V = #dv{value=empty, next=empty},
-	ets:insert(dvstore, {Clock, V}),
-   	{reply, {id, Clock}, State#state{clock=Clock}};
+        Clock = State#state.clock +1,
+        V = #dv{value=empty, next=empty},
+    ets:insert(dvstore, {Clock, V}),
+    {reply, {id, Clock}, State#state{clock=Clock}};
 
 handle_call({next, Id}, _From, State)->
     io:format("Requesting for next~w~n",[Id]),
@@ -70,35 +98,37 @@ handle_call({waitNeeded, Id}, From, State) ->
     case V#dv.waitingThreads of [_H|_T] ->
         {reply, ok, State};
         _ ->
-	ets:insert(dvstore, {Id, V#dv{lazy=true, creator=From}}),
+    ets:insert(dvstore, {Id, V#dv{lazy=true, creator=From}}),
         {noreply, State}
     end;
 
-%%%What if the Key does not exist in the map?%%%
+%%% TODO: What if the Key does not exist in the map?%%%
 handle_call({read,X}, From, State) ->
-	[{_Key,V}] = ets:lookup(dvstore, X),
-        Value = V#dv.value,
-	Bounded = V#dv.bounded,
-	Creator = V#dv.creator,
-	Lazy = V#dv.lazy,
-	%%%Need to distinguish that value is not calculated or is the end of a list%%%
-	if Bounded == true ->
-	  {reply, {Value, V#dv.next}, State};
-	 true ->
-	  if Lazy == true ->
-		WT = lists:append(V#dv.waitingThreads, [From]),
+    [{_Key,V}] = ets:lookup(dvstore, X),
+    Value = V#dv.value,
+    Bounded = V#dv.bounded,
+    Creator = V#dv.creator,
+    Lazy = V#dv.lazy,
+    %%% TODO: Need to distinguish that value is not calculated or is the end of a list%%%
+    if Bounded == true ->
+      {reply, {Value, V#dv.next}, State};
+    true ->
+        if Lazy == true ->
+            WT = lists:append(V#dv.waitingThreads, [From]),
                 V1 = V#dv{waitingThreads=WT},
                 ets:insert(dvstore, {X, V1}),
-	  	gen_server:reply(Creator, ok),
-		{noreply, State};
-	  true ->
-	  	WT = lists:append(V#dv.waitingThreads, [From]),
-	  	V1 = V#dv{waitingThreads=WT},
-	  	ets:insert(dvstore, {X, V1}),
-          	{noreply, State}
-	  end
-	end;
+                gen_server:reply(Creator, ok),
+                {noreply, State};
+        true ->
+            WT = lists:append(V#dv.waitingThreads, [From]),
+            V1 = V#dv{waitingThreads=WT},
+            ets:insert(dvstore, {X, V1}),
+            {noreply, State}
+        end
+    end;
+
 
+% TODO: unimplemented?
 handle_call({wait, _X}, _From, State) ->
    %_V = wait_for_value(X, State#state.kv),
    {reply, {ok}, State}.
@@ -108,43 +138,43 @@ handle_cast({_}, State) ->
 
 
 %putLazy(Value, Next, Key, From) ->
-%	%io:format("Put lazy ~w~n",[Key]),
-%	V1 = #dv{value= Value, next =Next, bounded= true, lazy = true, creator = From},
-%	ets:insert(dvstore, {Key, V1}).
+%   %io:format("Put lazy ~w~n",[Key]),
+%   V1 = #dv{value= Value, next =Next, bounded= true, lazy = true, creator = From},
+%   ets:insert(dvstore, {Key, V1}).
 
 %Bind the key with the result (either by assignment or calculating)
 %executeLazy(Key, V) ->
-%	%io:format("Execute lazy ~w~n",[Key]),
-%	Value= V#dv.value,
-%	case Value of {F, Arg} ->
-%	   Result = F(Arg);
-%	   _ ->
-%	   Result = Value
-%	end,
-%	V1 = V#dv{value= Result, lazy = false, waitingThreads= []},
-%	ets:insert(dvstore, {Key, V1}),
-%	Result.
+%   %io:format("Execute lazy ~w~n",[Key]),
+%   Value= V#dv.value,
+%   case Value of {F, Arg} ->
+%      Result = F(Arg);
+%      _ ->
+%      Result = Value
+%   end,
+%   V1 = V#dv{value= Result, lazy = false, waitingThreads= []},
+%   ets:insert(dvstore, {Key, V1}),
+%   Result.
 
 put(Value, Next, Key) ->
-	[{_Key,V}] = ets:lookup(dvstore, Key),
-	Threads = V#dv.waitingThreads,
-	V1 = #dv{value= Value, next =Next, bounded= true,lazy=false},
-	ets:insert(dvstore, {Key, V1}),
-	replyToAll(Threads, Value, Next).
+    [{_Key,V}] = ets:lookup(dvstore, Key),
+    Threads = V#dv.waitingThreads,
+    V1 = #dv{value= Value, next =Next, bounded= true,lazy=false},
+    ets:insert(dvstore, {Key, V1}),
+    replyToAll(Threads, Value, Next).
 
 execute_and_put(F, Arg, Next, Key) ->
-	[{_Key,V}] = ets:lookup(dvstore, Key),
-	Threads = V#dv.waitingThreads,
-	Value = F(Arg),
-	V1 = #dv{value= Value, next =Next, bounded= true, lazy=false},
-	ets:insert(dvstore, {Key, V1}),
-	replyToAll(Threads, Value, Next).
+    [{_Key,V}] = ets:lookup(dvstore, Key),
+    Threads = V#dv.waitingThreads,
+    Value = F(Arg),
+    V1 = #dv{value= Value, next =Next, bounded= true, lazy=false},
+    ets:insert(dvstore, {Key, V1}),
+    replyToAll(Threads, Value, Next).
 
 replyToAll([], _Value, _Next) ->
-	ok;
+    ok;
 replyToAll([H|T], Value, Next) ->
-	gen_server:reply(H,{Value,Next}),
-	replyToAll(T, Value, Next).
+    gen_server:reply(H,{Value,Next}),
+    replyToAll(T, Value, Next).
 
 code_change(_, State, _) ->
     {ok, State}.
@@ -158,9 +188,9 @@ terminate(normal, _State) ->
 %wait_for_value(X, KV)->
 %       IsKey = derflow_map:is_key(X, KV),
 %       if IsKey == true ->
-%	 derflow_map:get(X, KV);
-%	true ->
-%    	  waitTime(500),
+%    derflow_map:get(X, KV);
+%   true ->
+%         waitTime(500),
 %          wait_for_value(X, KV)
 %       end.
 %

test/bounded_buffer.erl

@@ -16,8 +16,8 @@ test2() ->
 producer(Init, N, Output) ->
     if (N>0) ->
         derflow:waitNeeded(Output),
-	{id,Next} = derflow:bind(Output,Init),
-	io:format("Prod:Bound for ~w next ~w Produced~w ~n",[Output, Next, 11-N]),
+  {id,Next} = derflow:bind(Output,Init),
+  io:format("Prod:Bound for ~w next ~w Produced~w ~n",[Output, Next, 11-N]),
         producer(Init + 1, N-1,  Next);
     true ->
         derflow:bind(Output, nil)
@@ -29,8 +29,8 @@ loop(S1, S2, End) ->
     {id, S2Next} = derflow:bind(S2, Value1),
     io:format("Buff:Bound for consumer ~w-> ~w ~w~n",[S1,S2,Value1]),
     case derflow:read(End) of {nil, _} ->
-	loop(S1Next, S2Next, End);
-	{_V,EndNext} ->
+  loop(S1Next, S2Next, End);
+  {_V,EndNext} ->
        loop(S1Next, S2Next, EndNext)    
     end.
 
@@ -41,20 +41,20 @@ buffer(S1, Size, S2) ->
 
 drop_list(S, Size) ->
     if Size == 0 ->
-	S;
+  S;
       true ->
-       	{_Value,Next}=derflow:read(S),
-    	drop_list(Next, Size-1)
+        {_Value,Next}=derflow:read(S),
+      drop_list(Next, Size-1)
     end.
 
 consumer(S2,F) ->
     case derflow:read(S2) of
-	{nil, _} ->
-	   io:format("Cons:Reading end~n");
-	{Value, Next} ->
-	   io:format("Cons:Id ~w Consume ~w, Get ~w, Next~w ~n",[S2,Value, F(Value),Next]),
-	   %timer:sleep(1000),
-	   consumer(Next, F)
+  {nil, _} ->
+     io:format("Cons:Reading end~n");
+  {Value, Next} ->
+     io:format("Cons:Id ~w Consume ~w, Get ~w, Next~w ~n",[S2,Value, F(Value),Next]),
+     %timer:sleep(1000),
+     consumer(Next, F)
     end.
 
 

test/prod_cons.erl

@@ -2,30 +2,33 @@
 -export([test1/0, producer/3, consumer/3]).
 
 test1() ->
-    {id, S1}=derflow:declare(),
-    derflow:thread(prod_cons,producer,[0,10,S1]),
-    {id, S2}=derflow:declare(),
-    derflow:thread(prod_cons,consumer,[S1,fun(X) -> X + 5 end,S2]),
-    derflow:async_print_stream(S2).
-    %L = derflow:get_stream(S2),
+    % create a variable Stream1 and have a producer write to it
+    {id, Stream1}=derflow:declare(),
+    derflow:thread(prod_cons,producer,[0,10,Stream1]),
+
+    % create a variable Stream2 and write transformations of values read from Stream1
+    {id, Stream2}=derflow:declare(),
+    derflow:thread(prod_cons,consumer,[Stream1,fun(X) -> X + 5 end,Stream2]),
+    derflow:async_print_stream(Stream2).
+    %L = derflow:get_stream(Stream2),
     %L.
     %io:format("Output: ~w~n",[L]).
-    %S2.
+    %Stream2.
 
 producer(Init, N, Output) ->
     if (N>0) ->
-	timer:sleep(1000),
-	{id, Next} = derflow:bind(Output, Init),
-	producer(Init + 1, N-1,  Next);
+    timer:sleep(1000),
+    {id, Next} = derflow:bind(Output, Init),
+    producer(Init + 1, N-1,  Next);
     true ->
-	derflow:bind(Output, nil)
+    derflow:bind(Output, nil)
     end.
 
-consumer(S1, F, S2) ->
-    case derflow:read(S1) of
-	{nil, _} ->
-	    derflow:bind(S2, nil);
-	{Value, Next} ->
-	    {id, NextOutput} = derflow:bind(S2, F, Value),
-	    consumer(Next, F, NextOutput)
+consumer(InStream, F, OutStream) ->
+    case derflow:read(InStream) of
+    {nil, _} ->
+        derflow:bind(OutStream, nil);
+    {Value, Next} ->
+        {id, NextOutput} = derflow:bind(OutStream, F, Value),
+        consumer(Next, F, NextOutput)
     end.