queue-rb branch merged

Author: Arkady Drovosekov

Makefile

@@ -11,7 +11,7 @@ HTML_DOC_DIR = doc/html
 ERLC_OPTS = +debug_info -DTEST
 #ERLC_OPTS = +debug_info -DTEST -DPROPER -pa ../proper/ebin
 ERLC := erlc $(ERLC_OPTS)
-VSN=1.8
+VSN=2.2
 APP_NAME=ejobman
 LICENSE=MIT
 

doc/readme

@@ -70,6 +70,8 @@ Version 1
 Version 2 (current)
 -------------------
 
+amqp routing key is a job group
+
 {type: rest},
 {info:
   {method: xxx,
@@ -81,7 +83,7 @@ Version 2 (current)
     {key2: val2}
     ]
   },
-  {group, xxx}
+  {group, xxx} % will be eliminated in future versions
 },
 
 Supported kinds of auth_info:

eall.config

@@ -16,6 +16,7 @@
         {error_logger_mf_maxfiles, 5}           % 5 files max
     ]},
     {ejobman, [
+        {temp_rt_key_for_group, <<"new">>},
         % log is common and may also be defined in eworkman
         {log0, "/var/log/erpher/ej"},
         {pid_file0, "/var/run/erpher.pid"},
@@ -33,7 +34,7 @@
                 {password, <<"guest">>},
                 {vhost , <<"/">>},
                 {exchange , <<"ejobman">>},
-                {exchange_type , <<"topic">>},
+                {exchange_type0 , <<"topic">>},
                 {queue , <<"test_queue_2">>},
                 {routing_key , <<"new">>}
         ]},
@@ -51,7 +52,24 @@
                 {run, 2}
             ]}
         ]},
