<commit_before>%% @author Northscale <<EMAIL>>
%% @copyright 2010 NorthScale, Inc.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%      http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%
-module(ns_port_server).

-behavior(gen_server).

-include("ns_common.hrl").

%% API
-export([start_link/1, start_link_named/2,
         is_active/1, activate/1]).

%% gen_server callbacks
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         code_change/3,
         terminate/2]).

-define(NUM_MESSAGES, 5). % Number of the most recent messages to log on crash
%% we're passing port stdout/stderr messages to log after delay of
%% INTERVAL milliseconds. Dropping messages once MAX_MESSAGES is
%% reached. Thus we're limiting rate of messages to
%% MAX_MESSAGES/INTERVAL msg/millisecond
%%
%% The following gives us 300 lines/second in bursts up to 60 lines
-define(MAX_MESSAGES, 60). % Max messages per interval
-define(INTERVAL, 200). % Interval over which to throttle

-include_lib("eunit/include/eunit.hrl").

%% Server state
-record(state, {port :: port() | pid() | undefined,
                params :: tuple(),
                messages,
                logger :: atom(),
                log_tref :: timer:tref(),
                log_buffer = [],
                dropped=0 :: non_neg_integer()}).

%% API

start_link(Fun) ->
    gen_server:start_link(?MODULE,
                          Fun, []).
start_link_named(Name, Fun) ->
    gen_server:start_link({local, Name}, ?MODULE, Fun, []).

is_active(Pid) ->
    gen_server:call(Pid, is_active, infinity).

activate(Pid) ->
    gen_server:call(Pid, activate, infinity).

%% gen_server callbacks
init(Fun) ->
    {Name, _Cmd, _Args, Opts} = Params = Fun(),
    process_flag(trap_exit, true), % Make sure terminate gets called
    Opts2 = lists:foldl(fun proplists:delete/2, Opts,
                        [port_server_dont_start, log]),
    DontStart = proplists:get_bool(port_server_dont_start, Opts),
    Log = proplists:get_value(log, Opts),
    Params2 = erlang:setelement(4, Params, Opts2),

    State =
        case Log of
            undefined ->
                #state{};
            _ when is_list(Log) ->
                Sink = Logger = Name,

                ok = ns_server:start_disk_sink(Sink, Log),

                ale:stop_logger(Logger),
                ok = ale:start_logger(Logger, debug, ale_noop_formatter),

                ok = ale:add_sink(Logger, Sink, debug),

                #state{logger=Name}
        end,

    Port = case DontStart of
               false -> port_open(Params2);
               true -> undefined
           end,
    {ok, State#state{port = Port,
                     params = Params2,
                     messages = ringbuffer:new(?NUM_MESSAGES)}}.

