Clean-up open bitcask_file instances when owner exits

src/bitcask_file.erl

@@ -8,15 +8,17 @@
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
--record(state, {fd}).
+-record(state, {fd,
+                owner}).
 
 %%%===================================================================
 %%% API
 %%%===================================================================
 
 file_open(Filename, Opts) ->
     {ok, Pid} = gen_server:start(?MODULE, [], []),
-    case gen_server:call(Pid, {file_open, Filename, Opts}, infinity) of
+    Owner = self(),
+    case gen_server:call(Pid, {file_open, Owner, Filename, Opts}, infinity) of
         ok ->
             {ok, Pid};
         Error ->
@@ -77,7 +79,8 @@ check_pid(Pid) ->
 init([]) ->
     {ok, #state{}}.
 
-handle_call({file_open, Filename, Opts}, _From, State) ->
+handle_call({file_open, Owner, Filename, Opts}, _From, State) ->
+    monitor(process, Owner),
     IsCreate = proplists:get_bool(create, Opts),
     IsReadOnly = proplists:get_bool(readonly, Opts),
     Mode = case {IsReadOnly, IsCreate} of
@@ -92,30 +95,37 @@ handle_call({file_open, Filename, Opts}, _From, State) ->
     %%                                               proplists:get_bool(Opt, Opts)],
     case file:open(Filename, Mode) of
         {ok, Fd} ->
-            State2 = State#state{fd=Fd},
+            State2 = State#state{fd=Fd, owner=Owner},
             {reply, ok, State2};
         Error ->
             {reply, Error, State}
     end;
-handle_call(file_close, _From, State=#state{fd=Fd}) ->
+handle_call(file_close, From, State=#state{fd=Fd}) -> 
+    check_owner(From, State),
     ok = file:close(Fd),
     {stop, normal, ok, State};
-handle_call(file_sync, _From, State=#state{fd=Fd}) ->
+handle_call(file_sync, From, State=#state{fd=Fd}) ->
+    check_owner(From, State),
     Reply = file:sync(Fd),
     {reply, Reply, State};
-handle_call({file_pread, Offset, Size}, _From, State=#state{fd=Fd}) ->
+handle_call({file_pread, Offset, Size}, From, State=#state{fd=Fd}) ->
+    check_owner(From, State),
     Reply = file:pread(Fd, Offset, Size),
     {reply, Reply, State};
-handle_call({file_pwrite, Offset, Bytes}, _From, State=#state{fd=Fd}) ->
+handle_call({file_pwrite, Offset, Bytes}, From, State=#state{fd=Fd}) ->
+    check_owner(From, State),
     Reply = file:pwrite(Fd, Offset, Bytes),
     {reply, Reply, State};
-handle_call({file_read, Size}, _From, State=#state{fd=Fd}) ->
+handle_call({file_read, Size}, From, State=#state{fd=Fd}) ->
+    check_owner(From, State),
     Reply = file:read(Fd, Size),
     {reply, Reply, State};
-handle_call({file_write, Bytes}, _From, State=#state{fd=Fd}) ->
+handle_call({file_write, Bytes}, From, State=#state{fd=Fd}) ->
+    check_owner(From, State),
     Reply = file:write(Fd, Bytes),
     {reply, Reply, State};
-handle_call(file_seekbof, _From, State=#state{fd=Fd}) ->
+handle_call(file_seekbof, From, State=#state{fd=Fd}) ->
+    check_owner(From, State),
     {ok, _} = file:position(Fd, bof),
     {reply, ok, State};
 
@@ -126,6 +136,11 @@ handle_call(_Request, _From, State) ->
 handle_cast(_Msg, State) ->
     {noreply, State}.
 
+handle_info({'DOWN', _Ref, _, _Pid, _Status}, State=#state{fd=Fd}) ->
+    %% Owner has stopped, close file and shutdown
+    %% ?debugFmt("Cleaning up: ~p/~p~n", [self(), Fd]),
+    ok = file:close(Fd),
+    {stop, normal, State};
 handle_info(_Info, State) ->
     {noreply, State}.
 
@@ -138,3 +153,13 @@ code_change(_OldVsn, State, _Extra) ->
 %%%===================================================================
 %%% Internal functions
 %%%===================================================================
+
+check_owner({Pid, _Mref}, #state{owner=Owner}) ->
+    case Pid == Owner of
+        true ->
+            ok;
+        false ->
+            %% ?debugFmt("NON-OWNER ACCESSING FILE!!! Owner=~p, Other=~p~n", [Owner, Pid]),
+            throw(owner_invariant_failed),
+            ok
+    end.