This is an automated email from the ASF dual-hosted git repository. rnewson pushed a commit to branch out-of-disk-handler in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit e8a2829f56af21bb2e7880a31f6330671dce0343 Author: Robert Newson <[email protected]> AuthorDate: Fri Jun 16 09:25:41 2023 +0100 Introduce optional countermeasures as we run out of disk space --- rel/overlay/etc/vm.args | 9 ++ rel/reltool.config | 2 + src/chttpd/src/chttpd.erl | 2 + src/couch/priv/stats_descriptions.cfg | 4 + src/couch/src/couch.app.src | 1 + .../src/{couch_app.erl => couch_alarm_handler.erl} | 42 ++--- src/couch/src/couch_app.erl | 2 + src/couch/src/couch_disk_monitor.erl | 171 +++++++++++++++++++++ src/couch/src/couch_secondary_sup.erl | 3 +- src/couch_mrview/src/couch_mrview.erl | 8 +- src/couch_mrview/src/couch_mrview_util.erl | 17 +- src/fabric/src/fabric_doc_update.erl | 19 +-- src/fabric/src/fabric_rpc.erl | 44 ++++-- src/fabric/src/fabric_streams.erl | 4 +- src/fabric/src/fabric_view.erl | 9 +- src/fabric/src/fabric_view_map.erl | 6 + src/fabric/src/fabric_view_reduce.erl | 4 + src/ken/src/ken_server.erl | 8 +- 18 files changed, 298 insertions(+), 57 deletions(-) diff --git a/rel/overlay/etc/vm.args b/rel/overlay/etc/vm.args index 886bbb903..8b2440fc8 100644 --- a/rel/overlay/etc/vm.args +++ b/rel/overlay/etc/vm.args @@ -122,3 +122,12 @@ # in the features list. # #-crypto fips_mode true + +# OS Mon Settings + +# only start disksup +-os_mon start_cpu_sup false +-os_mon start_memsup false + +# Check disk space every 5 minutes +-os_mon disk_space_check_interval 5 diff --git a/rel/reltool.config b/rel/reltool.config index d84ef597c..ebb15bb83 100644 --- a/rel/reltool.config +++ b/rel/reltool.config @@ -23,6 +23,7 @@ sasl, ssl, stdlib, + os_mon, syntax_tools, xmerl, %% couchdb @@ -86,6 +87,7 @@ {app, sasl, [{incl_cond, include}]}, {app, ssl, [{incl_cond, include}]}, {app, stdlib, [{incl_cond, include}]}, + {app, os_mon, [{incl_cond, include}]}, {app, syntax_tools, [{incl_cond, include}]}, {app, xmerl, [{incl_cond, include}]}, diff --git a/src/chttpd/src/chttpd.erl b/src/chttpd/src/chttpd.erl index 53abc731f..c8e6fdc97 100644 --- a/src/chttpd/src/chttpd.erl +++ b/src/chttpd/src/chttpd.erl @@ -1138,6 +1138,8 @@ error_info(timeout) -> >>}; error_info({service_unavailable, Reason}) -> {503, <<"service unavailable">>, Reason}; +error_info({insufficient_storage, Reason}) -> + {507, <<"insufficent_storage">>, Reason}; error_info({timeout, _Reason}) -> error_info(timeout); error_info({'EXIT', {Error, _Stack}}) -> diff --git a/src/couch/priv/stats_descriptions.cfg b/src/couch/priv/stats_descriptions.cfg index 6c0d4dad2..1983eed9b 100644 --- a/src/couch/priv/stats_descriptions.cfg +++ b/src/couch/priv/stats_descriptions.cfg @@ -266,6 +266,10 @@ {type, counter}, {desc, <<"number of HTTP 503 Service unavailable responses">>} ]}. +{[couchdb, httpd_status_codes, 507], [ + {type, counter}, + {desc, <<"number of HTTP 507 Insufficient Storage responses">>} +]}. {[couchdb, open_databases], [ {type, counter}, {desc, <<"number of open databases">>} diff --git a/src/couch/src/couch.app.src b/src/couch/src/couch.app.src index ef4e5e956..6fc293ac1 100644 --- a/src/couch/src/couch.app.src +++ b/src/couch/src/couch.app.src @@ -33,6 +33,7 @@ sasl, inets, ssl, + os_mon, % Upstream deps ibrowse, diff --git a/src/couch/src/couch_app.erl b/src/couch/src/couch_alarm_handler.erl similarity index 50% copy from src/couch/src/couch_app.erl copy to src/couch/src/couch_alarm_handler.erl index 8cd8c8482..71f427128 100644 --- a/src/couch/src/couch_app.erl +++ b/src/couch/src/couch_alarm_handler.erl @@ -10,29 +10,31 @@ % License for the specific language governing permissions and limitations under % the License. --module(couch_app). - --behaviour(application). +-module(couch_alarm_handler). +-behaviour(gen_event). +% gen_event callbacks -export([ - start/2, - stop/1, - uptime/0 + init/1, + handle_event/2, + handle_call/2, + handle_info/2 ]). -start(_Type, _) -> - case couch_sup:start_link() of - {ok, _} = Resp -> - {Time, _} = statistics(wall_clock), - application:set_env(couch, start_time, Time), - Resp; - Else -> - throw(Else) - end. +init(_Args) -> + {ok, nil}. + +handle_event({set_alarm, Alarm}, St) -> + error_logger:info_report([{alarm_handler, {set, Alarm}}]), + {ok, St}; +handle_event({clear_alarm, AlarmId}, St) -> + error_logger:info_report([{alarm_handler, {clear, AlarmId}}]), + {ok, St}; +handle_event(_Event, St) -> + {ok, St}. -stop(_) -> - ok. +handle_call(_Query, St) -> + {ok, {error, bad_query}, St}. -uptime() -> - {Time, _} = statistics(wall_clock), - Time - application:get_env(couch, start_time, Time). +handle_info(_Msg, St) -> + {ok, St}. diff --git a/src/couch/src/couch_app.erl b/src/couch/src/couch_app.erl index 8cd8c8482..46476d634 100644 --- a/src/couch/src/couch_app.erl +++ b/src/couch/src/couch_app.erl @@ -21,6 +21,8 @@ ]). start(_Type, _) -> + %% register our alarm handler + gen_event:swap_handler(alarm_handler, {alarm_handler, swap}, {couch_alarm_handler, ok}), case couch_sup:start_link() of {ok, _} = Resp -> {Time, _} = statistics(wall_clock), diff --git a/src/couch/src/couch_disk_monitor.erl b/src/couch/src/couch_disk_monitor.erl new file mode 100644 index 000000000..86824ed7c --- /dev/null +++ b/src/couch/src/couch_disk_monitor.erl @@ -0,0 +1,171 @@ +% 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. + +-module(couch_disk_monitor). +-behaviour(gen_server). + +% public api +-export([ + block_background_view_indexing/0, + block_interactive_view_indexing/0, + block_database_writes/0 +]). + +% testing +-export([ + set_database_dir_percent_used/1, + set_view_index_dir_percent_used/1 +]). + +% gen_server callbacks +-export([ + start_link/0, + init/1, + handle_call/3, + handle_cast/2, + handle_info/2 +]). + +-include_lib("kernel/include/file.hrl"). + +-define(SECTION, "disk_monitor"). + +-record(st, { + timer +}). + +block_background_view_indexing() -> + Enabled = enabled(), + if + Enabled -> + view_index_dir_percent_used() > background_view_indexing_threshold(); + true -> + false + end. + +block_interactive_view_indexing() -> + Enabled = enabled(), + if + Enabled -> view_index_dir_percent_used() > interactive_view_indexing_threshold(); + true -> false + end. + +block_database_writes() -> + Enabled = enabled(), + if + Enabled -> database_dir_percent_used() > database_writes_threshold(); + true -> false + end. + +set_database_dir_percent_used(PercentUsed) when PercentUsed >= 0, PercentUsed =< 100 -> + gen_server:cast(?MODULE, {set_database_dir_percent_used, PercentUsed}). + +set_view_index_dir_percent_used(PercentUsed) when PercentUsed >= 0, PercentUsed =< 100 -> + gen_server:cast(?MODULE, {set_view_index_dir_percent_used, PercentUsed}). + +start_link() -> + gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). + +init(_Args) -> + ets:new(?MODULE, [named_table]), + update_disk_data(), + {ok, TRef} = timer:send_after(timer_interval(), update_disk_data), + {ok, #st{timer = TRef}}. + +handle_call(_Msg, _From, St) -> + {reply, {error, unknown_msg}, St}. + +handle_cast({set_database_dir_percent_used, PercentUsed}, St) -> + ets:insert(?MODULE, {database_dir, PercentUsed}), + {noreply, St}; +handle_cast({set_view_index_dir_percent_used, PercentUsed}, St) -> + ets:insert(?MODULE, {view_index_dir, PercentUsed}), + {noreply, St}; +handle_cast(_Msg, St) -> + {noreply, St}. + +handle_info(update_disk_data, St) -> + timer:cancel(St#st.timer), + update_disk_data(), + {ok, TRef} = timer:send_after(timer_interval(), update_disk_data), + {noreply, St#st{timer = TRef}}; +handle_info(_Msg, St) -> + {noreply, St}. + +update_disk_data() -> + DiskData = disksup:get_disk_data(), + update_disk_data(DiskData). + +update_disk_data([]) -> + ok; +update_disk_data([{Id, _, PercentUsed} | Rest]) -> + IsDatabaseDir = is_database_dir(Id), + IsViewIndexDir = is_view_index_dir(Id), + if + IsDatabaseDir -> + ets:insert(?MODULE, {database_dir, PercentUsed}); + true -> + ok + end, + if + IsViewIndexDir -> + ets:insert(?MODULE, {view_index_dir, PercentUsed}); + true -> + ok + end, + update_disk_data(Rest). + +is_database_dir(MntOn) -> + same_device(config:get("couchdb", "database_dir"), MntOn). + +is_view_index_dir(MntOn) -> + same_device(config:get("couchdb", "view_index_dir"), MntOn). + +same_device(DirA, DirB) -> + case {device_id(DirA), device_id(DirB)} of + {{ok, DeviceId}, {ok, DeviceId}} -> + true; + _Else -> + false + end. + +device_id(Dir) -> + case file:read_file_info(Dir) of + {ok, FileInfo} -> + {ok, {FileInfo#file_info.minor_device, FileInfo#file_info.major_device}}; + {error, Reason} -> + {error, Reason} + end. + +database_dir_percent_used() -> + [{database_dir, PercentUsed}] = ets:lookup(?MODULE, database_dir), + PercentUsed. + +view_index_dir_percent_used() -> + [{view_index_dir, PercentUsed}] = ets:lookup(?MODULE, view_index_dir), + PercentUsed. + +background_view_indexing_threshold() -> + config:get_integer(?SECTION, "background_view_indexing_threshold", 80). + +interactive_view_indexing_threshold() -> + config:get_integer(?SECTION, "interactive_view_indexing_threshold", 90). + +database_writes_threshold() -> + config:get_integer(?SECTION, "database_writes_threshold", 90). + +enabled() -> + config:get_boolean(?SECTION, "enable", false). + +%% Align our refresh with the os_mon refresh +timer_interval() -> + application:get_env(os_mon, disk_space_check_interval, 30) * 60_000. diff --git a/src/couch/src/couch_secondary_sup.erl b/src/couch/src/couch_secondary_sup.erl index cfe38bbd4..8fc3c9a13 100644 --- a/src/couch/src/couch_secondary_sup.erl +++ b/src/couch/src/couch_secondary_sup.erl @@ -26,7 +26,8 @@ init([]) -> [ {query_servers, {couch_proc_manager, start_link, []}}, {vhosts, {couch_httpd_vhost, start_link, []}}, - {uuids, {couch_uuids, start, []}} + {uuids, {couch_uuids, start, []}}, + {disk_manager, {couch_disk_monitor, start_link, []}} ] ++ couch_index_servers(), MaybeHttp = diff --git a/src/couch_mrview/src/couch_mrview.erl b/src/couch_mrview/src/couch_mrview.erl index 1a4a3ebcc..a50fcd670 100644 --- a/src/couch_mrview/src/couch_mrview.erl +++ b/src/couch_mrview/src/couch_mrview.erl @@ -291,8 +291,8 @@ query_view(Db, DDoc, VName, Args0, Callback, Acc0) -> _ -> {ok, Acc0} end, query_view(Db, VInfo, Args, Callback, Acc1); - ddoc_updated -> - Callback(ok, ddoc_updated) + Error when Error == ddoc_updated; Error == insufficient_storage -> + Callback(ok, Error) end. get_view_index_pid(Db, DDoc, ViewName, Args0) -> @@ -752,8 +752,8 @@ default_cb({final, Info}, []) -> {ok, [Info]}; default_cb({final, _}, Acc) -> {ok, Acc}; -default_cb(ok, ddoc_updated) -> - {ok, ddoc_updated}; +default_cb(ok, Error) when Error == ddoc_updated; Error == insufficient_storage -> + {ok, Error}; default_cb(Row, Acc) -> {ok, [Row | Acc]}. diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl index 5913aa3d0..a478685da 100644 --- a/src/couch_mrview/src/couch_mrview_util.erl +++ b/src/couch_mrview/src/couch_mrview_util.erl @@ -158,8 +158,8 @@ get_view(Db, DDoc, ViewName, Args0) -> check_range(Args3, view_cmp(View)), Sig = view_sig(Db, State, View, Args3), {ok, {Type, View, Ref}, Sig, Args3}; - ddoc_updated -> - ddoc_updated + Error when Error == ddoc_updated; Error == insufficient_storage -> + Error end. get_view_index_pid(Db, DDoc, ViewName, Args0) -> @@ -176,6 +176,7 @@ get_view_index_state(_, DDoc, _, _, RetryCount) when RetryCount < 0 -> couch_log:warning("DDoc '~s' recreated too frequently", [DDoc#doc.id]), throw({get_view_state, exceeded_retry_count}); get_view_index_state(Db, DDoc, ViewName, Args0, RetryCount) -> + BlockInteractiveIndexing = couch_disk_monitor:block_interactive_view_indexing(), try {ok, Pid, Args} = get_view_index_pid(Db, DDoc, ViewName, Args0), UpdateSeq = couch_util:with_db(Db, fun(WDb) -> @@ -190,12 +191,24 @@ get_view_index_state(Db, DDoc, ViewName, Args0, RetryCount) -> couch_index:get_state(Pid, 0); false -> couch_index:get_state(Pid, 0); + _ when BlockInteractiveIndexing -> + %% if the view is already fresh enough, proceed. + %% otherwise it's an error until there's room to update. + case couch_index:get_state(Pid, 0) of + {ok, #mrst{update_seq = MaybeSeq} = St} when MaybeSeq >= UpdateSeq -> + {ok, St}; + {ok, #mrst{}} -> + insufficient_storage; + Else0 -> + Else0 + end; _ -> couch_index:get_state(Pid, UpdateSeq) end, case State of {ok, State0} -> {ok, State0, Args}; ddoc_updated -> ddoc_updated; + insufficient_storage -> insufficient_storage; Else -> throw(Else) end catch diff --git a/src/fabric/src/fabric_doc_update.erl b/src/fabric/src/fabric_doc_update.erl index 77b424911..44448c742 100644 --- a/src/fabric/src/fabric_doc_update.erl +++ b/src/fabric/src/fabric_doc_update.erl @@ -63,19 +63,15 @@ handle_message({rexi_DOWN, _, {_, NodeRef}, _}, _Worker, Acc0) -> NewGrpDocs = [X || {#shard{node = N}, _} = X <- GroupedDocs, N =/= NodeRef], skip_message({length(NewGrpDocs), LenDocs, W, NewGrpDocs, DocReplyDict}); handle_message({rexi_EXIT, _}, Worker, Acc0) -> - {WC, LenDocs, W, GrpDocs, DocReplyDict} = Acc0, - NewGrpDocs = lists:keydelete(Worker, 1, GrpDocs), - skip_message({WC - 1, LenDocs, W, NewGrpDocs, DocReplyDict}); + ignore_failed_worker(Worker, Acc0); handle_message({error, all_dbs_active}, Worker, Acc0) -> % treat it like rexi_EXIT, the hope at least one copy will return successfully - {WC, LenDocs, W, GrpDocs, DocReplyDict} = Acc0, - NewGrpDocs = lists:keydelete(Worker, 1, GrpDocs), - skip_message({WC - 1, LenDocs, W, NewGrpDocs, DocReplyDict}); + ignore_failed_worker(Worker, Acc0); handle_message(internal_server_error, Worker, Acc0) -> % happens when we fail to load validation functions in an RPC worker - {WC, LenDocs, W, GrpDocs, DocReplyDict} = Acc0, - NewGrpDocs = lists:keydelete(Worker, 1, GrpDocs), - skip_message({WC - 1, LenDocs, W, NewGrpDocs, DocReplyDict}); + ignore_failed_worker(Worker, Acc0); +handle_message({insufficient_storage, _Msg}, Worker, Acc0) -> + ignore_failed_worker(Worker, Acc0); handle_message(attachment_chunk_received, _Worker, Acc0) -> {ok, Acc0}; handle_message({ok, Replies}, Worker, Acc0) -> @@ -115,6 +111,11 @@ handle_message({forbidden, Msg}, _, _) -> handle_message({request_entity_too_large, Entity}, _, _) -> throw({request_entity_too_large, Entity}). +ignore_failed_worker(Worker, Acc0) -> + {WC, LenDocs, W, GrpDocs, DocReplyDict} = Acc0, + NewGrpDocs = lists:keydelete(Worker, 1, GrpDocs), + skip_message({WC - 1, LenDocs, W, NewGrpDocs, DocReplyDict}). + before_doc_update(DbName, Docs, Opts) -> % Use the same pattern as in couch_db:validate_doc_update/3. If the document was already % checked during the interactive edit we don't want to spend time in the internal replicator diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl index b781eea99..fa75370a3 100644 --- a/src/fabric/src/fabric_rpc.erl +++ b/src/fabric/src/fabric_rpc.erl @@ -274,21 +274,27 @@ get_missing_revs(DbName, IdRevsList, Options) -> with_db(DbName, Options, {couch_db, get_missing_revs, [IdRevsList]}). update_docs(DbName, Docs0, Options) -> - {Docs1, Type} = - case couch_util:get_value(read_repair, Options) of - NodeRevs when is_list(NodeRevs) -> - Filtered = read_repair_filter(DbName, Docs0, NodeRevs, Options), - {Filtered, ?REPLICATED_CHANGES}; - undefined -> - X = - case proplists:get_value(?REPLICATED_CHANGES, Options) of - true -> ?REPLICATED_CHANGES; - _ -> ?INTERACTIVE_EDIT - end, - {Docs0, X} - end, - Docs2 = make_att_readers(Docs1), - with_db(DbName, Options, {couch_db, update_docs, [Docs2, Options, Type]}). + %% only if there's room + case couch_disk_monitor:block_database_writes() of + true -> + rexi:reply({insufficient_storage, <<"database_dir almost full">>}); + false -> + {Docs1, Type} = + case couch_util:get_value(read_repair, Options) of + NodeRevs when is_list(NodeRevs) -> + Filtered = read_repair_filter(DbName, Docs0, NodeRevs, Options), + {Filtered, ?REPLICATED_CHANGES}; + undefined -> + X = + case proplists:get_value(?REPLICATED_CHANGES, Options) of + true -> ?REPLICATED_CHANGES; + _ -> ?INTERACTIVE_EDIT + end, + {Docs0, X} + end, + Docs2 = make_att_readers(Docs1), + with_db(DbName, Options, {couch_db, update_docs, [Docs2, Options, Type]}) + end. get_purge_seq(DbName, Options) -> with_db(DbName, Options, {couch_db, get_purge_seq, []}). @@ -493,7 +499,9 @@ view_cb(complete, Acc) -> ok = rexi:stream_last(complete), {ok, Acc}; view_cb(ok, ddoc_updated) -> - rexi:reply({ok, ddoc_updated}). + rexi:reply({ok, ddoc_updated}); +view_cb(ok, insufficient_storage) -> + rexi:reply({ok, insufficient_storage}). reduce_cb({meta, Meta}, Acc) -> % Map function starting @@ -511,7 +519,9 @@ reduce_cb(complete, Acc) -> ok = rexi:stream_last(complete), {ok, Acc}; reduce_cb(ok, ddoc_updated) -> - rexi:reply({ok, ddoc_updated}). + rexi:reply({ok, ddoc_updated}); +reduce_cb(ok, insufficient_storage) -> + rexi:reply({ok, insufficient_storage}). changes_enumerator(#full_doc_info{} = FDI, Acc) -> changes_enumerator(couch_doc:to_doc_info(FDI), Acc); diff --git a/src/fabric/src/fabric_streams.erl b/src/fabric/src/fabric_streams.erl index 2a3a2b004..318824814 100644 --- a/src/fabric/src/fabric_streams.erl +++ b/src/fabric/src/fabric_streams.erl @@ -144,11 +144,11 @@ handle_stream_start(rexi_STREAM_INIT, {Worker, From}, St) -> {stop, St#stream_acc{workers = [], ready = Ready1}} end end; -handle_stream_start({ok, ddoc_updated}, _, St) -> +handle_stream_start({ok, Error}, _, St) when Error == ddoc_updated; Error == insufficient_storage -> WaitingWorkers = [W || {W, _} <- St#stream_acc.workers], ReadyWorkers = [W || {W, _} <- St#stream_acc.ready], cleanup(WaitingWorkers ++ ReadyWorkers), - {stop, ddoc_updated}; + {stop, Error}; handle_stream_start(Else, _, _) -> exit({invalid_stream_start, Else}). diff --git a/src/fabric/src/fabric_view.erl b/src/fabric/src/fabric_view.erl index 82f9c2bc6..bad5e32c2 100644 --- a/src/fabric/src/fabric_view.erl +++ b/src/fabric/src/fabric_view.erl @@ -415,7 +415,14 @@ maybe_update_others( ViewName, #mrargs{update = lazy} = Args ) -> - ShardsNeedUpdated = mem3:shards(DbName) -- ShardsInvolved, + BlockInteractiveIndexing = couch_disk_monitor:block_interactive_view_indexing(), + ShardsNeedUpdated = + case BlockInteractiveIndexing of + true -> + []; + false -> + mem3:shards(DbName) -- ShardsInvolved + end, lists:foreach( fun(#shard{node = Node, name = ShardName}) -> rpc:cast(Node, fabric_rpc, update_mrview, [ShardName, DDoc, ViewName, Args]) diff --git a/src/fabric/src/fabric_view_map.erl b/src/fabric/src/fabric_view_map.erl index 7b3b75863..43922d5d5 100644 --- a/src/fabric/src/fabric_view_map.erl +++ b/src/fabric/src/fabric_view_map.erl @@ -56,6 +56,10 @@ go(Db, Options, DDoc, View, Args0, Callback, Acc, VInfo) -> of {ok, ddoc_updated} -> Callback({error, ddoc_updated}, Acc); + {ok, insufficient_storage} -> + Callback( + {error, {insufficient_storage, <<"not enough room to update index">>}}, Acc + ); {ok, Workers} -> try go(DbName, Workers, VInfo, CoordArgs, Callback, Acc) @@ -207,6 +211,8 @@ handle_message({execution_stats, _} = Msg, {_, From}, St) -> rexi:stream_ack(From), {Go, St#collector{user_acc = Acc}}; handle_message(ddoc_updated, _Worker, State) -> + {stop, State}; +handle_message(insufficient_storage, _Worker, State) -> {stop, State}. merge_row(Dir, Collation, undefined, Row, Rows0) -> diff --git a/src/fabric/src/fabric_view_reduce.erl b/src/fabric/src/fabric_view_reduce.erl index 90fa523a1..d4d17d5e1 100644 --- a/src/fabric/src/fabric_view_reduce.erl +++ b/src/fabric/src/fabric_view_reduce.erl @@ -47,6 +47,8 @@ go(Db, DDoc, VName, Args, Callback, Acc, VInfo) -> of {ok, ddoc_updated} -> Callback({error, ddoc_updated}, Acc); + {ok, insufficient_storage} -> + Callback({error, insufficient_storage}, Acc); {ok, Workers} -> try go2(DbName, Workers, VInfo, CoordArgs, Callback, Acc) @@ -175,6 +177,8 @@ handle_message(complete, Worker, #collector{counters = Counters0} = State) -> C1 = fabric_dict:update_counter(Worker, 1, Counters0), fabric_view:maybe_send_row(State#collector{counters = C1}); handle_message(ddoc_updated, _Worker, State) -> + {stop, State}; +handle_message(insufficient_storage, _Worker, State) -> {stop, State}. os_proc_needed(<<"_", _/binary>>) -> false; diff --git a/src/ken/src/ken_server.erl b/src/ken/src/ken_server.erl index 929ba47d7..f1d838a74 100644 --- a/src/ken/src/ken_server.erl +++ b/src/ken/src/ken_server.erl @@ -487,11 +487,15 @@ should_start_job(#job{name = Name, seq = Seq, server = Pid}, State) -> IncrementalChannels = list_to_integer(config("incremental_channels", "80")), BatchChannels = list_to_integer(config("batch_channels", "20")), TotalChannels = IncrementalChannels + BatchChannels, + BlockBackgroundViewIndexing = couch_disk_monitor:block_background_view_indexing(), + IsView = tuple_size(Name) == 2, % View name has two elements A = get_active_count(), #state{delay = Delay, batch_size = BS} = State, case ets:lookup(ken_workers, Name) of [] -> if + BlockBackgroundViewIndexing andalso IsView -> + false; A < BatchChannels -> true; A < TotalChannels -> @@ -501,7 +505,7 @@ should_start_job(#job{name = Name, seq = Seq, server = Pid}, State) -> {ok, CurrentSeq} = hastings_index:await(Pid, 0), (Seq - CurrentSeq) < Threshold; % View name has two elements. - {_, _} -> + _ when IsView -> % Since seq is 0, couch_index:get_state/2 won't % spawn an index update. {ok, MRSt} = couch_index:get_state(Pid, 0), @@ -525,6 +529,8 @@ should_start_job(#job{name = Name, seq = Seq, server = Pid}, State) -> Now = erlang:monotonic_time(), DeltaT = erlang:convert_time_unit(Now - LRU, native, millisecond), if + BlockBackgroundViewIndexing andalso IsView -> + false; A < BatchChannels, (Seq - OldSeq) >= BS -> true; A < BatchChannels, DeltaT > Delay ->
