<commit_before>-module(event_server).
-behaviour(gen_server).

-define(EVENT_PAGE_SIZE, 100).
-define(EVENT_BUFFER_SIZE, 1000).
-define(EVENT_BUFFER_TIMEOUT, 2000).

-export([event/4, event/5, event/6]).
-export([start_link/0, stop/0, init/1, handle_call/3, handle_cast/2, 
        handle_info/2, terminate/2, code_change/3]).

start_link() ->
        error_logger:info_report([{"Event server starts"}]),
        case gen_server:start_link({local, event_server}, event_server, [], []) of
                {ok, Server} -> {ok, Server};
                {error, {already_started, Server}} -> {ok, Server}
        end.

stop() ->
        gen_server:call(event_server, stop).

init(_Args) ->
        ets:new(event_files, [named_table]),
        {ok, {dict:new(), dict:new()}}.

json_list([], _) -> [];
json_list([X], L) ->
        [<<"[">>, lists:reverse([X|L]), <<"]">>];
json_list([X|R], L) ->
        json_list(R, [<<X/binary, ",">>|L]).

handle_call(get_jobnames, _From, {Events, _} = S) ->
        Lst = dict:fold(fun(JobName, {_, Nu, Pid}, L) ->
                [{JobName, Nu, Pid}|L]
        end, [], Events),
        {reply, {ok, Lst}, S};

handle_call({get_job_events, JobName, Q, N0}, _From, {_, MsgBuf} = S) ->
        N = if N0 > 1000 -> 1000; true -> N0 end,
        case dict:find(JobName, MsgBuf) of
                _ when Q =/= "" ->
                        {reply, {ok, 
                                json_list(grep_log(JobName, Q, N), [])}, S};
                {ok, {_, _, MsgLst}} ->
                        {reply, {ok,
                                json_list(lists:sublist(MsgLst, N), [])}, S};
                error ->
                        {reply, {ok,
                                json_list(tail_log(JobName, N), [])}, S}
        end;

handle_call({get_results, JobName}, _From, {Events, _} = S) ->
        case dict:find(JobName, Events) of
                error -> {reply, invalid_job, S};
                {ok, {[{ready, Res}|_], _, Pid}} ->
                        {reply, {ready, Pid, Res}, S};
                {ok, {_, _, Pid}} ->
                        Alive = is_process_alive(Pid),
                        if Alive ->
                                {reply, {active, Pid}, S};
                        true ->
                                {reply, {dead, Pid}, S}
                        end
        end;
                
handle_call({get_jobinfo, JobName}, _From, {Events, _} = S) ->
        case dict:find(JobName, Events) of
                error -> {reply, invalid_job, S};
                {ok, {EvLst, Nu, Pid}} ->
                        JobNfo = event_filter(job_data, EvLst),
                        Res = event_filter(ready, EvLst),
                        Ready = event_filter(task_ready, EvLst),
                        Failed = event_filter(task_failed, EvLst),
                        TStamp = format_timestamp(Nu),
                        {reply, {ok, {TStamp, Pid, 
                                JobNfo, Res, Ready, Failed}}, S}
        end.

% There's a small catch with adding a job event: If a job's records have been
% cleaned with clean_job already, we do not want a zombie worker to re-open
% the job's records by adding a new event. Thus only an event with the atom
% start is allowed to initialize records for a new job. Other events are 
% silently ignored if there are no previous records for this job.
handle_cast({add_job_event, Host, JobName, M, {start, Pid} = P},
        {Events0, MsgBuf0} = S) ->
        
        V = dict:is_key(JobName, Events0),
        if V -> {noreply, S};
        true ->
                Events = dict:store(JobName, {[], now(), Pid}, Events0),
                MsgBuf = dict:store(JobName, {0, 0, []}, MsgBuf0),
                
                {ok, Root} = application:get_env(disco_root),
                FName = filename:join([Root, disco_server:jobhome(JobName), "events"]),
                Pid = spawn(fun() -> job_event_handler(FName) end),
                ets:insert(event_files, {JobName, Pid}),

                {noreply, add_event(Host, JobName, M, P, {Events, MsgBuf})}
        end;

handle_cast({add_job_event, Host, JobName, M, P}, {_, MsgBuf} = S) ->
        V = dict:is_key(JobName, MsgBuf),
        if V -> {noreply, add_event(Host, JobName, M, P, S)};
        true -> {noreply, S}
        end;

handle_cast({job_done, JobName}, {Events, MsgBuf} = S) ->
        case ets:lookup(event_files, JobName) of
                [] -> ok;
                [{_, EventProc}] ->
                        EventProc ! done,
                        ets:delete(event_files, JobName)
        end,
        case dict:find(JobName, MsgBuf) of
                error -> {noreply, S};
                {ok, _} ->
                        {noreply, {Events, dict:erase(JobName, MsgBuf)}}<commit_msg>fixed a stupid minor bug<commit_after>-module(event_server).
-behaviour(gen_server).

