Skip to content

Commit e469c60

Browse files
nickvanickva
committed
Replace khash with maps in couch_event_server
Noticed when debugging an OTP 25 memory leak that couch_event_server didn't behave well when the system was overloaded. It blocked and prevent even process info inspection or is_alive checks: ``` > s:pinfo(whereis(couch_event_server)). ...blocked... > erlang:is_process_alive(whereis(couch_event_server)). ...blocked... ``` Added almost 100% test coverage for the couch_event_server module.
1 parent 6e54100 commit e469c60

File tree

2 files changed

+215
-73
lines changed

2 files changed

+215
-73
lines changed

‎src/couch_event/src/couch_event.app.src

+1-1
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,6 @@
1717
couch_event_sup,
1818
couch_event_server
1919
]},
20-
{applications, [kernel, stdlib, khash, couch_log, config]},
20+
{applications, [kernel, stdlib, couch_log, config]},
2121
{mod, {couch_event_app, []}}
2222
]}.

‎src/couch_event/src/couch_event_server.erl

+214-72
Original file line numberDiff line numberDiff line change
@@ -24,8 +24,6 @@
2424
handle_info/2
2525
]).
2626

27-
-include("couch_event_int.hrl").
28-
2927
-record(st, {
3028
by_pid,
3129
by_dbname
@@ -35,34 +33,28 @@ start_link() ->
3533
gen_server:start_link({local, ?MODULE}, ?MODULE, nil, []).
3634

3735
init(_) ->
38-
{ok, ByPid} = khash:new(),
39-
{ok, ByDbName} = khash:new(),
4036
{ok, #st{
41-
by_pid = ByPid,
42-
by_dbname = ByDbName
37+
by_pid = #{},
38+
by_dbname = #{}
4339
}}.
4440

4541
handle_call({register, Pid, NewDbNames}, _From, St) ->
46-
case khash:get(St#st.by_pid, Pid) of
42+
case maps:get(Pid, St#st.by_pid, undefined) of
4743
undefined ->
4844
NewRef = erlang:monitor(process, Pid),
49-
register(St, NewRef, Pid, NewDbNames);
45+
{reply, ok, register(St, NewRef, Pid, NewDbNames)};
5046
{ReuseRef, OldDbNames} ->
5147
unregister(St, Pid, OldDbNames),
52-
register(St, ReuseRef, Pid, NewDbNames)
53-
end,
54-
{reply, ok, St};
55-
handle_call({unregister, Pid}, _From, St) ->
56-
Reply =
57-
case khash:get(St#st.by_pid, Pid) of
58-
undefined ->
59-
not_registered;
60-
{Ref, OldDbNames} ->
61-
unregister(St, Pid, OldDbNames),
62-
erlang:demonitor(Ref, [flush]),
63-
ok
64-
end,
65-
{reply, Reply, St};
48+
{reply, ok, register(St, ReuseRef, Pid, NewDbNames)}
49+
end;
50+
handle_call({unregister, Pid}, _From, #st{by_pid = ByPid} = St) ->
51+
case maps:get(Pid, ByPid, undefined) of
52+
undefined ->
53+
{reply, not_registered, St};
54+
{Ref, OldDbNames} ->
55+
erlang:demonitor(Ref, [flush]),
56+
{reply, ok, unregister(St, Pid, OldDbNames)}
57+
end;
6658
handle_call(Msg, From, St) ->
6759
couch_log:notice("~s ignoring call ~w from ~w", [?MODULE, Msg, From]),
6860
{reply, ignored, St}.
@@ -74,68 +66,218 @@ handle_cast(Msg, St) ->
7466
couch_log:notice("~s ignoring cast ~w", [?MODULE, Msg]),
7567
{noreply, St}.
7668

77-
handle_info({'DOWN', Ref, process, Pid, _Reason}, St) ->
78-
case khash:get(St#st.by_pid, Pid) of
79-
{Ref, OldDbNames} ->
80-
unregister(St, Pid, OldDbNames);
81-
undefined ->
82-
ok
83-
end,
84-
{noreply, St};
69+
handle_info({'DOWN', Ref, process, Pid, _Reason}, #st{by_pid = ByPid} = St) ->
70+
#{Pid := {Ref, OldDbNames}} = ByPid,
71+
{noreply, unregister(St, Pid, OldDbNames)};
8572
handle_info(Msg, St) ->
8673
couch_log:notice("~s ignoring info ~w", [?MODULE, Msg]),
8774
{noreply, St}.
8875

89-
notify_listeners(ByDbName, DbName, Event) ->
76+
notify_listeners(#{} = ByDbName, DbName, Event) ->
9077
Msg = {'$couch_event', DbName, Event},
91-
notify_listeners(khash:get(ByDbName, all_dbs), Msg),
92-
notify_listeners(khash:get(ByDbName, DbName), Msg).
78+
notify_listeners(maps:get(all_dbs, ByDbName, undefined), Msg),
79+
notify_listeners(maps:get(DbName, ByDbName, undefined), Msg).
9380

9481
notify_listeners(undefined, _) ->
9582
ok;
96-
notify_listeners(Listeners, Msg) ->
97-
khash:fold(
98-
Listeners,
99-
fun(Pid, _, _) ->
100-
Pid ! Msg,
101-
nil
102-
end,
103-
nil
104-
).
83+
notify_listeners(#{} = Listeners, Msg) ->
84+
maps:foreach(fun(Pid, _) -> Pid ! Msg end, Listeners).
10585

106-
register(St, Ref, Pid, DbNames) ->
107-
khash:put(St#st.by_pid, Pid, {Ref, DbNames}),
108-
lists:foreach(
109-
fun(DbName) ->
110-
add_listener(St#st.by_dbname, DbName, Pid)
111-
end,
112-
DbNames
113-
).
86+
register(#st{by_pid = ByPid, by_dbname = ByDbName} = St, Ref, Pid, DbNames) ->
87+
FoldFun = fun(DbName, Acc) -> add_listener(Acc, DbName, Pid) end,
88+
ByDbName1 = lists:foldl(FoldFun, ByDbName, DbNames),
89+
St#st{by_pid = ByPid#{Pid => {Ref, DbNames}}, by_dbname = ByDbName1}.
11490

115-
add_listener(ByDbName, DbName, Pid) ->
116-
case khash:lookup(ByDbName, DbName) of
117-
{value, Listeners} ->
118-
khash:put(Listeners, Pid, nil);
119-
not_found ->
120-
{ok, NewListeners} = khash:new(),
121-
khash:put(NewListeners, Pid, nil),
122-
khash:put(ByDbName, DbName, NewListeners)
91+
add_listener(#{} = ByDbName, DbName, Pid) ->
92+
case maps:get(DbName, ByDbName, not_found) of
93+
#{} = Listeners -> ByDbName#{DbName => Listeners#{Pid => nil}};
94+
not_found -> ByDbName#{DbName => #{Pid => nil}}
12395
end.
12496

125-
unregister(St, Pid, OldDbNames) ->
126-
ok = khash:del(St#st.by_pid, Pid),
127-
lists:foreach(
128-
fun(DbName) ->
129-
rem_listener(St#st.by_dbname, DbName, Pid)
130-
end,
131-
OldDbNames
97+
unregister(#st{by_pid = ByPid, by_dbname = ByDbName} = St, Pid, OldDbNames) ->
98+
FoldFun = fun(DbName, Acc) -> rem_listener(Acc, DbName, Pid) end,
99+
ByDbName1 = lists:foldl(FoldFun, ByDbName, OldDbNames),
100+
St#st{by_pid = maps:remove(Pid, ByPid), by_dbname = ByDbName1}.
101+
102+
rem_listener(#{} = ByDbName, DbName, Pid) ->
103+
#{DbName := Listeners} = ByDbName,
104+
Listeners1 = maps:remove(Pid, Listeners),
105+
case map_size(Listeners1) of
106+
0 -> maps:remove(DbName, ByDbName);
107+
_ -> ByDbName#{DbName := Listeners1}
108+
end.
109+
110+
-ifdef(TEST).
111+
112+
-include_lib("couch/include/couch_eunit.hrl").
113+
114+
couch_event_test_() ->
115+
{
116+
foreach,
117+
fun setup/0,
118+
fun teardown/1,
119+
[
120+
?TDEF_FE(t_register_proc_basic),
121+
?TDEF_FE(t_unregister_proc_basic),
122+
?TDEF_FE(t_unregister_on_death),
123+
?TDEF_FE(t_notify_basic),
124+
?TDEF_FE(t_notify_all_dbs),
125+
?TDEF_FE(t_register_multiple),
126+
?TDEF_FE(t_reregister),
127+
?TDEF_FE(t_invalid_gen_server_messages)
128+
]
129+
}.
130+
131+
setup() ->
132+
meck:expect(couch_log, notice, fun(_, _) -> ok end),
133+
{ok, SPid} = start_link(),
134+
Listeners = [start_listener(), start_listener(), start_listener()],
135+
{SPid, Listeners}.
136+
137+
teardown({SPid, Listeners}) ->
138+
[kill_sync(Pid) || Pid <- [SPid | Listeners]],
139+
meck:unload(),
140+
ok.
141+
142+
t_register_proc_basic({_, [LPid | _]}) ->
143+
?assertEqual(ok, reg(LPid, [db])),
144+
#st{by_pid = ByPid, by_dbname = ByName} = state(),
145+
?assertMatch(#{LPid := {_, [db]}}, ByPid),
146+
?assertMatch(#{db := #{LPid := nil}}, ByName).
147+
148+
t_unregister_proc_basic({_, [LPid | _]}) ->
149+
?assertEqual(ok, reg(LPid, [db])),
150+
#st{by_pid = ByPid, by_dbname = ByName} = state(),
151+
?assertMatch(#{LPid := {_, [db]}}, ByPid),
152+
?assertMatch(#{db := #{LPid := nil}}, ByName),
153+
?assertEqual(not_registered, unreg(self())),
154+
?assertEqual(ok, unreg(LPid)),
155+
#st{by_pid = #{}, by_dbname = #{}} = state().
156+
157+
t_unregister_on_death({_, [LPid | _]}) ->
158+
?assertEqual(ok, reg(LPid, [db])),
159+
#st{by_pid = ByPid, by_dbname = ByName} = state(),
160+
?assertMatch(#{LPid := {_, [db]}}, ByPid),
161+
?assertMatch(#{db := #{LPid := nil}}, ByName),
162+
kill_sync(LPid),
163+
test_util:wait(fun() ->
164+
#st{by_pid = Pids} = state(),
165+
case map_size(Pids) of
166+
0 -> ok;
167+
N when is_integer(N), N > 1 -> wait
168+
end
169+
end),
170+
?assertEqual(#st{by_pid = #{}, by_dbname = #{}}, state()).
171+
172+
t_notify_basic({_, [LPid | _]}) ->
173+
reg(LPid, [db]),
174+
notify(other_db, foo),
175+
notify(db, bar),
176+
?assertEqual([{db, bar}], wait(LPid, 1)).
177+
178+
t_notify_all_dbs({_, [LPid | _]}) ->
179+
ok = reg(LPid, [all_dbs]),
180+
notify(db, bar),
181+
?assertEqual([{db, bar}], wait(LPid, 1)).
182+
183+
t_register_multiple({_, [L1, L2, L3 | _]}) ->
184+
reg(L1, [dbx, dby]),
185+
reg(L2, [dby, all_dbs]),
186+
reg(L3, [dbx, all_dbs]),
187+
% all_dbs broadcast works
188+
notify(dbw, e1),
189+
?assertEqual([{dbw, e1}], wait(L2, 1)),
190+
?assertEqual([{dbw, e1}], wait(L3, 1)),
191+
% dbx updated, only dbx and all_dbs are notified
192+
notify(dbx, e2),
193+
?assertEqual([{dbx, e2}], wait(L1, 1)),
194+
?assertEqual([{dbx, e2}], wait(L2, 1)),
195+
% L3 gets 2 notifications once for all_dbs and dbx
196+
?assertEqual([{dbx, e2}, {dbx, e2}], wait(L3, 2)),
197+
unreg(L1),
198+
unreg(L2),
199+
unreg(L3),
200+
#st{by_pid = #{}, by_dbname = #{}} = state().
201+
202+
t_reregister({_, [LPid | _]}) ->
203+
?assertEqual(ok, reg(LPid, [dbx, dby])),
204+
?assertEqual(ok, reg(LPid, [dbz, dbw])),
205+
#st{by_pid = ByPid, by_dbname = ByName} = state(),
206+
?assertEqual(1, map_size(ByPid)),
207+
?assertMatch(#{LPid := {_, [dbz, dbw]}}, ByPid),
208+
?assertMatch(
209+
#{
210+
dbz := #{LPid := nil},
211+
dbw := #{LPid := nil}
212+
},
213+
ByName
132214
).
133215

134-
rem_listener(ByDbName, DbName, Pid) ->
135-
{value, Listeners} = khash:lookup(ByDbName, DbName),
136-
khash:del(Listeners, Pid),
137-
Size = khash:size(Listeners),
138-
if
139-
Size > 0 -> ok;
140-
true -> khash:del(ByDbName, DbName)
216+
t_invalid_gen_server_messages(_) ->
217+
meck:reset(couch_log),
218+
whereis(?MODULE) ! random_msg,
219+
gen_server:cast(?MODULE, bad_cast),
220+
?assertEqual(ignored, gen_server:call(?MODULE, bad_call)),
221+
?assertEqual(3, meck:num_calls(couch_log, notice, 2)).
222+
223+
-record(lst, {
224+
events = [],
225+
wait_cnt = 0,
226+
wait_pid = undefined
227+
}).
228+
229+
kill_sync(Pid) ->
230+
unlink(Pid),
231+
Ref = erlang:monitor(process, Pid),
232+
exit(Pid, kill),
233+
receive
234+
{'DOWN', Ref, _, _, _} -> ok
141235
end.
236+
237+
reg(Pid, Dbs) ->
238+
gen_server:call(?MODULE, {register, Pid, Dbs}).
239+
240+
unreg(Pid) ->
241+
gen_server:call(?MODULE, {unregister, Pid}).
242+
243+
notify(Db, Event) ->
244+
gen_server:cast(?MODULE, {notify, Db, Event}).
245+
246+
state() ->
247+
sys:get_state(?MODULE).
248+
249+
% Listeners receive and save their notification events. {get_events, MinNum,
250+
% self()} call will block until the listener had received at least MinNum
251+
% events. We use this since event notification is asynchronous.
252+
253+
start_listener() ->
254+
spawn(fun() -> loop(#lst{}) end).
255+
256+
loop(#lst{} = St0) ->
257+
St = #lst{events = Events, wait_pid = WaitPid} = respond(St0),
258+
receive
259+
{'$couch_event', Db, Ev} ->
260+
loop(St#lst{events = [{Db, Ev} | Events]});
261+
{get_events, _, WPid} when is_pid(WaitPid) ->
262+
WPid ! {error, existing_waiter};
263+
{get_events, WCnt, WPid} ->
264+
loop(St#lst{wait_cnt = WCnt, wait_pid = WPid})
265+
end.
266+
267+
respond(#lst{events = Events, wait_cnt = Cnt, wait_pid = Pid} = St) ->
268+
case {length(Events) >= Cnt, is_pid(Pid)} of
269+
{true, true} ->
270+
Pid ! Events,
271+
St#lst{events = [], wait_cnt = 0, wait_pid = undefined};
272+
{_, _} ->
273+
St
274+
end.
275+
276+
wait(LPid, N) ->
277+
LPid ! {get_events, N, self()},
278+
receive
279+
{error, Error} -> {error, Error};
280+
Events when is_list(Events) -> Events
281+
end.
282+
283+
-endif.

0 commit comments

Comments
 (0)