umgeher/leveled
0
Fork 0
Code Issues Pull requests Releases Activity Actions

Add put timing points

Browse source
This commit is contained in:
martinsumner 2016-12-20 23:11:50 +00:00
parent 299e8e6de3
commit 060ce2e263
5 changed files with 74 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

25
src/leveled_bookie.erl
View file

@ -150,6 +150,7 @@
-define(CHECKJOURNAL_PROB, 0.2).
-define(CACHE_SIZE_JITTER, 25).
-define(JOURNAL_SIZE_JITTER, 20).
-define(PUT_TIMING_LOGPOINT, 10000).
-record(ledger_cache, {skiplist = leveled_skiplist:empty(true) :: tuple(),
min_sqn = infinity :: integer()|infinity,
@ -160,7 +161,8 @@
cache_size :: integer(),
ledger_cache = #ledger_cache{},
is_snapshot :: boolean(),
slow_offer = false :: boolean()}).
slow_offer = false :: boolean(),
put_timing = {0, {0, 0}, {0, 0}} :: tuple()}).
%%%============================================================================
@ -262,10 +264,12 @@ init([Opts]) ->
handle_call({put, Bucket, Key, Object, IndexSpecs, Tag, TTL}, From, State) ->
LedgerKey = leveled_codec:to_ledgerkey(Bucket, Key, Tag),
SW = os:timestamp(),
{ok, SQN, ObjSize} = leveled_inker:ink_put(State#state.inker,
LedgerKey,
Object,
{IndexSpecs, TTL}),
T0 = timer:now_diff(os:timestamp(), SW),
Changes = preparefor_ledgercache(no_type_assigned,
LedgerKey,
SQN,
@ -273,6 +277,8 @@ handle_call({put, Bucket, Key, Object, IndexSpecs, Tag, TTL}, From, State) ->
ObjSize,
{IndexSpecs, TTL}),
Cache0 = addto_ledgercache(Changes, State#state.ledger_cache),
T1 = timer:now_diff(os:timestamp(), SW) - T0,
PutTimings = update_put_timings(State#state.put_timing, T0, T1),
% If the previous push to memory was returned then punish this PUT with a
% delay. If the back-pressure in the Penciller continues, these delays
% will beocme more frequent
@ -286,9 +292,13 @@ handle_call({put, Bucket, Key, Object, IndexSpecs, Tag, TTL}, From, State) ->
Cache0,
State#state.penciller) of
{ok, NewCache} ->
{noreply, State#state{ledger_cache=NewCache, slow_offer=false}};
{noreply, State#state{ledger_cache=NewCache,
put_timing=PutTimings,
slow_offer=false}};
{returned, NewCache} ->
{noreply, State#state{ledger_cache=NewCache, slow_offer=true}}
{noreply, State#state{ledger_cache=NewCache,
put_timing=PutTimings,
slow_offer=true}}
end;
handle_call({get, Bucket, Key, Tag}, _From, State) ->
LedgerKey = leveled_codec:to_ledgerkey(Bucket, Key, Tag),
@ -454,6 +464,15 @@ push_ledgercache(Penciller, Cache) ->
%%% Internal functions
%%%============================================================================
update_put_timings({?PUT_TIMING_LOGPOINT, {Total0, Total1}, {Max0, Max1}},
T0, T1) ->
leveled_log:log("B0012",
[?PUT_TIMING_LOGPOINT, Total0, Total1, Max0, Max1]),
{1, {T0, T1}, {T0, T1}};
update_put_timings({N, {Total0, Total1}, {Max0, Max1}}, T0, T1) ->
{N + 1, {Total0 + T0, Total1 + T1}, {max(Max0, T0), max(Max1, T1)}}.
cache_size(LedgerCache) ->
leveled_skiplist:size(LedgerCache#ledger_cache.skiplist).