-define(EVENT_PAGE_SIZE, 100).
-define(EVENT_BUFFER_SIZE, 1000).
-define(EVENT_BUFFER_TIMEOUT, 2000).

-export([event/4, event/5, event/6]).
-export([start_link/0, stop/0, init/1, handle_call/3, handle_cast/2, 
        handle_info/2, terminate/2, code_change/3]).

start_link() ->
        error_logger:info_report([{"Event server starts"}]),
        case gen_server:start_link({local, event_server}, event_server, [], []) of
                {ok, Server} -> {ok, Server};
                {error, {already_started, Server}} -> {ok, Server}
        end.

stop() ->
        gen_server:call(event_server, stop).

init(_Args) ->
        ets:new(event_files, [named_table]),
        {ok, {dict:new(), dict:new()}}.

json_list([], _) -> [];
json_list([X], L) ->
        [<<"[">>, lists:reverse([X|L]), <<"]">>];
json_list([X|R], L) ->
        json_list(R, [<<X/binary, ",">>|L]).

handle_call(get_jobnames, _From, {Events, _} = S) ->
        Lst = dict:fold(fun(JobName, {_, Nu, Pid}, L) ->
                [{JobName, Nu, Pid}|L]
        end, [], Events),
        {reply, {ok, Lst}, S};

handle_call({get_job_events, JobName, Q, N0}, _From, {_, MsgBuf} = S) ->
        N = if N0 > 1000 -> 1000; true -> N0 end,
        case dict:find(JobName, MsgBuf) of
                _ when Q =/= "" ->
                        {reply, {ok, 
                                json_list(grep_log(JobName, Q, N), [])}, S};
                {ok, {_, _, MsgLst}} ->
                        {reply, {ok,
                                json_list(lists:sublist(MsgLst, N), [])}, S};
                error ->
                        {reply, {ok,
                                json_list(tail_log(JobName, N), [])}, S}
        end;

handle_call({get_results, JobName}, _From, {Events, _} = S) ->
        case dict:find(JobName, Events) of
                error -> {reply, invalid_job, S};
                {ok, {[{ready, Res}|_], _, Pid}} ->
                        {reply, {ready, Pid, Res}, S};
                {ok, {_, _, Pid}} ->
                        Alive = is_process_alive(Pid),
                        if Alive ->
                                {reply, {active, Pid}, S};
                        true ->
                                {reply, {dead, Pid}, S}
                        end
        end;
                
handle_call({get_jobinfo, JobName}, _From, {Events, _} = S) ->
        case dict:find(JobName, Events) of
                error -> {reply, invalid_job, S};
                {ok, {EvLst, Nu, Pid}} ->
                        JobNfo = event_filter(job_data, EvLst),
                        Res = event_filter(ready, EvLst),
                        Ready = event_filter(task_ready, EvLst),
                        Failed = event_filter(task_failed, EvLst),
                        TStamp = format_timestamp(Nu),
                        {reply, {ok, {TStamp, Pid, 
                                JobNfo, Res, Ready, Failed}}, S}
        end.

% There's a small catch with adding a job event: If a job's records have been
% cleaned with clean_job already, we do not want a zombie worker to re-open
% the job's records by adding a new event. Thus only an event with the atom
% start is allowed to initialize records for a new job. Other events are 
% silently ignored if there are no previous records for this job.
handle_cast({add_job_event, Host, JobName, M, {start, Pid} = P},
        {Events0, MsgBuf0} = S) ->
        
        V = dict:is_key(JobName, Events0),
        if V -> {noreply, S};
        true ->
                Events = dict:store(JobName, {[], now(), Pid}, Events0),
                MsgBuf = dict:store(JobName, {0, 0, []}, MsgBuf0),
                
                {ok, Root} = application:get_env(disco_root),
                FName = filename:join([Root, disco_server:jobhome(JobName), "events"]),
                EventProc = spawn(fun() -> job_event_handler(FName) end),
                ets:insert(event_files, {JobName, EventProc}),

                {noreply, add_event(Host, JobName, M, P, {Events, MsgBuf})}
        end;

handle_cast({add_job_event, Host, JobName, M, P}, {_, MsgBuf} = S) ->
        V = dict:is_key(JobName, MsgBuf),
        if V -> {noreply, add_event(Host, JobName, M, P, S)};
        true -> {noreply, S}
        end;

handle_cast({job_done, JobName}, {Events, MsgBuf} = S) ->
        case ets:lookup(event_files, JobName) of
                [] -> ok;
                [{_, EventProc}] ->
                        EventProc ! done,
                        ets:delete(event_files, JobName)
        end,
        case dict:find(JobName, MsgBuf) of
                error -> {noreply, S};
                {ok, _} ->
                        {noreply, {Events, dict:erase(JobName, MsgBuf)}}
{'fraction_non_alphanumeric': 0.15113244963991232, 'fraction_numerical': 0.008349859096127753, 'mean_word_length': 1.6728033472803348, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 91, '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': '4863501', 'n_tokens_mistral': 3169, 'n_tokens_neox': 2972, 'n_words': 727}