Merge pull request #1331 from basho/bugfix/dr/memory_backend_ets_leaks

Fix memory leak in TimeRef ETS table in memory_backend [RIAK-1576]

Reviewed-by: macintux

src/riak_kv_memory_backend.erl

@@ -170,6 +170,7 @@ stop(#state{data_ref=DataRef,
                  {error, term(), state()}.
 get(Bucket, Key, State=#state{data_ref=DataRef,
                               index_ref=IndexRef,
+                              time_ref = TimeRef,
                               used_memory=UsedMemory,
                               max_memory=MaxMemory,
                               ttl=TTL}) ->
@@ -183,6 +184,8 @@ get(Bucket, Key, State=#state{data_ref=DataRef,
                     %% entries blindly using match_delete.
                     ets:delete(DataRef, {Bucket, Key}),
                     ets:match_delete(IndexRef, ?DELETE_PTN(Bucket, Key)),
+                    %% prevents leak in the TimeRef table on expire
+                    delete_time_entry(TimeRef, Timestamp),
                     case MaxMemory of
                         undefined ->
                             UsedMemory1 = UsedMemory;
@@ -222,6 +225,10 @@ put(Bucket, PrimaryKey, IndexSpecs, Val, State=#state{data_ref=DataRef,
         case ets:lookup(DataRef, {Bucket, PrimaryKey}) of
             [] ->
                 0;
+            [{_BKey, {{ts, OldTimestamp}, _OldVal}}=OldObject] ->
+                %% prevents leak in the TimeRef table on update
+                delete_time_entry(TimeRef, OldTimestamp),
+                object_size(OldObject);
             [OldObject] ->
                 object_size(OldObject)
         end,
@@ -653,6 +660,13 @@ update_indexes(Bucket, Key, [{add, Field, Value}|Rest], IndexRef) ->
 time_entry(Bucket, Key, Now, TimeRef) ->
     ets:insert(TimeRef, {Now, {Bucket, Key}}).
 
+%% @private
+delete_time_entry(TimeRef, Timestamp) ->
+    case TimeRef of
+      undefined -> ok;
+      _ -> ets:delete(TimeRef, Timestamp)
+    end.
+
 %% @private
 %% @doc Dump some entries if the max memory size has
 %% been breached.
@@ -788,6 +802,43 @@ regression_367_key_range_test_() ->
      ?_assertEqual(ok, stop(State1))
     ].
 
+ttl_ets_timeref_leak_put_path_test() ->
+    Config = [{ttl, 5}, {max_memory, 1024*10}], %% Need max_memory to get Timer, thus TimeRef
+    {ok, State} = start(142, Config),
+
+    Bucket = <<"Bucket">>,
+    Key = <<"Key">>,
+    Value = <<"Value">>,
+
+
+    %% Put an object, but ignore its put_obj_size
+    {ok, State1} = put(Bucket, Key, [], Value, State),
+    {ok, #state{time_ref=TimeRef} = State2} = put(Bucket, Key, [], Value, State1),
+    ?assertEqual(1, get_time_ref_count(TimeRef)),
+    {ok, _State3} = put(Bucket, Key, [], Value, State2),
+    ?assertEqual(1, get_time_ref_count(TimeRef)).
+
+ttl_ets_timeref_leak_get_after_expiry_test() ->
+    Config = [{ttl, 1}, {max_memory, 1024*10}], %% Need max_memory to get Timer, thus TimeRef
+    {ok, State} = start(142, Config),
+
+    Bucket = <<"Bucket">>,
+    Key = <<"Key">>,
+    Value = <<"Value">>,
+
+
+    %% Put an object, but ignore its put_obj_size
+    {ok, State1} = put(Bucket, Key, [], Value, State),
+    {ok, #state{time_ref=TimeRef} = State2} = put(Bucket, Key, [], Value, State1),
+    ?assertEqual(1, get_time_ref_count(TimeRef)),
+    timer:sleep(timer:seconds(1)),
+    {error, not_found, _State3} = get(Bucket, Key, State2),
+    ?assertEqual(0, get_time_ref_count(TimeRef)).
+
+get_time_ref_count(TimeRef) ->
+    ets:info(TimeRef, size).
+
+
 -ifdef(EQC).
 
 eqc_test_() ->