src/ddoc_cache/src/ddoc_cache_entry.erl - couchdb - Git at Google

 % 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(ddoc_cache_entry).
 -behaviour(gen_server).
 -vsn(1).


 -export([
     dbname/1,
     ddocid/1,
     recover/1,
     insert/2,

     start_link/2,
     shutdown/1,
     open/2,
     accessed/1,
     refresh/1
 ]).

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

 -export([
     do_open/1
 ]).


 -include("ddoc_cache.hrl").


 -ifndef(TEST).
 -define(ENTRY_SHUTDOWN_TIMEOUT, 5000).
 -else.
 -define(ENTRY_SHUTDOWN_TIMEOUT, 500).
 -endif.


 -record(st, {
     key,
     val,
     opener,
     waiters,
     ts,
     accessed
 }).


 dbname({Mod, Arg}) ->
     Mod:dbname(Arg).


 ddocid({Mod, Arg}) ->
     Mod:ddocid(Arg).


 recover({Mod, Arg}) ->
     Mod:recover(Arg).


 insert({Mod, Arg}, Value) ->
     Mod:insert(Arg, Value).


 start_link(Key, Default) ->
     Pid = proc_lib:spawn_link(?MODULE, init, [{Key, Default}]),
     {ok, Pid}.


 shutdown(Pid) ->
     Ref = erlang:monitor(process, Pid),
     ok = gen_server:cast(Pid, shutdown),
     receive
         {'DOWN', Ref, process, Pid, normal} ->
             ok;
         {'DOWN', Ref, process, Pid, Reason} ->
             erlang:exit(Reason)
     after ?ENTRY_SHUTDOWN_TIMEOUT ->
         erlang:demonitor(Ref, [flush]),
         erlang:exit({timeout, {entry_shutdown, Pid}})
     end.


 open(Pid, Key) ->
     try
         Resp = gen_server:call(Pid, open),
         case Resp of
             {open_ok, Val} ->
                 Val;
             {open_error, {T, R, S}} ->
                 erlang:raise(T, R, S)
         end
     catch
         error:database_does_not_exist ->
             erlang:error(database_does_not_exist);
         exit:_ ->
             % Its possible that this process was evicted just
             % before we tried talking to it. Just fallback
             % to a standard recovery
             recover(Key)
     end.


 accessed(Pid) ->
     gen_server:cast(Pid, accessed).


 refresh(Pid) ->
     gen_server:cast(Pid, force_refresh).


 init({Key, undefined}) ->
     true = ets:update_element(?CACHE, Key, {#entry.pid, self()}),
     St = #st{
         key = Key,
         opener = spawn_opener(Key),
         waiters = [],
         accessed = 1
     },
     ?EVENT(started, Key),
     gen_server:enter_loop(?MODULE, [], St);

 init({Key, Wrapped}) ->
     Default = ddoc_cache_value:unwrap(Wrapped),
     Updates = [
         {#entry.val, Default},
         {#entry.pid, self()}
     ],
     NewTs = os:timestamp(),
     true = ets:update_element(?CACHE, Key, Updates),
     true = ets:insert(?LRU, {{NewTs, Key, self()}}),
     St = #st{
         key = Key,
         val = {open_ok, {ok, Default}},
         opener = start_timer(),
         waiters = [],
         ts = NewTs,
         accessed = 1
     },
     ?EVENT(default_started, Key),
     gen_server:enter_loop(?MODULE, [], St, hibernate).


 terminate(_Reason, St) ->
     #st{
         key = Key,
         opener = Pid,
         ts = Ts
     } = St,
     % We may have already deleted our cache entry
     % during shutdown
     Pattern = #entry{key = Key, pid = self(), _ = '_'},
     CacheMSpec = [{Pattern, [], [true]}],
     true = ets:select_delete(?CACHE, CacheMSpec) < 2,
     % We may have already deleted our LRU entry
     % during shutdown
     if Ts == undefined -> ok; true ->
         LruMSpec = [{{{Ts, Key, self()}}, [], [true]}],
         true = ets:select_delete(?LRU, LruMSpec) < 2
     end,
     % Blow away any current opener if it exists
     if not is_pid(Pid) -> ok; true ->
         catch exit(Pid, kill)
     end,
     ok.


 handle_call(open, From, #st{opener = Pid} = St) when is_pid(Pid) ->
     NewSt = St#st{
         waiters = [From | St#st.waiters]
     },
     {noreply, NewSt};

 handle_call(open, _From, St) ->
     {reply, St#st.val, St};

 handle_call(Msg, _From, St) ->
     {stop, {bad_call, Msg}, {bad_call, Msg}, St}.


 handle_cast(accessed, St) ->
     ?EVENT(accessed, St#st.key),
     drain_accessed(),
     NewSt = St#st{
         accessed = St#st.accessed + 1
     },
     {noreply, update_lru(NewSt)};

 handle_cast(force_refresh, St) ->
     % If we had frequent design document updates
     % they could end up racing accessed events and
     % end up prematurely evicting this entry from
     % cache. To prevent this we just make sure that
     % accessed is set to at least 1 before we
     % execute a refresh.
     NewSt = if St#st.accessed > 0 -> St; true ->
         St#st{accessed = 1}
     end,
     % We remove the cache entry value so that any
     % new client comes to us for the refreshed
     % value.
     true = ets:update_element(?CACHE, St#st.key, {#entry.val, undefined}),
     handle_cast(refresh, NewSt);

 handle_cast(refresh, #st{accessed = 0} = St) ->
     {stop, normal, St};

 handle_cast(refresh, #st{opener = Ref} = St) when is_reference(Ref) ->
     #st{
         key = Key
     } = St,
     erlang:cancel_timer(Ref),
     NewSt = St#st{
         opener = spawn_opener(Key),
         accessed = 0
     },
     {noreply, NewSt};

 handle_cast(refresh, #st{opener = Pid} = St) when is_pid(Pid) ->
     catch exit(Pid, kill),
     receive
         {'DOWN', _, _, Pid, _} -> ok
     end,
     NewSt = St#st{
         opener = spawn_opener(St#st.key),
         accessed = 0
     },
     {noreply, NewSt};

 handle_cast(shutdown, St) ->
     remove_from_cache(St),
     {stop, normal, St};

 handle_cast(Msg, St) ->
     {stop, {bad_cast, Msg}, St}.


 handle_info({'DOWN', _, _, Pid, Resp}, #st{key = Key, opener = Pid} = St) ->
     case Resp of
         {open_ok, Key, {ok, Val}} ->
             update_cache(St, Val),
             NewSt1 = St#st{
                 val = {open_ok, {ok, Val}},
                 opener = start_timer(),
                 waiters = []
             },
             NewSt2 = update_lru(NewSt1),
             respond(St#st.waiters, {open_ok, {ok, Val}}),
             {noreply, NewSt2};
         {Status, Key, Other} ->
             NewSt = St#st{
                 val = {Status, Other},
                 opener = undefined,
                 waiters = undefined
             },
             remove_from_cache(NewSt),
             respond(St#st.waiters, {Status, Other}),
             {stop, normal, NewSt}
     end;

 handle_info(Msg, St) ->
     {stop, {bad_info, Msg}, St}.


 code_change(_, St, _) ->
     {ok, St}.


 spawn_opener(Key) ->
     {Pid, _} = erlang:spawn_monitor(?MODULE, do_open, [Key]),
     Pid.


 start_timer() ->
     TimeOut = config:get_integer(
             "ddoc_cache", "refresh_timeout", ?REFRESH_TIMEOUT),
     erlang:send_after(TimeOut, self(), {'$gen_cast', refresh}).


 do_open(Key) ->
     try recover(Key) of
         Resp ->
             erlang:exit({open_ok, Key, Resp})
     catch T:R ->
         S = erlang:get_stacktrace(),
         erlang:exit({open_error, Key, {T, R, S}})
     end.


 update_lru(#st{key = Key, ts = Ts} = St) ->
     remove_from_lru(Ts, Key),
     NewTs = os:timestamp(),
     true = ets:insert(?LRU, {{NewTs, Key, self()}}),
     St#st{ts = NewTs}.


 update_cache(#st{val = undefined} = St, Val) ->
     true = ets:update_element(?CACHE, St#st.key, {#entry.val, Val}),
     ?EVENT(inserted, St#st.key);

 update_cache(#st{val = V1} = _St, V2) when {open_ok, {ok, V2}} == V1 ->
     ?EVENT(update_noop, _St#st.key);

 update_cache(St, Val) ->
     true = ets:update_element(?CACHE, St#st.key, {#entry.val, Val}),
     ?EVENT(updated, {St#st.key, Val}).


 remove_from_cache(St) ->
     #st{
         key = Key,
         ts = Ts
     } = St,
     Pattern = #entry{key = Key, pid = self(), _ = '_'},
     CacheMSpec = [{Pattern, [], [true]}],
     1 = ets:select_delete(?CACHE, CacheMSpec),
     remove_from_lru(Ts, Key),
     ?EVENT(removed, St#st.key),
     ok.


 remove_from_lru(Ts, Key) ->
     if Ts == undefined -> ok; true ->
         LruMSpec = [{{{Ts, Key, self()}}, [], [true]}],
         1 = ets:select_delete(?LRU, LruMSpec)
     end.


 drain_accessed() ->
     receive
         {'$gen_cast', accessed} ->
             drain_accessed()
     after 0 ->
         ok
     end.


 respond(Waiters, Resp) ->
     [gen_server:reply(W, Resp) || W <- Waiters].
	% 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(ddoc_cache_entry).
	-behaviour(gen_server).
	-vsn(1).


	-export([
	dbname/1,
	ddocid/1,
	recover/1,
	insert/2,

	start_link/2,
	shutdown/1,
	open/2,
	accessed/1,
	refresh/1
	]).

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

	-export([
	do_open/1
	]).


	-include("ddoc_cache.hrl").


	-ifndef(TEST).
	-define(ENTRY_SHUTDOWN_TIMEOUT, 5000).
	-else.
	-define(ENTRY_SHUTDOWN_TIMEOUT, 500).
	-endif.


	-record(st, {
	key,
	val,
	opener,
	waiters,
	ts,
	accessed
	}).


	dbname({Mod, Arg}) ->
	Mod:dbname(Arg).


	ddocid({Mod, Arg}) ->
	Mod:ddocid(Arg).


	recover({Mod, Arg}) ->
	Mod:recover(Arg).


	insert({Mod, Arg}, Value) ->
	Mod:insert(Arg, Value).


	start_link(Key, Default) ->
	Pid = proc_lib:spawn_link(?MODULE, init, [{Key, Default}]),
	{ok, Pid}.


	shutdown(Pid) ->
	Ref = erlang:monitor(process, Pid),
	ok = gen_server:cast(Pid, shutdown),
	receive
	{'DOWN', Ref, process, Pid, normal} ->
	ok;
	{'DOWN', Ref, process, Pid, Reason} ->
	erlang:exit(Reason)
	after ?ENTRY_SHUTDOWN_TIMEOUT ->
	erlang:demonitor(Ref, [flush]),
	erlang:exit({timeout, {entry_shutdown, Pid}})
	end.


	open(Pid, Key) ->
	try
	Resp = gen_server:call(Pid, open),
	case Resp of
	{open_ok, Val} ->
	Val;
	{open_error, {T, R, S}} ->
	erlang:raise(T, R, S)
	end
	catch
	error:database_does_not_exist ->
	erlang:error(database_does_not_exist);
	exit:_ ->
	% Its possible that this process was evicted just
	% before we tried talking to it. Just fallback
	% to a standard recovery
	recover(Key)
	end.


	accessed(Pid) ->
	gen_server:cast(Pid, accessed).


	refresh(Pid) ->
	gen_server:cast(Pid, force_refresh).


	init({Key, undefined}) ->
	true = ets:update_element(?CACHE, Key, {#entry.pid, self()}),
	St = #st{
	key = Key,
	opener = spawn_opener(Key),
	waiters = [],
	accessed = 1
	},
	?EVENT(started, Key),
	gen_server:enter_loop(?MODULE, [], St);

	init({Key, Wrapped}) ->
	Default = ddoc_cache_value:unwrap(Wrapped),
	Updates = [
	{#entry.val, Default},
	{#entry.pid, self()}
	],
	NewTs = os:timestamp(),
	true = ets:update_element(?CACHE, Key, Updates),
	true = ets:insert(?LRU, {{NewTs, Key, self()}}),
	St = #st{
	key = Key,
	val = {open_ok, {ok, Default}},
	opener = start_timer(),
	waiters = [],
	ts = NewTs,
	accessed = 1
	},
	?EVENT(default_started, Key),
	gen_server:enter_loop(?MODULE, [], St, hibernate).


	terminate(_Reason, St) ->
	#st{
	key = Key,
	opener = Pid,
	ts = Ts
	} = St,
	% We may have already deleted our cache entry
	% during shutdown
	Pattern = #entry{key = Key, pid = self(), _ = '_'},
	CacheMSpec = [{Pattern, [], [true]}],
	true = ets:select_delete(?CACHE, CacheMSpec) < 2,
	% We may have already deleted our LRU entry
	% during shutdown
	if Ts == undefined -> ok; true ->
	LruMSpec = [{{{Ts, Key, self()}}, [], [true]}],
	true = ets:select_delete(?LRU, LruMSpec) < 2
	end,
	% Blow away any current opener if it exists
	if not is_pid(Pid) -> ok; true ->
	catch exit(Pid, kill)
	end,
	ok.


	handle_call(open, From, #st{opener = Pid} = St) when is_pid(Pid) ->
	NewSt = St#st{
	waiters = [From \| St#st.waiters]
	},
	{noreply, NewSt};

	handle_call(open, _From, St) ->
	{reply, St#st.val, St};

	handle_call(Msg, _From, St) ->
	{stop, {bad_call, Msg}, {bad_call, Msg}, St}.


	handle_cast(accessed, St) ->
	?EVENT(accessed, St#st.key),
	drain_accessed(),
	NewSt = St#st{
	accessed = St#st.accessed + 1
	},
	{noreply, update_lru(NewSt)};

	handle_cast(force_refresh, St) ->
	% If we had frequent design document updates
	% they could end up racing accessed events and
	% end up prematurely evicting this entry from
	% cache. To prevent this we just make sure that
	% accessed is set to at least 1 before we
	% execute a refresh.
	NewSt = if St#st.accessed > 0 -> St; true ->
	St#st{accessed = 1}
	end,
	% We remove the cache entry value so that any
	% new client comes to us for the refreshed
	% value.
	true = ets:update_element(?CACHE, St#st.key, {#entry.val, undefined}),
	handle_cast(refresh, NewSt);

	handle_cast(refresh, #st{accessed = 0} = St) ->
	{stop, normal, St};

	handle_cast(refresh, #st{opener = Ref} = St) when is_reference(Ref) ->
	#st{
	key = Key
	} = St,
	erlang:cancel_timer(Ref),
	NewSt = St#st{
	opener = spawn_opener(Key),
	accessed = 0
	},
	{noreply, NewSt};

	handle_cast(refresh, #st{opener = Pid} = St) when is_pid(Pid) ->
	catch exit(Pid, kill),
	receive
	{'DOWN', _, _, Pid, _} -> ok
	end,
	NewSt = St#st{
	opener = spawn_opener(St#st.key),
	accessed = 0
	},
	{noreply, NewSt};

	handle_cast(shutdown, St) ->
	remove_from_cache(St),
	{stop, normal, St};

	handle_cast(Msg, St) ->
	{stop, {bad_cast, Msg}, St}.


	handle_info({'DOWN', _, _, Pid, Resp}, #st{key = Key, opener = Pid} = St) ->
	case Resp of
	{open_ok, Key, {ok, Val}} ->
	update_cache(St, Val),
	NewSt1 = St#st{
	val = {open_ok, {ok, Val}},
	opener = start_timer(),
	waiters = []
	},
	NewSt2 = update_lru(NewSt1),
	respond(St#st.waiters, {open_ok, {ok, Val}}),
	{noreply, NewSt2};
	{Status, Key, Other} ->
	NewSt = St#st{
	val = {Status, Other},
	opener = undefined,
	waiters = undefined
	},
	remove_from_cache(NewSt),
	respond(St#st.waiters, {Status, Other}),
	{stop, normal, NewSt}
	end;

	handle_info(Msg, St) ->
	{stop, {bad_info, Msg}, St}.


	code_change(_, St, _) ->
	{ok, St}.


	spawn_opener(Key) ->
	{Pid, _} = erlang:spawn_monitor(?MODULE, do_open, [Key]),
	Pid.


	start_timer() ->
	TimeOut = config:get_integer(
	"ddoc_cache", "refresh_timeout", ?REFRESH_TIMEOUT),
	erlang:send_after(TimeOut, self(), {'$gen_cast', refresh}).


	do_open(Key) ->
	try recover(Key) of
	Resp ->
	erlang:exit({open_ok, Key, Resp})
	catch T:R ->
	S = erlang:get_stacktrace(),
	erlang:exit({open_error, Key, {T, R, S}})
	end.


	update_lru(#st{key = Key, ts = Ts} = St) ->
	remove_from_lru(Ts, Key),
	NewTs = os:timestamp(),
	true = ets:insert(?LRU, {{NewTs, Key, self()}}),
	St#st{ts = NewTs}.


	update_cache(#st{val = undefined} = St, Val) ->
	true = ets:update_element(?CACHE, St#st.key, {#entry.val, Val}),
	?EVENT(inserted, St#st.key);

	update_cache(#st{val = V1} = _St, V2) when {open_ok, {ok, V2}} == V1 ->
	?EVENT(update_noop, _St#st.key);

	update_cache(St, Val) ->
	true = ets:update_element(?CACHE, St#st.key, {#entry.val, Val}),
	?EVENT(updated, {St#st.key, Val}).


	remove_from_cache(St) ->
	#st{
	key = Key,
	ts = Ts
	} = St,
	Pattern = #entry{key = Key, pid = self(), _ = '_'},
	CacheMSpec = [{Pattern, [], [true]}],
	1 = ets:select_delete(?CACHE, CacheMSpec),
	remove_from_lru(Ts, Key),
	?EVENT(removed, St#st.key),
	ok.


	remove_from_lru(Ts, Key) ->
	if Ts == undefined -> ok; true ->
	LruMSpec = [{{{Ts, Key, self()}}, [], [true]}],
	1 = ets:select_delete(?LRU, LruMSpec)
	end.


	drain_accessed() ->
	receive
	{'$gen_cast', accessed} ->
	drain_accessed()
	after 0 ->
	ok
	end.


	respond(Waiters, Resp) ->
	[gen_server:reply(W, Resp) \|\| W <- Waiters].