From 411f6240932aab3721ff842c674d9610216cfd88 Mon Sep 17 00:00:00 2001
From: Micael Karlberg <bmk@erlang.org>
Date: Tue, 10 Jan 2012 14:52:00 +0100
Subject: [snmp/agent] Synchronization feature added
OTP-9851
---
lib/snmp/src/agent/snmpa_agent.erl | 110 +++++++++++++++++++++++++++++++++++--
lib/snmp/src/agent/snmpa_vacm.erl | 77 +++++++++++++++-----------
2 files changed, 150 insertions(+), 37 deletions(-)
(limited to 'lib/snmp/src')
diff --git a/lib/snmp/src/agent/snmpa_agent.erl b/lib/snmp/src/agent/snmpa_agent.erl
index 46c634969d..b78f520c50 100644
--- a/lib/snmp/src/agent/snmpa_agent.erl
+++ b/lib/snmp/src/agent/snmpa_agent.erl
@@ -69,7 +69,7 @@
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3, tr_var/2, tr_varbind/1,
handle_pdu/7, worker/2, worker_loop/1,
- do_send_trap/7, do_send_trap/8]).
+ do_send_trap/7, do_send_trap/8, serialize/2, serialize/3]).
-include("snmpa_internal.hrl").
@@ -271,6 +271,14 @@ update_mibs_cache_age(Agent, Age) ->
call(Agent, {mibs_cache_request, {update_age, Age}}).
+serialize(Tag, Exec) ->
+ serialize(snmp_master_agent, Tag, Exec).
+
+serialize(Agent, Tag, Exec)
+ when (is_pid(Agent) orelse is_atom(Agent)) andalso is_function(Exec, 0) ->
+ call(Agent, {serialize, Tag, Exec}).
+
+
increment_counter(Counter, Initial, Max) ->
%% This is to make sure no one else increments our counter
Key = {Counter, self()},
@@ -934,6 +942,7 @@ handle_info({'EXIT', Pid, Reason}, S) ->
end,
{noreply, S}
end;
+
handle_info({'DOWN', Ref, process, Pid, {mibs_cache_reply, Reply}},
#state{mibs_cache_request = {Pid, Ref, From}} = S) ->
?vlog("reply from the mibs cache request handler (~p): ~n~p",
@@ -941,6 +950,11 @@ handle_info({'DOWN', Ref, process, Pid, {mibs_cache_reply, Reply}},
gen_server:reply(From, Reply),
{noreply, S#state{mibs_cache_request = undefined}};
+handle_info({'DOWN', Mon, process, _Pid, {serializer_done, Tag}}, S) ->
+ ?vlog("handle_info(DOWN) -> serializer done (~w)", [Tag]),
+ handle_serializer_down(Mon, Tag),
+ {noreply, S};
+
handle_info(Info, S) ->
warning_msg("received unexpected info: ~n~p", [Info]),
{noreply, S}.
@@ -1283,27 +1297,33 @@ handle_call({me_of, Oid}, _From, S) ->
{reply, Reply, S};
handle_call(get_log_type, _From, S) ->
- ?vlog("get_log_type", []),
+ ?vlog("handle_call(get_log_type) -> entry with", []),
Reply = handle_get_log_type(S),
{reply, Reply, S};
handle_call({set_log_type, NewType}, _From, S) ->
- ?vlog("set_log_type -> "
+ ?vlog("handle_call(set_log_type) -> entry with"
"~n NewType: ~p", [NewType]),
Reply = handle_set_log_type(S, NewType),
{reply, Reply, S};
handle_call(get_request_limit, _From, S) ->
- ?vlog("get_request_limit", []),
+ ?vlog("handle_call(get_request_limit) -> entry with", []),
Reply = handle_get_request_limit(S),
{reply, Reply, S};
handle_call({set_request_limit, NewLimit}, _From, S) ->
- ?vlog("set_request_limit -> "
+ ?vlog("handle_call(set_request_limit) -> entry with"
"~n NewLimit: ~p", [NewLimit]),
Reply = handle_set_request_limit(S, NewLimit),
{reply, Reply, S};
-
+
+handle_call({serialize, Tag, Exec}, From, S) ->
+ ?vlog("handle_call(serialize) -> entry with"
+ "~n Tag: ~p", [Tag]),
+ handle_serialize(Tag, Exec, From),
+ {noreply, S};
+
handle_call(stop, _From, S) ->
{stop, normal, ok, S};
@@ -3859,6 +3879,7 @@ mapfoldl(F, Eas, Accu0, [Hd|Tail]) ->
{Accu2,[R|Rs]};
mapfoldl(_F, _Eas, Accu, []) -> {Accu,[]}.
+
%%-----------------------------------------------------------------
%% Runtime debugging of the agent.
%%-----------------------------------------------------------------
@@ -3983,6 +4004,83 @@ handle_set_request_limit(_, _) ->
{error, not_supported}.
+%% ------------------------------------------------------------------------
+%% Funtions handling the serialize mechanism
+%% ------------------------------------------------------------------------
+
+serialize_key(Mon) -> {Mon, serialize}.
+storage_key(Tag) -> {Tag, storage}.
+
+start_serializer(Exec, Tag) ->
+ Serializer = fun() ->
+ (catch Exec()),
+ exit({serializer_done, Tag})
+ end,
+ {_Pid, Mon} = erlang:spawn_opt(Serializer, [monitor]),
+ Mon.
+
+handle_serialize(Tag, Exec, From) ->
+ StorageKey = storage_key(Tag),
+ case ets:lookup(snmp_agent_table, StorageKey) of
+ [] ->
+ %% We are first
+ ?vtrace("handle_serialize -> nothing found"
+ "~n (for ~p)", [StorageKey]),
+ Mon = start_serializer(Exec, Tag),
+ SerializeKey = serialize_key(Mon),
+ %% These are waiting for reply
+ ets:insert(snmp_agent_table, {SerializeKey, [From]}),
+ %% These are waiting for exec: That is none at the moment
+ ets:insert(snmp_agent_table, {StorageKey, {Exec, []}}),
+ ok;
+ [{_StorageKey, {_Exec, Froms}}] ->
+ %% There is a process already running, so store the request
+ ?vtrace("handle_serialize -> found"
+ "~n (for ~p)"
+ "~n Forms: ~p", [StorageKey]),
+ NewStorage = {Exec, [From|Froms]},
+ ets:insert(snmp_agent_table, {StorageKey, NewStorage}),
+ ok
+ end.
+
+
+deliver_serialize_reply([From]) ->
+ %% This is the normal case, so we handle it here to optimize
+ gen_server:reply(From, ok);
+deliver_serialize_reply(SerializeReqs) ->
+ lists:foreach(fun(From) -> gen_server:reply(From, ok) end,
+ lists:reverse(SerializeReqs)).
+
+handle_serializer_down(Mon, Tag) ->
+ SerializeKey = serialize_key(Mon),
+ [{_, SerializeReqs}] = ets:lookup(snmp_agent_table, SerializeKey),
+ ?vtrace("handle_serializer_down -> found"
+ "~n (for ~p)"
+ "~n ~p",
+ [SerializeKey, SerializeReqs]),
+ deliver_serialize_reply(SerializeReqs),
+ ets:delete(snmp_agent_table, SerializeKey),
+ StorageKey = storage_key(Tag),
+ case ets:lookup(snmp_agent_table, StorageKey) of
+ [{_StorageKey, {_Exec, []}}] ->
+ %% We are done
+ ?vtrace("handle_serializer_down -> we are done", []),
+ ets:delete(snmp_agent_table, StorageKey),
+ ok;
+ [{_StorageKey, {Exec, Froms}}] ->
+ %% We have some waiting, so start a new one
+ ?vtrace("handle_serializer_down -> waiting requests for ~p: "
+ "~n ~p", [StorageKey, Froms]),
+ Mon2 = start_serializer(Exec, Tag),
+ SerializeKey2 = serialize_key(Mon2),
+ ets:insert(snmp_agent_table, {SerializeKey2, Froms}),
+ ets:insert(snmp_agent_table, {StorageKey, {Exec, []}}),
+ ok
+ end.
+
+
+%% ------------------------------------------------------------------------
+
agent_info(#state{worker = W, set_worker = SW}) ->
case (catch get_agent_info(W, SW)) of
Info when is_list(Info) ->
diff --git a/lib/snmp/src/agent/snmpa_vacm.erl b/lib/snmp/src/agent/snmpa_vacm.erl
index c31b8e61a3..efe6378105 100644
--- a/lib/snmp/src/agent/snmpa_vacm.erl
+++ b/lib/snmp/src/agent/snmpa_vacm.erl
@@ -276,45 +276,60 @@ dump_table(_) ->
%% If there is already a running dumper process, instead increment
%% the dump_request counter.
%% When the dumper process exits, the master agent checks the
-%% the dump_request counter, and if that is greated than zero,
+%% the dump_request counter, and if that is greater than zero,
%% create another dumper process and resets the counter.
-%% In this way the dumping is serializede, but the master-agent
-%% process is not burdened by the dumping.
+%% In this way the dumping is serialized, but the master-agent
+%% process is not burdend by the dumping.
%% </SUGGESTION>
dump_table() ->
- [{_, FName}] = ets:lookup(snmp_agent_table, snmpa_vacm_file),
- TmpName = unique(FName),
- case ets:tab2file(snmpa_vacm, TmpName) of
- ok ->
- case file:rename(TmpName, FName) of
- ok ->
- ok;
- Else -> % What is this? Undocumented return code...
- user_err("Warning: could not move VACM db ~p"
- " (~p)", [FName, Else])
- end;
- {error, Reason} ->
- user_err("Warning: could not save vacm db ~p (~p)",
- [FName, Reason])
- end.
+ %% The dumper fun is executed in a specially started process,
+ %% that does that one thing and then exits.
+ %% Also, to prevent the system to run "wild" (keep calling
+ %% dump function before they are done), the agents serialize
+ %% function return when that dump is done!
+ Dumper =
+ fun() ->
+ [{_, FName}] = ets:lookup(snmp_agent_table, snmpa_vacm_file),
+ %% TmpName = FName ++ ".tmp",
+ TmpName = unique_name(FName),
+ case ets:tab2file(snmpa_vacm, TmpName) of
+ ok ->
+ case file:rename(TmpName, FName) of
+ ok ->
+ ok;
+ Else -> % What is this? Undocumented return code...
+ user_err("Warning: could not move VACM db ~p"
+ " (~p)", [FName, Else])
+ end;
+ {error, Reason} ->
+ user_err("Warning: could not save vacm db ~p (~p)",
+ [FName, Reason])
+ end
+ end,
+ snmpa_agent:serialize(snmpa_vacm_dump_request, Dumper).
+
%% This little thing is an attempt to create a "unique" filename
%% in order to minimize the risk of two processes at the same
%% time dumping the table.
-unique(Pre) ->
- {A, B, C} = os:timestamp(),
- {D, _} = erlang:statistics(reductions),
- {E, _} = erlang:statistics(runtime),
- {F, _} = erlang:statistics(wall_clock),
- {G, H, _} = erlang:statistics(garbage_collection),
- Data = [A, B, C, D, E, F, G, H],
- unique(Pre, Data, 0).
-
-unique(Pre, [], Unique) ->
+%% The serialization handled by the agent does this much better,
+%% but this also gives us a "timestamp" which could be usefull for
+%% debugging reasons.
+unique_name(Pre) ->
+ unique_name(Pre, os:timestamp()).
+
+unique_name(Pre, {_A, _B, C} = Timestamp) ->
+ {Date, Time} = calendar:now_to_datetime(Timestamp),
+ {YYYY, MM, DD} = Date,
+ {Hour, Min, Sec} = Time,
+ FormatDate =
+ io_lib:format("~.4w~.2.0w~.2.0w_~.2.0w~.2.0w~.2.0w_~w",
+ [YYYY, MM, DD, Hour, Min, Sec, round(C/1000)]),
+ unique_name2(Pre, FormatDate).
+
+unique_name2(Pre, FormatedDate) ->
PidPart = unique_pid(),
- lists:flatten(io_lib:format("~s.~s~w.tmp", [Pre, PidPart, Unique]));
-unique(Pre, [H|T], Unique) ->
- unique(Pre, T, Unique bxor H).
+ lists:flatten(io_lib:format("~s.~s~s.tmp", [Pre, PidPart, FormatedDate])).
unique_pid() ->
case string:tokens(pid_to_list(self()), [$<,$.,$>]) of
--
cgit v1.2.3