1 -module(beam_stats_producer).
3 -behaviour(gen_server).
12 , sync_produce_consume/0
15 %% gen_server callbacks
25 %% ============================================================================
27 %% ============================================================================
29 -define(PRODUCE_SYNC , produce_sync).
30 -define(PRODUCE_ASYNC , produce_async).
33 { consumers = ordsets:new() :: ordsets:ordset(pid())
34 , stats_state :: beam_stats_state:t()
40 %% ============================================================================
42 %% ============================================================================
45 gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
47 -spec subscribe(pid()) ->
50 gen_server:cast(?MODULE, {subscribe, PID}).
52 -spec unsubscribe(pid()) ->
55 gen_server:cast(?MODULE, {unsubscribe, PID}).
57 -spec sync_produce_consume() ->
59 sync_produce_consume() ->
60 {} = gen_server:call(?MODULE, ?PRODUCE_SYNC).
62 %% ============================================================================
63 %% gen_server callbacks (unused)
64 %% ============================================================================
66 code_change(_OldVsn, State, _Extra) ->
69 terminate(_Reason, _State) ->
72 %% ============================================================================
73 %% gen_server callbacks
74 %% ============================================================================
77 ok = schedule_first_production(),
78 Consumers = ordsets:new(),
79 StatsState = beam_stats_state:new(),
80 {ok, #state{consumers=Consumers, stats_state=StatsState}}.
82 handle_cast({subscribe, PID}, #state{consumers=Consumers1}=State) ->
83 Consumers2 = ordsets:add_element(PID, Consumers1),
84 {noreply, State#state{consumers=Consumers2}};
86 handle_cast({unsubscribe, PID}, #state{consumers=Consumers1}=State) ->
87 Consumers2 = ordsets:del_element(PID, Consumers1),
88 {noreply, State#state{consumers=Consumers2}}.
90 handle_call(?PRODUCE_SYNC, _From, State1) ->
91 State2 = produce_sync(State1),
94 handle_info(?PRODUCE_ASYNC, #state{}=State1) ->
95 State2 = produce_async(State1),
96 ok = schedule_next_production(),
99 %% ============================================================================
101 %% ============================================================================
103 -spec produce_sync(state()) ->
105 produce_sync(#state{}=State) ->
106 produce(State, fun beam_stats_consumer:consume_sync/2).
108 -spec produce_async(state()) ->
110 produce_async(#state{}=State) ->
111 produce(State, fun beam_stats_consumer:consume_async/2).
113 -spec produce(state(), fun((pid(), term()) -> ok)) ->
117 { consumers = ConsumersSet
118 , stats_state = StatsState1
122 StatsState2 = beam_stats_state:update(StatsState1),
123 Stats = beam_stats_state:export(StatsState2),
124 ConsumersList = ordsets:to_list(ConsumersSet),
125 Send = fun (Consumer) -> MsgSendFun(Consumer, Stats) end,
126 ok = lists:foreach(Send, ConsumersList),
127 State#state{stats_state = StatsState2}.
129 -spec schedule_first_production() ->
131 schedule_first_production() ->
132 _ = self() ! ?PRODUCE_ASYNC,
135 -spec schedule_next_production() ->
137 schedule_next_production() ->
138 ProductionInterval = beam_stats_config:production_interval(),
139 _ = erlang:send_after(ProductionInterval, self(), ?PRODUCE_ASYNC),