handle_info({send_to_port, Msg}, #state{port = undefined} = State) ->
    ?log_debug("Got send_to_port when there's no port running yet. Will kill myself."),
    {stop, {unexpected_send_to_port, {send_to_port, Msg}}, State};
handle_info({send_to_port, Msg}, State) ->
    ?log_debug("Sending the following to port: ~p", [Msg]),
    port_write(State#state.port, Msg),
    {noreply, State};
handle_info({Port, {data, Data}}, #state{port = Port,
                                         logger = Logger} = State) ->
    %% Store the last messages in case of a crash
    Msg = extract_message(Data),
    Messages = ringbuffer:add(Msg, State#state.messages),
    State1 = State#state{messages=Messages},

    NewState =
        case Logger of
            undefined ->
                {Buf, Dropped} = case {State#state.log_buffer, State#state.dropped} of
                                     {B, D} when length(B) < ?MAX_MESSAGES ->
                                         {[Msg|B], D};
                                     {B, D} ->
                                         {B, D + 1}
                                 end,
                TRef = case State#state.log_tref of
                           undefined ->
                               timer2:send_after(?INTERVAL, log);
                           T ->
                               T
                       end,
                State1#state{log_buffer=Buf, log_tref=TRef, dropped=Dropped};
            _ ->
                ale:debug(Logger, [Msg, $\n]),
                State1
        end,

    port_deliver(Port),
    {noreply, NewState};
handle_info(log, State) ->
    State1 = log(State),<commit_msg>ns_port_server to keep recent messages based on their size<commit_after>%% @author Northscale <<EMAIL>>
%% @copyright 2010 NorthScale, Inc.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%      http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%
-module(ns_port_server).

-behavior(gen_server).

-include("ns_common.hrl").

%% API
-export([start_link/1, start_link_named/2,
         is_active/1, activate/1]).

%% gen_server callbacks
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         code_change/3,
         terminate/2]).

%% Keep on 1KiB worth of recent messages to log if process crashes.
-define(KEEP_MESSAGES_BYTES, 1024).

%% we're passing port stdout/stderr messages to log after delay of
%% INTERVAL milliseconds. Dropping messages once MAX_MESSAGES is
%% reached. Thus we're limiting rate of messages to
%% MAX_MESSAGES/INTERVAL msg/millisecond
%%
%% The following gives us 300 lines/second in bursts up to 60 lines
-define(MAX_MESSAGES, 60). % Max messages per interval
-define(INTERVAL, 200). % Interval over which to throttle

-include_lib("eunit/include/eunit.hrl").

%% Server state
-record(state, {port :: port() | pid() | undefined,
                params :: tuple(),
                messages,
                logger :: atom(),
                log_tref :: timer:tref(),
                log_buffer = [],
                dropped=0 :: non_neg_integer()}).

%% API

start_link(Fun) ->
    gen_server:start_link(?MODULE,
                          Fun, []).
start_link_named(Name, Fun) ->
    gen_server:start_link({local, Name}, ?MODULE, Fun, []).

is_active(Pid) ->
    gen_server:call(Pid, is_active, infinity).

activate(Pid) ->
    gen_server:call(Pid, activate, infinity).

%% gen_server callbacks
init(Fun) ->
    {Name, _Cmd, _Args, Opts} = Params = Fun(),
    process_flag(trap_exit, true), % Make sure terminate gets called
    Opts2 = lists:foldl(fun proplists:delete/2, Opts,
                        [port_server_dont_start, log]),
    DontStart = proplists:get_bool(port_server_dont_start, Opts),
    Log = proplists:get_value(log, Opts),
    Params2 = erlang:setelement(4, Params, Opts2),

    State =
        case Log of
            undefined ->
                #state{};
            _ when is_list(Log) ->
                Sink = Logger = Name,

                ok = ns_server:start_disk_sink(Sink, Log),

                ale:stop_logger(Logger),
                ok = ale:start_logger(Logger, debug, ale_noop_formatter),

                ok = ale:add_sink(Logger, Sink, debug),

                #state{logger=Name}
        end,

    Port = case DontStart of
               false -> port_open(Params2);
               true -> undefined
           end,
    {ok, State#state{port = Port,
                     params = Params2,
                     messages = ringbuffer:new(?KEEP_MESSAGES_BYTES)}}.

handle_info({send_to_port, Msg}, #state{port = undefined} = State) ->
    ?log_debug("Got send_to_port when there's no port running yet. Will kill myself."),
    {stop, {unexpected_send_to_port, {send_to_port, Msg}}, State};
handle_info({send_to_port, Msg}, State) ->
    ?log_debug("Sending the following to port: ~p", [Msg]),
    port_write(State#state.port, Msg),
    {noreply, State};
handle_info({Port, {data, Data}}, #state{port = Port,
                                         logger = Logger} = State) ->
    %% Store the last messages in case of a crash
    Msg = extract_message(Data),
    Messages = ringbuffer:add(Msg, byte_size(Msg), State#state.messages),
    State1 = State#state{messages=Messages},

    NewState =
        case Logger of
            undefined ->
                {Buf, Dropped} = case {State#state.log_buffer, State#state.dropped} of
                                     {B, D} when length(B) < ?MAX_MESSAGES ->
                                         {[Msg|B], D};
                                     {B, D} ->
                                         {B, D + 1}
                                 end,
                TRef = case State#state.log_tref of
                           undefined ->
                               timer2:send_after(?INTERVAL, log);
                           T ->
                               T
                       end,
                State1#state{log_buffer=Buf, log_tref=TRef, dropped=Dropped};
            _ ->
                ale:debug(Logger, [Msg, $\n]),
                State1
        end,

    port_deliver(Port),
    {noreply, NewState};
handle_info(log, State) ->
    State1 = log(State),
{'fraction_non_alphanumeric': 0.11575010109179135, 'fraction_numerical': 0.00909826122118884, 'mean_word_length': 2.051511412708205, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6332749', 'n_tokens_mistral': 2998, 'n_tokens_neox': 2827, 'n_words': 776}