Changed flow_graph representation.

include/flow_graph.hrl

@@ -0,0 +1,62 @@
+% operation A B = A |->| B
+% task B = operation () B = () |->| B
+
+% operation ::= func  | value    | dynamic  | route
+%             | pipe  | parallel | sequence
+%             | split | merge    | switch   | loop
+
+% func (A_1 * ... * A_n) B ::=
+%   { in = n : number, code : fun(A_1, ..., A_n) -> B }
+-record(func, {in, code}).
+
+% value () B ::= { value : B }
+-record(value, {value}).
+
+% dynamic (A_1 * ... * A_n) (operation () B) ::=
+%   { in = n : number, code : fun(A_1, ..., A_n) -> operation () B }
+-record(dynamic, {in, code}).
+
+% route_R A (route_map R A) ::=
+%   { in = size(A) : number, out = length(R) : number,
+%     map = R : [ (1..size(A)) ] }
+-record(route, {in, out, map}).
+
+% pipe (A_1 * ... * A_n) (B_1 * ... * B_m) ::=
+%   { in = n : number, out = m : number,
+%     ops : [ operation (A_1_1 * ... * A_1_n1) (A_2_1 * ... * A_2_n2),
+%             operation (A_2_1 * ... * A_2_n2) (A_3_1 * ... * A_3_n3),
+%             ...,
+%             operation (A_i_1 * ... * A_i_ni) (B_1 * ... * B_m) ] }
+-record(pipe, {in, out, ops}).
+
+% parallel (flatten([A_1, ..., A_n])) (flatten([B_1, ..., B_n])) ::=
+%   { in = size(flatten([A_1, ..., A_n])) : number,
+%     out = size(flatten([B_1, ..., B_n])) : number,
+%     ops : [ operation A_1 B_1, ..., operation A_n B_n ] }
+-record(parallel, {in, out, ops}).
+
+% sequence (flatten([A_1, ..., A_n])) (flatten([B_1, ..., B_n])) ::=
+%   { in = size(flatten([A_1, ..., A_n])) : number,
+%     out = size(flatten([B_1, ..., B_n])) : number,
+%     ops : [ operation A_1 B_1, ..., operation A_n B_n ] }
+-record(sequence, {in, out, ops}).
+
+% split (unit (A_1 * ... * A_n)) (A_1 * ... * A_n) ::= { size = n : number }
+-record(split, {size}).
+
+% merge (A_1 * ... * A_n) (unit (A_1 * ... * A_n)) ::= { size = n : number }
+-record(merge, {size}).
+
+% switch (A_1 * ... * A_n) (B_1 + ... + B_m) ::=
+%   { in = n : number, out = m : number,
+%     switch : operation (A_1, ..., A_n) (Tag_1 + ... + Tag_m),
+%     map    : [ {Tag_1, operation (A_1 * ... * A_n) B_1},
+%                ...,
+%                {Tag_m, operation (A_1 * ... * A_n) B_m} ] }
+-record(switch, {in, out, switch, map}).
+
+% loop (A_1 * ... * A_n) (B_1 * ... * B_m) ::=
+%   { in = n : number, out = m : number,
+%     size : number,
+%     op   : operation (A_1 * ... * A_n+size) (B_1 * ... * B_m+size) }
+-record(loop, {in, out, size, op}).

src/flow_graph.erl

@@ -1,201 +1,149 @@
 -module(flow_graph).
