Cleanup timing, error handling, and exec order

rebar.config

@@ -2,7 +2,8 @@
 {erl_opts, [
 %%	bin_opt_info,
 %%	warn_missing_spec,
-	warn_export_all
+	warn_export_all,
+        {platform_define, "18", erlang18}
 ]}.
 {edoc_opts, [{stylesheet_file, "./priv/edoc.css"}]}.
 

src/glc.erl

@@ -71,7 +71,8 @@
     get/2,
     delete/1,
     reset_counters/1,
-    reset_counters/2
+    reset_counters/2,
+    start/0
 ]).
 
 -export([
@@ -91,6 +92,7 @@
 ]).
 
 -export([
+    info/1,
     input/1,
     output/1,
     job_input/1,
@@ -240,6 +242,12 @@ run(Module, Fun, Event) ->
     Module:runjob(Fun, Event).
 
 
+info(Module) ->
+    Counters = [input, filter, output, 
+                job_input, job_run, 
+                job_time, job_error],
+    [ {C, Module:info(C)} || C <- ['query' | Counters] ].
+
 %% @doc The number of input events for this query module.
 -spec input(atom()) -> non_neg_integer().
 input(Module) ->
@@ -364,6 +372,10 @@ manage_params_name(Module) -> reg_name(Module, "_params_mgr").
 manage_counts_name(Module) -> reg_name(Module, "_counters_mgr").
 
 
+start() ->
+    ok = application:start(syntax_tools),
+    ok = application:start(compiler),
+    ok = application:start(goldrush).
 
 %% @todo Move comment.
 %% @private Map a query to a simplified query tree term.
@@ -751,25 +763,25 @@ events_test_() ->
                     ?assertEqual(3, Mod:info(filter)),
                     ?assertEqual(1, receive {Msg, undefined} -> Msg after 0 -> notcalled end),
 
-                    Msg1 = glc:run(Mod, fun(_Event, _EStore) -> 
+                    {_, Msg1} = glc:run(Mod, fun(_Event, _EStore) -> 
                         timer:sleep(200),
                         {error, badtest}
                         
                     end, gre:make([{a,1}], [list])),
-                    ?assertEqual(2, Mod:info(output)),
+                    ?assertEqual(3, Mod:info(output)),
                     ?assertEqual(3, Mod:info(filter)),
                     ?assertEqual(2, Mod:info(job_input)),
                     ?assertEqual(1, Mod:info(job_error)),
                     ?assertEqual(1, Mod:info(job_run)),
                     ?assertEqual({error, badtest}, Msg1),
 
-                    Msg2 = glc:run(Mod, fun(_Event, _EStore) -> 
-                        timer:sleep(200),
+                    {_, Msg2} = glc:run(Mod, fun(_Event, _EStore) -> 
+                        timer:sleep(20),
                         {ok, goodtest}
                         
-                    end, gre:make([{a,1}], [list])),
+                    end, gre:make([{a,2}], [list])),
                     ?assertEqual(3, Mod:info(output)),
-                    ?assertEqual(3, Mod:info(filter)),
+                    ?assertEqual(4, Mod:info(filter)),
                     ?assertEqual(3, Mod:info(job_input)),
                     ?assertEqual(1, Mod:info(job_error)),
                     ?assertEqual(2, Mod:info(job_run)),
@@ -778,7 +790,7 @@ events_test_() ->
 
                     glc:reset_counters(Mod, input),
                     ?assertEqual(0, Mod:info(input)),
-                    ?assertEqual(3, Mod:info(filter)),
+                    ?assertEqual(4, Mod:info(filter)),
                     ?assertEqual(3, Mod:info(output)),
                     ?assertEqual(3, Mod:info(job_input)),
                     ?assertEqual(1, Mod:info(job_error)),
@@ -822,22 +834,7 @@ events_test_() ->
             },
             {"variable storage test",
                 fun() ->
-                    {compiled, Mod} = setup_query(testmod21,
-                        glc:eq(a, 2), [{stream, time}]),
-                    glc:handle(Mod, gre:make([{'a', 2}], [list])),
-                    glc:handle(Mod, gre:make([{'b', 1}], [list])),
-                    ?assertEqual(2, Mod:info(input)),
-                    ?assertEqual(1, Mod:info(filter)),
-                    glc:handle(Mod, gre:make([{'b', 2}], [list])),
-                    ?assertEqual(3, Mod:info(input)),
-                    ?assertEqual(2, Mod:info(filter)),
-                    ?assertEqual({ok, time}, glc:get(Mod, stream)),
-                    ?assertEqual({error, undefined}, glc:get(Mod, beam))
-                end
-            },
-            {"variable storage test",
-                fun() ->
-                    {compiled, Mod} = setup_query(testmod19,
+                    {compiled, Mod} = setup_query(testmod20a,
                         glc:eq(a, 2), [{stream, time}]),
                     glc:handle(Mod, gre:make([{'a', 2}], [list])),
                     glc:handle(Mod, gre:make([{'b', 1}], [list])),
@@ -860,7 +857,7 @@ events_test_() ->
                     G2 = glc:with(glc:eq(b, 2), fun(_Event, EStore) -> 
                        Self ! {b, EStore} end),
 
-                    {compiled, Mod} = setup_query(testmod20, any([G1, G2]),
+                    {compiled, Mod} = setup_query(testmod20b, any([G1, G2]),
                          Store),
                     glc:handle(Mod, gre:make([{a,1}], [list])),
                     ?assertEqual(1, Mod:info(output)),
@@ -946,15 +943,15 @@ events_test_() ->
                     ?assertEqual(2, Mod:info(output)),
                     ?assertEqual(1, Mod:info(input)),
                     ?assertEqual(1, Mod:info(filter)),
-                    ?assertEqual(b, receive {Msg, _Store} -> Msg after 0 -> notcalled end),
                     ?assertEqual(a, receive {Msg, _Store} -> Msg after 0 -> notcalled end),
+                    ?assertEqual(b, receive {Msg, _Store} -> Msg after 0 -> notcalled end),
                     %
                     glc:handle(Mod, gre:make([{a,1}, {r, 0.6}], [list])),
                     ?assertEqual(4, Mod:info(output)),
                     ?assertEqual(2, Mod:info(input)),
                     ?assertEqual(2, Mod:info(filter)),
-                    ?assertEqual(b, receive {Msg, _Store} -> Msg after 0 -> notcalled end),
                     ?assertEqual(a, receive {Msg, _Store} -> Msg after 0 -> notcalled end),
+                    ?assertEqual(b, receive {Msg, _Store} -> Msg after 0 -> notcalled end),
                     %
                     glc:handle(Mod, gre:make([{a,2}, {r, 0.7}, {b, 3}], [list])),
                     ?assertEqual(5, Mod:info(output)),
@@ -966,9 +963,65 @@ events_test_() ->
                     ?assertEqual(8, Mod:info(output)),
                     ?assertEqual(4, Mod:info(input)),
                     ?assertEqual(4, Mod:info(filter)),
-                    ?assertEqual(c, receive {Msg, _Store} -> Msg after 0 -> notcalled end),
+                    ?assertEqual(a, receive {Msg, _Store} -> Msg after 0 -> notcalled end),
                     ?assertEqual(b, receive {Msg, _Store} -> Msg after 0 -> notcalled end),
-                    ?assertEqual(a, receive {Msg, _Store} -> Msg after 0 -> notcalled end)
+                    ?assertEqual(c, receive {Msg, _Store} -> Msg after 0 -> notcalled end)
+                end
+            },
+            {"with single-function run test",
+                fun() ->
+                    Self = self(),
+                    Store = [{stored, value}],
+                    {compiled, Mod1} = setup_query(testmod25a,
+                        glc:with(glc:all([glc:gt(runtime, 0.15), glc:lt(a, 3)]), fun(Event, EStore) -> 
+                           Self ! {gre:fetch(a, Event), EStore} end),
+                         Store),
+                    glc:run(Mod1, fun(_Event, _EStore) -> timer:sleep(200), ok end, gre:make([{a, 2}], [list])),
+                    ?assertEqual(1, Mod1:info(output)),
+                    ?assertEqual(2, receive {Msg, Store} -> Msg after 250 -> notcalled end),
+                    {compiled, Mod2} = setup_query(testmod25b,
+                        glc:with(glc:all([glc:gt(runtime, 0.15), glc:lt(a, 3)]), fun(Event, EStore) -> 
+                           Self ! {gre:fetch(a, Event), EStore}
+                        end), Store),
+                    {_, {error, later}} = glc:run(Mod2, fun(_Event, _EStore) -> 
+                           timer:sleep(200), 
+                           erlang:exit(later)
+                    end, gre:make([{a, 2}], [list])),
+                    ?assertEqual(1, Mod2:info(output)),
+                    ?assertEqual(1, Mod2:info(job_error)),
+                    ?assertEqual(2, receive {Msg, Store} -> Msg after 250 -> notcalled end)
+                end
+            },
+            {"with multi-function output run error test",
+                fun() ->
+                    Self = self(),
+                    Store = [{stored, value}],
+                    {compiled, Mod} = setup_query(testmod26,
+                          [glc:with(glc:gt(runtime, 0.15), fun(Event, _EStore) -> 
+                              Self ! {a, gre:fetch(b, Event)} 
+                           end),
+                           glc:with(glc:eq(c, 3), fun(Event, _EStore) -> 
+                              Self ! {a, gre:fetch(a, Event)} 
+                           end),
+                           glc:with(glc:eq(b, 3), fun(Event, _EStore) -> 
+                              Self ! {a, gre:fetch(a, Event)} 
+                           end),
+                           glc:with(glc:eq(a, 1), fun(Event, _EStore) -> 
+                              receive {a, _Store} -> 
+                                  Self ! {b, gre:fetch(b, Event)}
+                              after 10 -> notcalled end
+                           end)
+                           ],
+                         Store),
+                    Event = gre:make([{a,1}, {b, 3}, {c, 4}], [list]),
+                    {_, {error, bye}} = glc:run(Mod, fun(_Event, _EStore) -> 
+                            timer:sleep(200),
+                            erlang:error(bye) 
+                    end, Event),
+                    ?assertEqual(3, Mod:info(output)),
+                    ?assertEqual(1, Mod:info(filter)),
+                    ?assertEqual(1, Mod:info(job_error)),
+                    ?assertEqual(b, receive {b=Msg, _Store} -> Msg after 0 -> notcalled end)
                 end
             }
         ]

src/glc_code.erl

@@ -93,9 +93,6 @@ abstract_module_(Module, #module{tables=Tables,
         ?erl:arity_qualifier(
             ?erl:atom(table),
             ?erl:integer(1)),
-        %?erl:arity_qualifier(
-        %    ?erl:atom(sidejob),
-        %    ?erl:integer(2)),
         ?erl:arity_qualifier(
             ?erl:atom(runjob),
             ?erl:integer(2)),
@@ -162,11 +159,11 @@ abstract_module_(Module, #module{tables=Tables,
              [?erl:application(none,
                         ?erl:atom(job_result), [
                           ?erl:catch_expr(
-                            abstract_apply(timer, tc, [
+                            abstract_apply(glc_run, execute, [
                                 ?erl:variable("Fun"),
                                 ?erl:list([?erl:variable("Meta"), 
                                            ?erl:abstract(Store)])
-                                ])),
+                            ])),
                             ?erl:variable("Meta")])
              ] 
         )]),
@@ -231,12 +228,6 @@ abstract_runjob(#module{'query'=_Query, store=_Store}) ->
       [
        ?erl:case_expr(?erl:variable("JobResult"),
          [
-          ?erl:clause(
-            [?erl:tuple([?erl:atom('EXIT'),?erl:variable("Reason")])],
-            none,
-            [abstract_count(job_error),
-             ?erl:tuple([?erl:atom(error), ?erl:variable("Reason")])]),
-
           ?erl:clause(
             [?erl:tuple([?erl:variable("Time"), ?erl:variable("Result")])],
             none,
@@ -245,16 +236,24 @@ abstract_runjob(#module{'query'=_Query, store=_Store}) ->
                 ?erl:clause(
                   [?erl:tuple([?erl:atom(error),?erl:variable("Reason")])],
                   none,
-                  [abstract_count(job_error),
-                   ?erl:tuple([?erl:atom(error), ?erl:variable("Reason")])]),
+                  [abstract_count(input), abstract_count(job_error),
+                   ?erl:application(none, ?erl:atom(handle_), 
+                        abstract_job(Time, [?erl:tuple([?erl:atom(error), 
+                                                        ?erl:variable("Reason")])])),
+                   abstract_count(job_time, ?erl:variable("Time")),
+                   ?erl:tuple([?erl:variable("Time"), 
+                               ?erl:tuple([?erl:atom(error), 
+                                           ?erl:variable("Reason")])])]),
 
                 ?erl:clause(
                   [?erl:variable("Result")],
                   none,
-                  [abstract_count(job_run),
-                   ?erl:application(none, ?erl:atom(handle_), abstract_job(Time)),
+                  [abstract_count(input), abstract_count(job_run),
+                   ?erl:application(none, ?erl:atom(handle_), 
+                        abstract_job(Time)),
                    abstract_count(job_time, ?erl:variable("Time")),
-                   ?erl:variable("Result")])
+                   ?erl:tuple([?erl:variable("Time"), 
+                               ?erl:variable("Result")])])
                ])
             ])
          ])
