%% %% Copyright 2012 Dmitry Kolesnikov, All Rights Reserved %% %% 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. %% %% @description %% cache bucket process -module(cache_bucket). -behaviour(gen_server). -author('Dmitry Kolesnikov '). -include("cache.hrl"). -export([ start_link/1, start_link/2, init/1, terminate/2, handle_call/3, handle_cast/2, handle_info/2, code_change/3 ]). %% internal bucket state -record(cache, { name = undefined :: atom() %% name of cache bucket ,heap = undefined :: list() %% cache heap segments ,n = ?DEF_CACHE_N :: integer() %% number of segments ,policy = ?DEF_CACHE_POLICY :: integer() %% eviction policy ,check = ?DEF_CACHE_CHECK :: integer() %% status check timeout ,evict = undefined :: integer() %% evict timeout ,stats = undefined :: any() %% stats aggregation functor ,heir = undefined :: pid() %% the heir of evicted cache segment }). %%%---------------------------------------------------------------------------- %%% %%% Factory %%% %%%---------------------------------------------------------------------------- start_link(Opts) -> gen_server:start_link(?MODULE, [undefined, Opts], []). start_link({global, Name}, Opts) -> gen_server:start_link({global, Name}, ?MODULE, [Name, Opts], []); start_link(Name, Opts) -> gen_server:start_link({local, Name}, ?MODULE, [Name, Opts], []). init([Name, Opts]) -> {ok, init(Opts, Opts, #cache{name=Name})}. init([{policy, X} | Tail], Opts, State) -> init(Tail, Opts, State#cache{policy=X}); init([{n, X} | Tail], Opts, State) -> init(Tail, Opts, State#cache{n=X}); init([{check, X} | Tail], Opts, State) -> init(Tail, Opts, State#cache{check=X * 1000}); init([{stats, X} | Tail], Opts, State) -> init(Tail, Opts, State#cache{stats=X}); init([{heir, X} | Tail], Opts, State) -> init(Tail, Opts, State#cache{heir=X}); init([_ | Tail], Opts, State) -> init(Tail, Opts, State); init([], Opts, State) -> Type = proplists:get_value(type, Opts, ?DEF_CACHE_TYPE), TTL = proplists:get_value(ttl, Opts, ?DEF_CACHE_TTL), Size = proplists:get_value(size, Opts), Mem = proplists:get_value(memory, Opts), Evict= cache_util:mmul(cache_util:mdiv(TTL, State#cache.n), 1000), Heap = cache_heap:new( Type ,cache_util:mdiv(TTL, State#cache.n) ,cache_util:mdiv(Size, State#cache.n) ,cache_util:mdiv(cache_util:mdiv(Mem, State#cache.n), erlang:system_info(wordsize)) ), (catch erlang:send_after(State#cache.check, self(), check_heap)), (catch erlang:send_after(Evict, self(), evict_heap)), State#cache{heap=Heap, evict=Evict}. %% %% terminate(_Reason, State) -> cache_heap:purge(State#cache.heap, State#cache.heir), ok. %%%---------------------------------------------------------------------------- %%% %%% gen_server %%% %%%---------------------------------------------------------------------------- handle_call({put, Key, Val, TTL}, _, State) -> {reply, ok, cache_put(Key, Val, TTL, State)}; handle_call({get, Key}, _, State) -> {reply, cache_get(Key, State), State}; handle_call({lookup, Key}, _, State) -> {reply, cache_lookup(Key, State), State}; handle_call({has, Key}, _, State) -> {reply, cache_has(Key, State), State}; handle_call({ttl, Key}, _, State) -> {reply, cache_ttl(Key, State), State}; handle_call({remove, Key}, _, State) -> {reply, ok, cache_remove(Key, State)}; handle_call({apply, Key, Fun}, _, State0) -> {Result, State1} = cache_apply(Key, Fun, State0), {reply, Result, State1}; handle_call({acc, Key, Val}, _, State0) -> {Result, State1} = cache_acc(Key, Val, State0), {reply, Result, State1}; handle_call({add, Key, Val, TTL}, _, State0) -> {Result, State1} = cache_add(Key, Val, TTL, State0), {reply, Result, State1}; handle_call({replace, Key, Val, TTL}, _, State0) -> {Result, State1} = cache_replace(Key, Val, TTL, State0), {reply, Result, State1}; handle_call({prepend, Key, Val}, _, State0) -> {Result, State1} = cache_prepend(Key, Val, State0), {reply, Result, State1}; handle_call({append, Key, Val}, _, State0) -> {Result, State1} = cache_append(Key, Val, State0), {reply, Result, State1}; handle_call(i, _, State) -> Heap = cache_heap:refs(State#cache.heap), Refs = [X || {_, X} <- Heap], Expire = [X || {X, _} <- Heap], Size = [ets:info(X, size) || {_, X} <- Heap], Memory = [ets:info(X, memory) || {_, X} <- Heap], {reply, [{heap, Refs}, {expire, Expire}, {size, Size}, {memory, Memory}], State}; handle_call({heap, N}, _, State) -> try {_, Ref} = lists:nth(N, cache_heap:refs(State#cache.heap)), {reply, Ref, State} catch _:_ -> {reply, badarg, State} end; handle_call(drop, _, State) -> {stop, normal, ok, State}; handle_call(purge, _, State) -> {reply, ok, State#cache{heap=cache_heap:purge(State#cache.heap, State#cache.heir)}}; handle_call(_, _, State) -> {noreply, State}. %% %% handle_cast({put, Key, Val, TTL}, State) -> {noreply, cache_put(Key, Val, TTL, State)}; handle_cast({remove, Key}, State) -> {noreply, cache_remove(Key, State)}; handle_cast({acc, Key, Val}, State0) -> {_, State1} = cache_acc(Key, Val, State0), {noreply, State1}; handle_cast({add, Key, Val, TTL}, State0) -> {_, State1} = cache_add(Key, Val, TTL, State0), {noreply, State1}; handle_cast({replace, Key, Val, TTL}, State0) -> {_, State1} = cache_replace(Key, Val, TTL, State0), {noreply, State1}; handle_cast({prepend, Key, Val}, State0) -> {_, State1} = cache_prepend(Key, Val, State0), {noreply, State1}; handle_cast({append, Key, Val}, State0) -> {_, State1} = cache_append(Key, Val, State0), {noreply, State1}; handle_cast(_, State) -> {noreply, State}. %% %% handle_info(check_heap, #cache{n=N, check=Check}=State) -> erlang:send_after(Check, self(), check_heap), {Reason, Heap} = cache_heap:slip(State#cache.heap), case cache_heap:size(Heap) of X when X > N -> cache_util:stats(State#cache.stats, {cache, State#cache.name, Reason}), {noreply, State#cache{heap=cache_heap:drop(Heap, State#cache.heir)}}; _ -> {noreply, State#cache{heap=Heap}} end; handle_info(evict_heap, #cache{n=N, evict=Evict}=State) -> erlang:send_after(Evict, self(), evict_heap), {Reason, Heap} = cache_heap:slip(State#cache.heap), case cache_heap:size(Heap) of X when X > N -> cache_util:stats(State#cache.stats, {cache, State#cache.name, Reason}), {noreply, State#cache{heap=cache_heap:drop(Heap, State#cache.heir)}}; _ -> {noreply, State#cache{heap=Heap}} end; handle_info(_, S) -> {noreply, S}. %% %% code_change(_Vsn, S, _Extra) -> {ok, S}. %%%---------------------------------------------------------------------------- %%% %%% private %%% %%%---------------------------------------------------------------------------- %% %% insert value to cache cache_put(Key, Val, undefined, #cache{name=_Name, heap=Heap}=State) -> {_, Head} = cache_heap:head(Heap), true = ets:insert(Head, {Key, Val}), ok = heap_remove(Key, cache_heap:tail(Heap)), _ = stats(put, State), ?DEBUG("cache ~p: put ~p to heap ~p~n", [_Name, Key, Head]), State; cache_put(Key, Val, TTL, #cache{name=_Name, heap=Heap}=State) -> Expire = cache_util:now() + TTL, Refs = cache_heap:refs(Heap), case lists:splitwith(fun({X, _}) -> X > Expire end, Refs) of {[], _Tail} -> cache_put(Key, Val, undefined, State); {Head, Tail} -> [{_, Inst} | Rest] = lists:reverse(Head), true = ets:insert(Inst, {Key, Val}), ok = heap_remove(Key, Rest ++ Tail), _ = stats(put, State), ?DEBUG("cache ~p: put ~p to heap ~p~n", [_Name, Key, Inst]), State end. %% %% get cache value cache_get(Key, #cache{policy=mru}=State) -> % cache MRU should not move key anywhere because % cache always evicts last generation % fall-back to cache lookup cache_lookup(Key, State); cache_get(Key, #cache{name=_Name, heap=Heap}=State) -> {_, Head} = cache_heap:head(Heap), case heap_lookup(Key, cache_heap:refs(Heap)) of undefined -> stats(miss, State), undefined; {Head, Val} -> ?DEBUG("cache ~p: get ~p at cell ~p~n", [_Name, Key, Head]), stats(hit, State), Val; {Cell, Val} -> true = ets:insert(Head, {Key, Val}), _ = ets:delete(Cell, Key), ?DEBUG("cache ~p: get ~p at cell ~p~n", [_Name, Key, Cell]), stats(hit, State), Val end. %% %% lookup cache value cache_lookup(Key, #cache{name=_Name, heap=Heap}=State) -> case heap_lookup(Key, cache_heap:refs(Heap)) of undefined -> stats(miss, State), undefined; {_Cell, Val} -> ?DEBUG("cache ~p: get ~p at cell ~p~n", [_Name, Key, _Cell]), stats(hit, State), Val end. %% %% check if key exists cache_has(Key, #cache{name=_Name, heap=Heap}) -> case heap_has(Key, cache_heap:refs(Heap)) of false -> false; _Heap -> ?DEBUG("cache ~p: has ~p at cell ~p~n", [_Name, Key, _Heap]), true end. %% %% check key ttl cache_ttl(Key,#cache{heap=Heap}) -> case heap_has(Key, cache_heap:refs(Heap)) of false -> undefined; {Expire, _} -> Expire - cache_util:now() end. %% %% cache_remove(Key, #cache{name=_Name, heap=Heap}=State) -> ok = heap_remove(Key, cache_heap:refs(Heap)), _ = stats(remove, State), ?DEBUG("cache ~p: remove ~p~n", [_Name, Key]), State. %% %% cache_apply(Key, Fun, State) -> case Fun(cache_get(Key, State)) of undefined -> {undefined, State}; Val -> {Val, cache_put(Key, Val, undefined, State)} end. %% %% @todo: reduce one write cache_acc(Key, Val, State) when is_integer(Val) -> case cache_get(Key, State) of undefined -> {undefined, cache_put(Key, Val, undefined, State)}; X when is_integer(X) -> {X, cache_put(Key, X + Val, undefined, State)}; X when is_tuple(X) -> {erlang:element(1, X), cache_put(Key, tuple_acc({1, Val}, X), undefined, State)}; _ -> {badarg, State} end; cache_acc(Key, Val, State) -> case cache_get(Key, State) of X when is_tuple(X) -> {X, cache_put(Key, tuple_acc(Val, X), undefined, State)}; _ -> {badarg, State} end. tuple_acc({Pos, Val}, X) -> erlang:setelement(Pos, X, erlang:element(Pos, X) + Val); tuple_acc(List, X) -> lists:foldl( fun({Pos, Val}, Acc) -> erlang:setelement(Pos, Acc, erlang:element(Pos, Acc) + Val) end, X, List ). %% %% cache_add(Key, Val, TTL, State) -> case cache_has(Key, State) of true -> {{error, conflict}, State}; false -> {ok, cache_put(Key, Val, TTL, State)} end. %% %% cache_replace(Key, Val, TTL, State) -> case cache_has(Key, State) of true -> {ok, cache_put(Key, Val, TTL, State)}; false -> {{error, not_found}, State} end. %% %% cache_prepend(Key, Val, State) -> % @todo: reduce one write case cache_get(Key, State) of undefined -> {ok, cache_put(Key, [Val], undefined, State)}; X when is_list(X) -> {ok, cache_put(Key, [Val|X], undefined, State)}; X -> {ok, cache_put(Key, [Val,X], undefined, State)} end. %% %% cache_append(Key, Val, State) -> % @todo: reduce one write case cache_get(Key, State) of undefined -> {ok, cache_put(Key, [Val], undefined, State)}; X when is_list(X) -> {ok, cache_put(Key, X++[Val], undefined, State)}; X -> {ok, cache_put(Key, [X, Val], undefined, State)} end. %% %% remove key from heap segments heap_remove(Key, Heap) -> lists:foreach( fun({_, Id}) -> ets:delete(Id, Key) end, Heap ). %% %% heap_lookup(Key, [{_, Heap} | Tail]) -> case ets:lookup(Heap, Key) of [] -> heap_lookup(Key, Tail); [{_, Val}] -> {Heap, Val} end; heap_lookup(_Key, []) -> undefined. %% %% heap_has(Key, [{_, Heap}=X | Tail]) -> case ets:member(Heap, Key) of false -> heap_has(Key, Tail); true -> X end; heap_has(_Key, []) -> false. %% %% update statistic stats(_, #cache{stats = undefined}) -> ok; stats(X, #cache{stats = Stats, name = Name}) -> cache_util:stats(Stats, {cache, Name, X}).