-
-% Vertex construction
--export([hole/0, value/1, void/0, split/0, merge/0, switch/2, select/2]).
-% Edge construction
--export([flow/2, func/3, index/3, range/4, tagged/3, dynamic/2, dynamic/3]).
-% Graph construction
--export([singleton/1, from_list/3]).
-% Graph queries
--export([get_in/1, get_out/1, out_edges/2]).
-% Graph manipulation
--export([append/2, append/1, add_edge/2, add_edge_list/2]).
-% Checks for validity
--export([valid_edge/1, valid_edge_list/1, valid_graph/1]).
-% Branching
--export([detour/3, reroute/2]).
-
-% Use the orddict module for clarity, and the dict module for speed.
--define(DICT, orddict).
-
-% flow_graph ::= {in    : vertex,
-%                 out   : vertex,
-%                 graph : dict(vertex, [edge])}
--record(flow_graph, {in, out, graph}).
-
-% vertex ::= {ref, node}
--record(vertex, {ref, node}).
-
-% node ::= hole | value | void | split | merge | switch | select | delayed
-% value ::= term
-% switch ::= {fun(A, State) -> {Tag, State}, State}
-% select ::= {fun(A, State) -> {select, State} | {block, State}, State}
--record(value, {value}).
--record(switch, {switch, state}).
--record(select, {select, state}).
-
-% edge ::= {from : vertex, to : vertex, through : link
--record(edge, {from, to, through}).
-
-% link ::= flow | func | range | tagged | dynamic
-% func ::= fun(A) -> B
-% range ::= {int, int}
-% tagged ::= Tag
-% dyanmic ::= fun(A) -> flow_graph
--record(func, {func}).
--record(range, {from, to}).
--record(tagged, {tag}).
--record(dynamic, {code}).
-
-%% Vertex construction
-make_vertex(N) ->
-    #vertex{ref=make_ref(), node=N}.
-
-hole() ->
-    make_vertex(hole).
-
-value(X) ->
-    make_vertex(#value{value=X}).
-
-void() ->
-    make_vertex(void).
-
-split() ->
-    make_vertex(split).
-
-merge() ->
-    make_vertex(merge).
-
-switch(Switch, Init) when is_function(Switch) ->
-    make_vertex(#switch{switch=Switch, state=Init}).
-
-select(Select, Init) when is_function(Select) ->
-    make_vertex(#select{select=Select, state=Init}).
-
-delayed() ->
-    make_vertex(delayed).
-
-%% Edge construction
-
-make_edge(From, To, Through) ->
-    #edge{from=From, to=To, through=Through}.
-
-flow(From, To) ->
-    make_edge(From, To, flow).
-
-func(From, To, Func) ->
-    make_edge(From, To, #func{func=Func}).
-
-index(From, To, N) when is_number(N) ->
-    range(From, To, N, N).
-
-range(From, To, Low, High) when is_number(Low), is_number(High), Low =< High->
-    make_edge(From, To, #range{from=Low, to=High}).
-
-tagged(From, To, Tag) ->
-    make_edge(From, To, #tagged{tag=Tag}).
-
-dynamic(From, To = #vertex{node=delayed}, Code) ->
-    make_edge(From, To, #dynamic{code=Code}).
-
-dynamic(From, Code) ->
-    make_edge(From, delayed(), #dynamic{code=Code}).
-
-%% Graph constructon
-
-singleton(V = #vertex{}) ->
-    #flow_graph{in=V, out=V, graph=orddict:store(V, [], ?DICT:new())}.
-
-from_list(In = #vertex{}, Out = #vertex{}, Edges) ->
-    FG = lists:foldl(fun(E, G) -> add_edge(G, E) end,
-		     #flow_graph{in=In, out=Out, graph=?DICT:new()},
-		     Edges),
-    FG#flow_graph{graph=add_vertex(FG#flow_graph.graph, Out)}.
-
-%% Graph manipulation
-
-merge_graphs(G1, G2) ->
-    ?DICT:merge(fun(_, E1, E2) -> lists:append(E1,E2) end, G1, G2).
-
-append(#flow_graph{in=In1, out=Out1, graph=G1},
-       #flow_graph{in=In2, out=Out2, graph=G2}) ->
-    add_edge(#flow_graph{in=In1, out=Out2, graph=merge_graphs(G1,G2)},
-	     flow(Out1, In2)).
-
-append([FG1|FGs]) ->
-    lists:foldl(fun(FG, All) -> append(All, FG) end, FG1, FGs).
-
-get_in(#flow_graph{in=In}) -> In.
-
-get_out(#flow_graph{out=Out}) -> Out.
-
-out_edges(#flow_graph{graph=G}, #vertex{ref=R}) ->
-    ?DICT:find(R, G).
-
-add_edge(FG = #flow_graph{graph=G}, E = #edge{from=From, to=To}) ->
-    Add = fun(Edges) -> [E|Edges] end,
-    add_vertex(FG#flow_graph{graph=?DICT:update(From, Add, [E], G)}, To).
-
-add_edge_list(FG = #flow_graph{}, Edges) ->
-    lists:foldl(fun(E, Acc) -> add_edge(Acc, E) end, FG, Edges).
-
-add_vertex(FG = #flow_graph{graph=G}, V = #vertex{}) ->
-    FG#flow_graph{graph=?DICT:update(V, fun(Es)->Es end, [], G)}.
-
-% Checks for validity
-valid_edge(#edge{from=From, to=To, through=Through}) ->
-    IsNeutral = fun(flow)       -> true;
-		   (#func{})    -> true;
-		   (#dynamic{}) -> true;
-		   (_)          -> false
+-include("../include/flow_graph.hrl").
+
+%%% Operation construction
+-export([func/1, value/1, values/1, dynamic/1, route/2,
+	 pipe/1, parallel/1, sequence/1,
+	 split/1, merge/1, switch/2, loop/2]).
+
+%%% Operation querying
+-export([in_arity/1, out_arity/1, can_connect/2]).
+
+%%%-----------------------------------------------------------------------------
+%%% Operation construction
+%%%-----------------------------------------------------------------------------
+
+func(F) when is_function(F) ->
+    {arity, N} = erlang:fun_info(F, arity),
+    #func{in=N, code=F}.
+
+
+value(X) -> #value{value=X}.
+
+values(Xs = [_|_]) ->
+    parallel(lists:map(fun value/1, Xs)).
+
+dynamic(Gen) when is_function(Gen) ->
+    {arity, N} = erlang:fun_info(Gen, arity),
+    #dynamic{in=N, code=Gen}.
+
+route(N, Map) ->
+    try
+	lists:foldl(fun(Target, Count) when Target =< N -> Count+1;
+		       (Target, _Count) when Target > N -> throw(out_of_bounds)
+		    end,
+		    0, Map)
+    of 
+	Count -> #route{in=N, out=Count, map=Map}
+    catch
+	throw:out_of_bounds -> erlang:error(badarg, [N, Map])
+    end.
+
+pipe(Ops = [First|Rest]) ->
+    try 
+	lists:foldl(fun(Op2, Op1) ->
+			       case can_connect(Op1, Op2) of
+				   true  -> Op2;
+				   false -> throw(bad_connection)
+			       end
+		    end,
+		    First, Rest)
+    of
+	Last -> #pipe{in=in_arity(First), out=out_arity(Last), ops=Ops}
+    catch
+	throw:bad_connection -> erlang:error(badarg, [Ops])
+    end.
+
+parallel(Ops = [_|_]) ->
+    {In, Out} = flatten_arity(Ops),
+    #parallel{in=In, out=Out, ops=Ops}.
+
+sequence(Ops = [_|_]) ->
+    {In, Out} = flatten_arity(Ops),
+    #sequence{in=In, out=Out, ops=Ops}.
+
+split(N) -> #split{size=N}.
+
+merge(N) -> #merge{size=N}.
+
+switch(Switch, Map = [{_, Op1}|Rest]) ->
+    CheckIO = fun(Op, In, Out) ->
+		      case {similar_arity(in_arity(Op), In),
+			    similar_arity(out_arity(Op), Out)}
+		      of
+			  {false, _} -> throw(arity_mismatch);
+			  {_, false} -> throw(arity_mismatch);
+			  {I, O}     -> {I, O}
+		      end
+	      end,
+    try
+	{I, O} = lists:foldl(
+		   fun({_, Op}, {In, Out}) -> CheckIO(Op, In, Out) end,
+		   {in_arity(Op1), out_arity(Op1)}, Rest),
+	CheckIO(Switch, I, 1),
+	{I, O}
+    of
+	{In, Out} -> #switch{in=In, out=Out, switch=Switch, map=Map}
+    catch
+	throw:arity_mismatch -> erlang:error(badarg, [Switch, Map])
+    end.
+
+loop(N, Op) ->
+    {In, Out} = case {in_arity(Op)-N, out_arity(Op)-N} of
+		    {I, O} when I >= 0, O > 0 -> {I, O};
+		    {_, _} -> erlang:error(badarg, [N, Op])
 		end,
-    Left = case {From, Through, To} of
-	       {delayed,   _,          _}        -> false;
-	       {hole,      _,          _}        -> IsNeutral(Through);
-	       {#value{},  _,          _}        -> IsNeutral(Through);
-	       {void,      flow,       #value{}} -> true;
-	       {void,      flow,       void}     -> true;
-	       {void,      #dynamic{}, delayed}  -> true;
-	       {void,      _,          _}        -> false;
-	       {split,     #range{},   _}        -> true;
-	       {split,     _,          _}        -> false;
-	       {merge,     _,          _}        -> IsNeutral(Through);
-	       {#switch{}, #tagged{},  _}        -> true;
-	       {#switch{}, _,          _}        -> false;
-	       {#select{}, _,          _}        -> IsNeutral(Through);
-	       _                                 -> false
-	   end,
-    Right = case {From, Through, To} of
-		{_,    dynamic,  delayed}   -> true;
-		{_,    dynamic,  _}         -> false;
-		{_,    _,        hole}      -> true;
-		{void, flow,     #value{}}  -> true;
-		{_,    _,        #value{}}  -> false;
-		{void, flow,     void}      -> true;
-		{_,    _,        void}      -> false;
-		{_,    _,        split}     -> true;
-		{_,    #range{}, merge}     -> true;
-		{_,    _,        merge}     -> false;
-		{_,    _, 	 #switch{}} -> true;
-		{_,    _, 	 #select{}} -> true
-	    end,
-    Left and Right.
-
-valid_edge_list(Edges) ->
-    lists:all(fun valid_edge/1, Edges).
-
-valid_graph(#flow_graph{graph=G}) ->
-    lists:all(fun({_, Es}) -> valid_edge_list(Es) end,
-	      ?DICT:to_list(G)).
-
-% Branching
-detour(From = #flow_graph{out=FOut, graph=FG},
-       _Between = #flow_graph{in=BIn, out=BOut, graph=BG},
-       _To = #flow_graph{in=TIn}) ->
-    G = add_edge(add_edge(merge_graphs(FG,BG), flow(FOut, BIn)),
-		 flow(BOut, TIn)),
-    From#flow_graph{graph=G}.
-
-reroute(#flow_graph{in=In1, graph=G1},
-	#flow_graph{out=Out2, graph=G2}) ->
-    #flow_graph{in=In1, out=Out2, graph=merge_graphs(G1, G2)}.
+    #loop{in=In, out=Out, size=N, op=Op}.
+
+%%%-----------------------------------------------------------------------------
+%%% Operation querying
+%%%-----------------------------------------------------------------------------
+
+in_arity(#func{in=In})     -> In;
+in_arity(#value{})         -> 0;
+in_arity(#dynamic{in=In})  -> In;
+in_arity(#route{in=In})    -> In;
+in_arity(#pipe{in=In})     -> In;
+in_arity(#parallel{in=In}) -> In;
+in_arity(#sequence{in=In}) -> In;
+in_arity(#split{})         -> 1;
+in_arity(#merge{size=N})   -> N;
+in_arity(#switch{in=In})   -> In;
+in_arity(#loop{in=In})     -> In;
+in_arity(Bad)              -> erlang:error(badarg, [Bad]).
+
+out_arity(#func{})            -> 1;
+out_arity(#value{})           -> 1;
+out_arity(#dynamic{})         -> unknown;
+out_arity(#route{out=Out})    -> Out;
+out_arity(#pipe{out=Out})     -> Out;
+out_arity(#parallel{out=Out}) -> Out;
+out_arity(#sequence{out=Out}) -> Out;
+out_arity(#split{size=N})     -> N;
+out_arity(#merge{})           -> 1;
+out_arity(#switch{out=Out})   -> Out;
+out_arity(#loop{out=Out})     -> Out;
+out_arity(Bad)                -> erlang:error(badarg, [Bad]).
+
+flatten_arity([First|Rest]) ->
+    lists:foldl(fun(Op, {In, Out}) ->
+			{add_arity(In, in_arity(Op)),
+			 add_arity(Out, out_arity(Op))}
+	  end,
+	  {in_arity(First), out_arity(First)}, Rest).
+
+add_arity(unknown, _) -> unknown;
+add_arity(_, unknown) -> unknown;
+add_arity(N, M)       -> N + M.
+
+similar_arity(N, N)       -> N;
+similar_arity(unknown, N) -> N;
+similar_arity(N, unknown) -> N;
+similar_arity(_, _)       -> false.
+
+can_connect(Op1, Op2) ->
+    case {out_arity(Op1), in_arity(Op2)} of
+	{N,       N} -> true;
+	{unknown, _} -> true;
+	_            -> false
+    end.