nickva commented on code in PR #4266:
URL: https://github.com/apache/couchdb/pull/4266#discussion_r1026746244
##########
src/smoosh/src/smoosh_channel.erl:
##########
@@ -235,245 +190,146 @@ handle_info(check_window, State) ->
State;
{false, true} ->
% resume is always safe even if we did not previously suspend
- {reply, ok, NewState} = handle_call(resume_and_activate, nil,
State),
- NewState;
+ do_resume(State);
+ {true, false} when StrictWindow =:= "true" ->
+ % suspend immediately
+ do_suspend(State);
{true, false} ->
- if
- StrictWindow =:= "true" ->
- {reply, ok, NewState} = handle_call(suspend, nil,
State),
- NewState;
- true ->
- State#state{paused = true}
- end
+ % prevent new jobs from starting, active ones keep running
+ State#state{paused = true}
end,
- erlang:send_after(60 * 1000, self(), check_window),
+ schedule_check_window(),
{noreply, FinalState};
-handle_info(start_recovery, #state{name = Name, waiting = Waiting0} = State0)
->
- RecActive = recover(active_file_name(Name)),
- Waiting1 = lists:foldl(
- fun(DbName, Acc) ->
- case couch_db:exists(DbName) andalso
couch_db:is_compacting(DbName) of
- true ->
- Priority = smoosh_server:get_priority(Name, DbName),
- smoosh_priority_queue:in(DbName, Priority, Priority, Acc);
- false ->
- Acc
- end
- end,
- Waiting0,
- RecActive
- ),
- State1 = maybe_start_compaction(State0#state{paused = false, waiting =
Waiting1}),
- Level = smoosh_utils:log_level("compaction_log_level", "debug"),
- couch_log:Level(
- "~p Previously active compaction jobs (if any) have been successfully
recovered and restarted.",
- [?MODULE]
- ),
- erlang:send_after(?ACTIVATE_DELAY_IN_MSEC, self(), activate),
- {noreply, State1#state{paused = true}};
-handle_info(activate, State) ->
- {noreply, activate_channel(State)};
-handle_info(persist, State) ->
- ok = persist_and_reschedule(State),
+handle_info(update_status, #state{} = State) ->
+ schedule_update_status(),
+ {noreply, set_status(State)};
+handle_info(checkpoint, #state{cref = Ref} = State) when is_reference(Ref) ->
+ % If a checkpointer process is still running, don't start another one.
+ schedule_checkpoint(),
{noreply, State};
-handle_info(pause, State) ->
- {noreply, State#state{paused = true}};
+handle_info(checkpoint, #state{cref = undefined} = State) ->
+ % Start an asyncronous checkpoint process so we don't block the channel
+ #state{waiting = Waiting, active = Active, starting = Starting} = State,
+ Args = [Waiting, Active, Starting],
+ {_, Ref} = spawn_monitor(smoosh_persist, persist, Args),
+ schedule_checkpoint(),
+ {noreply, State#state{cref = Ref}};
handle_info(unpause, State) ->
{noreply, maybe_start_compaction(State#state{paused = false})}.
-terminate(_Reason, _State) ->
- ok.
-
% private functions.
-recover(FilePath) ->
- case do_recover(FilePath) of
- {ok, List} ->
- List;
- error ->
- []
- end.
-
-do_recover(FilePath) ->
- case file:read_file(FilePath) of
- {ok, Content} ->
- <<Vsn, Binary/binary>> = Content,
- try parse_state(Vsn, ?VSN, Binary) of
- Term ->
- Level = smoosh_utils:log_level("compaction_log_level",
"debug"),
- couch_log:Level(
- "~p Successfully restored state file ~s", [?MODULE,
FilePath]
- ),
- {ok, Term}
- catch
- error:Reason ->
- couch_log:error(
- "~p Invalid state file (~p). Deleting ~s", [?MODULE,
Reason, FilePath]
- ),
- file:delete(FilePath),
- error
- end;
- {error, enoent} ->
- Level = smoosh_utils:log_level("compaction_log_level", "debug"),
- couch_log:Level(
- "~p (~p) State file ~s does not exist. Not restoring.",
[?MODULE, enoent, FilePath]
- ),
- error;
- {error, Reason} ->
- couch_log:error(
- "~p Cannot read the state file (~p). Deleting ~s", [?MODULE,
Reason, FilePath]
- ),
- file:delete(FilePath),
- error
- end.
-
-parse_state(1, ?VSN, Binary) ->
- erlang:binary_to_term(Binary, [safe]);
-parse_state(Vsn, ?VSN, _) ->
- error({unsupported_version, Vsn}).
-
-persist_and_reschedule(State) ->
- persist_queue(State),
- erlang:send_after(?CHECKPOINT_INTERVAL_IN_MSEC, self(), persist),
- ok.
-
-persist_queue(#state{name = Name, active = Active, starting = Starting,
waiting = Waiting}) ->
- Active1 = lists:foldl(
- fun({DbName, _}, Acc) ->
- [DbName | Acc]
- end,
- [],
- Active
- ),
- Starting1 = lists:foldl(
- fun({_, DbName}, Acc) ->
- [DbName | Acc]
- end,
- [],
- Starting
- ),
- smoosh_utils:write_to_file(Active1, active_file_name(Name), ?VSN),
- smoosh_utils:write_to_file(Starting1, starting_file_name(Name), ?VSN),
- smoosh_priority_queue:write_to_file(Waiting).
-
-active_file_name(Name) ->
- filename:join(config:get("smoosh", "state_dir", "."), Name ++ ".active").
-
-starting_file_name(Name) ->
- filename:join(config:get("smoosh", "state_dir", "."), Name ++ ".starting").
+unpersist(Name) ->
+ % Insert into the access table with a current
+ % timestamp to prevent the same dbs from being re-enqueued
+ % again after startup startup
Review Comment:
Definitely. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