-        {handler, [
+        {group_handler, [
+            {http_connect_timeout, 15000},
+            {http_timeout, 3600000},
+            {max_children, 10}, % default group
+            {job_groups, [
+                [
+                    {name, <<"g1">>},
+                    {max_children, 1}
+                ],
+                [
+                    {name, <<"g2">>},
+                    {max_children, 10}
+                ],
+                [
+                    {name, <<"g3">>},
+                    {max_children, 100}
+                ]
+            ]},
             {schema_rewrite, [
                 [
                     {src_type, regex},
@@ -91,28 +109,24 @@
                     {dst_host_hdr, "host2.localdomain"}
                 ]
             ]},
+            {debug, [
+                    {handler_child, 5},
+                    {handler_run, 2},
+                    {handler_job, 5},
+                    {msg, 3},
+                    {run, 2},
+                    {job, 5},
+                    {http, 3},
+                    {config, 0},
+                    {rewrite, 0}
+            ]}
+        ]},
+        {handler, [
             {job_log, "/var/log/erpher/job"},
             % never | minute | hour | day | {dow, 0..7} | month | year
             {job_log_rotate, day},
-            {http_connect_timeout, 15000},
-            {http_timeout, 3600000},
             {stat_limit_n, 101}, % amount
             {stat_limit_t, 123}, % seconds
-            {max_children, 10}, % to process short command
-            {job_groups, [
-                [
-                {name, <<"g1">>},
-                {max_children, 1}
-                ],
-                [
-                {name, <<"g2">>},
-                {max_children, 10}
-                ],
-                [
-                {name, <<"g3">>},
-                {max_children, 100}
-                ]
-            ]},
             {debug,
                 [
                     {handler_child, 2},
@@ -121,12 +135,10 @@
                     {job_queue, 0},
                     {job_result, 2}, % to log job results
                     {run, 2},
-                    {stat, 0},
-                    {http, 3},
-                    {rewrite, 0}
+                    {stat, 0}
                 ]
             }
-        ]}
+        ]} % handler
     ]},
     {eworkman, [
         {delay_for_log0, 10}, % ms. Start delay for ejobman to create a log

include/chi.hrl

@@ -0,0 +1,15 @@
+-ifndef(ejobman_chi).
+-define(ejobman_chi, true).
+
+-record(chi, {
+    pid,
+    id,
+    mon,
+    os_pid,
+    tag,
+    alive=true,
+    stop={0,0,0}, % time of marking dead
+    start={0,0,0} % time in now() format
+}).
+
+-endif.

include/ejobman.hrl

@@ -2,13 +2,15 @@
 -define(ejobman_params, true).
 
 -include("nums.hrl").
+-include("chi.hrl").
 
 % state of a worker gen_server
 -record(child, {
     name,
     port,
     id,
     os_pid,
+    gh_pid,
     group,
     tag,
     duration,
@@ -31,17 +33,6 @@
     max_children
 }).
 
--record(chi, {
-    pid,
-    id,
-    mon,
-    os_pid,
-    tag,
-    alive=true,
-    stop={0,0,0}, % time of marking dead
-    start={0,0,0} % time in now() format
-}).
-
 -record(pool, {
     id,
     w_duration = 86400, % seconds
@@ -61,8 +52,8 @@
 
 % state of a handler and a receiver gen_server
 -record(ejm, {
-    ch_queues, % dict: group -> queue of jobs
-    ch_data, % dict: group -> spawned children list
+    ch_queues :: dict(), % dict: group -> queue of jobs
+    ch_data   :: dict(), % dict: group -> spawned children list
     max_children = 32767,
     http_connect_timeout = ?HTTP_CONNECT_TIMEOUT,
     http_timeout = ?HTTP_TIMEOUT,
@@ -72,6 +63,8 @@
     web_server_opts,
     log,
     pid_file,
+    group_handler = [], % config for group handlers
+    group_handler_run = [], % started group handlers. Unnecessary, in fact
     job_groups = [], % configured job groups
     job_log, % filename for job log
     job_log_last,

include/group_handler.hrl

@@ -0,0 +1,24 @@
+-ifndef(group_handler).
+-define(group_handler, true).
+
+-include("nums.hrl").
+
+% state of a group_handler gen_server
+-record(egh, {
+    id         :: reference(), % id that is used by supervisor
+    group, % group id
+    max        :: non_neg_integer(), % max children for the group
+    ch_run = []:: list(),   % running children
+    ch_queue   :: queue(),  % fetched from amqp but not yet started children
+    vhost      :: binary(), % amqp virtual host
+    conn,
+    exchange   :: binary(), % amqp exchange for a group
+    queue      :: binary(), % amqp queue for a group
+    http_connect_timeout = ?HTTP_CONNECT_TIMEOUT :: non_neg_integer(),
+    http_timeout         = ?HTTP_TIMEOUT         :: non_neg_integer(),
+    schema_rewrite = []  :: list(),
+    url_rewrite = []     :: list(),
+    debug = []           :: list()
+}).
+
+-endif.

include/job.hrl

@@ -4,7 +4,7 @@
 %-define(T, 1000).
 
 -record(job, {
-    id :: reference(),
+    id :: reference() | binary(),
     type = 'rest',
     method,
     url,

include/nums.hrl

@@ -2,6 +2,7 @@
 -define(ejobman_nums, true).
 
 -define(LOG_PROCS_INTERVAL, 60). % seconds
+-define(GID_DEFAULT, default).
 
 -define(STAT_T_KEEP_MINUTES, 62).
 -define(STAT_T_KEEP_HOURS, 26).

include/rabbit_session.hrl

@@ -16,7 +16,8 @@
 -record(conn, {
     'channel' = false,
     'connection' = false,
-    'consumer_tag' = false
+    'consumer_tag' = false,
+    'ticket' = false
 }).
 
 -endif.

include/receiver.hrl

@@ -9,7 +9,12 @@
     conn,
     log,
     pid_file,
-    debug
+
+    % for this rt key look into job's group field for real rt key
+    temp_rt_key_for_group :: binary(),
+
+    groups = [] :: list(),
+    debug = [] :: list()
 }).
 
 -endif.

src/ejobman.app.src

@@ -1,7 +1,7 @@
 {application, ejobman, [
     {description, "ejobman"},
     {id, "ejobman"},
-    {vsn, "1.8"},
+    {vsn, "2.2"},
     {modules, [
         ejobman_app,
         ejobman_child,
@@ -10,9 +10,14 @@
         ejobman_conf,
         ejobman_conf_rabbit,
         ejobman_data,
+        ejobman_group_handler_cmd,
+        ejobman_group_handler,
+        ejobman_group_handler_spawn,
+        ejobman_group_sup,
         ejobman_handler_cmd,
         ejobman_handler,
         ejobman_log,
+        ejobman_print_stat,
         ejobman_rb,
         ejobman_receiver_cmd,
         ejobman_receiver,
@@ -24,6 +29,7 @@
         ejobman_stat,
         ejobman_receiver,
         ejobman_handler,
+        ejobman_group_supervisor,
         ejobman_child_supervisor
     ]},
     {mod, {ejobman_app,[]}},

src/ejobman_child.erl

@@ -160,8 +160,6 @@ main_action(State) ->
 %%
 -spec process_cmd(#child{}) -> ok.
 
-process_cmd(#child{from = 'undefined'}) ->
-    ok;
 process_cmd(#child{url = <<>>}) ->
     ok;
 process_cmd(#child{url = [_ | _]} = St) ->
@@ -181,7 +179,7 @@ process_cmd(_) ->
 %% https://github.com/cmullaparthi/ibrowse
 %% @since 2011-07-18
 %%
-real_cmd(#child{id=Id, method=Method_bin, params=Params, tag=Tag,
+real_cmd(#child{id=Id, method=Method_bin, params=Params, tag=Tag, gh_pid=Gh_pid,
         http_connect_timeout=Conn_t, http_timeout=Http_t} = St) ->
     mpln_p_debug:pr({?MODULE, real_cmd, ?LINE, params, Id, self(),
         St}, St#child.debug, run, 4),
@@ -195,7 +193,7 @@ real_cmd(#child{id=Id, method=Method_bin, params=Params, tag=Tag,
         Req}, St#child.debug, http, 5),
     mpln_p_debug:pr({?MODULE, real_cmd, ?LINE, start, Id, self()},
         St#child.debug, run, 2),
-    ejobman_receiver:send_ack(Id, Tag),
+    ejobman_group_handler:send_ack(Gh_pid, Id, Tag),
     T1 = now(),
     ejobman_stat:add(Id, 'http_start',
                      [{'header', mpln_misc_web:make_proplist_binary(Hdr)},
@@ -216,7 +214,8 @@ real_cmd(#child{id=Id, method=Method_bin, params=Params, tag=Tag,
 %%
 %% @doc sends result to ejobman_handler and ejobman_stat
 %%
-process_result(#child{id=Id, group=Group}, Res, T1, T2) ->
+process_result(#child{id=Id, gh_pid=Pid, group=Group}, Res, T1, T2) ->
+    ejobman_group_handler:cmd_result(Pid, Res, T1, T2, Id),
     send_stat(Id, Res),
     ejobman_handler:cmd_result(Res, T1, T2, Group, Id).