@@ -262,10 +261,13 @@ abstract_runjob(#module{'query'=_Query, store=_Store}) ->
     )].
 
 abstract_job(Time) ->
+    abstract_job(Time, []).
+abstract_job(Time, Error) ->
     Pairs = abstract_apply(gre, pairs, [?erl:variable("Meta")]),
     Runtime = ?erl:list([?erl:tuple([?erl:atom(runtime), Time])]),
     [abstract_apply(gre, make, 
-              [abstract_apply(erlang, '++', [Pairs, Runtime]),
+              [abstract_apply(erlang, '++', [?erl:list(Error), 
+                             abstract_apply(erlang, '++', [Pairs, Runtime])]),
                ?erl:abstract([list])])].
 
 %% @private Return the clauses of the info/1 function.
@@ -312,7 +314,7 @@ abstract_filter({Type, [{with, _Cond, _Fun}|_] = I}, Data, State) when Type =:=
         _OnNomatch=fun(_State2) -> [abstract_count(filter)] end, State);
 abstract_filter([{with, _Cond, _Fun}|_] = I, Data, State) ->
     OnNomatch = fun(_State2) -> [abstract_count(filter, 0)] end,
-    Funs = lists:foldl(fun({with, Cond, Fun}, Acc) -> 
+    Funs = lists:foldr(fun({with, Cond, Fun}, Acc) -> 
               [{Cond, Fun, Data}|Acc]
       end, [], I),
     abstract_within(Funs, OnNomatch, State);

src/glc_run.erl

@@ -0,0 +1,27 @@
+-module(glc_run).
+
+-export([execute/2]).
+
+-ifdef(erlang18).
+-define(time_now(), erlang:monotonic_time()).
+-define(time_diff(T1, T2), erlang:convert_time_unit(T2 - T1, native, micro_seconds)).
+-else.
+-define(time_now(), os:timestamp()). 
+-define(time_diff(T1, T2), timer:now_diff(T2, T1)). 
+-endif.
+
+execute(Fun, [Event, Store]) ->
+    T1 = ?time_now(),
+    case (catch Fun(Event, Store)) of
+        {'EXIT', {Reason, _ST}} ->
+            T2 = ?time_now(),
+            {?time_diff(T1, T2), {error, Reason}};
+        {'EXIT', Reason} ->
+            T2 = ?time_now(),
+            {?time_diff(T1, T2), {error, Reason}};
+        Else -> 
+            T2 = ?time_now(),
+            {?time_diff(T1, T2), Else}
+    end.
+
+