<commit_before>
-module(job_coordinator).
-export([new/1]).

-include("disco.hrl").

-define(TASK_MAX_FAILURES, 100).
-define(FAILED_TASK_PAUSE, 1000).

% In theory we could keep the HTTP connection pending until the job
% finishes but in practice long-living HTTP connections are a bad idea.
% Thus, the HTTP request spawns a new process, job_coordinator, that
% takes care of coordinating the whole map-reduce show, including
% fault-tolerance. The HTTP request returns immediately. It may poll
% the job status e.g. by using handle_ctrl's get_results.
-spec new(binary()) -> _.
new(PostData) ->
    TMsg = "couldn't start a new job coordinator in 10s (master busy?)",
    S = self(),
    P = spawn(fun() -> init_job_coordinator(S, PostData) end),
    receive
    {P, ok, JobName} -> {ok, JobName};
    {P, env_failed} -> throw("creating job directories failed "
             "(disk full?)");
    {P, invalid_prefix} -> throw("invalid prefix");
    {P, invalid_jobdesc} -> throw("invalid job description");
    {P, timeout} -> throw(TMsg);
    _ -> throw("job coordinator failed")
    after 10000 ->
    throw(TMsg)
    end.

% job_coordinator() orchestrates map/reduce tasks for a job
-spec init_job_coordinator(pid(), binary()) -> _.
init_job_coordinator(Parent, PostData) ->
    Msg = netstring:decode_netstring_fd(PostData),
    case catch find_values(Msg) of
    {'EXIT', _} ->
        Parent ! {self(), invalid_jobdesc};
    Params ->
        init_job_coordinator(Parent, Params, PostData)
    end.

-spec init_job_coordinator(pid(), {nonempty_string(), jobinfo()}, binary()) -> _.
init_job_coordinator(Parent, {Prefix, JobInfo}, PostData) ->
    C = string:chr(Prefix, $/) + string:chr(Prefix, $.),
    if C > 0 ->
    Parent ! {self(), invalid_prefix};
    true ->
    case gen_server:call(event_server, {new_job, Prefix, self()}, 10000) of
        {ok, JobName} ->
        save_params(JobName, PostData),
        Parent ! {self(), ok, JobName},
        job_coordinator(JobName, JobInfo);
        _ ->
        Parent ! {self(), timeout}
    end
    end.

save_params(Name, PostData) -><commit_msg>increased the timeout for starting a new job from 10s to 30s<commit_after>
-module(job_coordinator).
-export([new/1]).

-include("disco.hrl").

-define(TASK_MAX_FAILURES, 100).
-define(FAILED_TASK_PAUSE, 1000).

% In theory we could keep the HTTP connection pending until the job
% finishes but in practice long-living HTTP connections are a bad idea.
% Thus, the HTTP request spawns a new process, job_coordinator, that
% takes care of coordinating the whole map-reduce show, including
% fault-tolerance. The HTTP request returns immediately. It may poll
% the job status e.g. by using handle_ctrl's get_results.
-spec new(binary()) -> _.
new(PostData) ->
    TMsg = "couldn't start a new job coordinator in 30s (master busy?)",
    S = self(),
    P = spawn(fun() -> init_job_coordinator(S, PostData) end),
    receive
    {P, ok, JobName} -> {ok, JobName};
    {P, env_failed} -> throw("creating job directories failed "
             "(disk full?)");
    {P, invalid_prefix} -> throw("invalid prefix");
    {P, invalid_jobdesc} -> throw("invalid job description");
    {P, timeout} -> throw(TMsg);
    _ -> throw("job coordinator failed")
    after 30000 ->
    throw(TMsg)
    end.

% job_coordinator() orchestrates map/reduce tasks for a job
-spec init_job_coordinator(pid(), binary()) -> _.
init_job_coordinator(Parent, PostData) ->
    Msg = netstring:decode_netstring_fd(PostData),
    case catch find_values(Msg) of
    {'EXIT', _} ->
        Parent ! {self(), invalid_jobdesc};
    Params ->
        init_job_coordinator(Parent, Params, PostData)
    end.

-spec init_job_coordinator(pid(), {nonempty_string(), jobinfo()}, binary()) -> _.
init_job_coordinator(Parent, {Prefix, JobInfo}, PostData) ->
    C = string:chr(Prefix, $/) + string:chr(Prefix, $.),
    if C > 0 ->
    Parent ! {self(), invalid_prefix};
    true ->
    case gen_server:call(event_server, {new_job, Prefix, self()}, 10000) of
        {ok, JobName} ->
        save_params(JobName, PostData),
        Parent ! {self(), ok, JobName},
        job_coordinator(JobName, JobInfo);
        _ ->
        Parent ! {self(), timeout}
    end
    end.

save_params(Name, PostData) ->
{'fraction_non_alphanumeric': 0.14034264257216617, 'fraction_numerical': 0.010795587890166627, 'mean_word_length': 3.4303534303534304, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4203297', 'n_tokens_mistral': 1408, 'n_tokens_neox': 1343, 'n_words': 393}