You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2020/03/09 19:10:41 UTC
[couchdb] branch prototype/fdb-layer updated (97458c3 -> 132cf7a)
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a change to branch prototype/fdb-layer
in repository https://gitbox.apache.org/repos/asf/couchdb.git.
from 97458c3 Test size calculations for view rows
new 17ce741 Optimize resubmitting pending jobs
new 02ca72b Implement a simple index auto-updater
new 132cf7a Enable index auto-updating for couch_views
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
rel/files/eunit.ini | 6 +-
rel/overlay/etc/default.ini | 10 +
src/couch_jobs/src/couch_jobs_fdb.erl | 11 +
src/couch_jobs/test/couch_jobs_tests.erl | 11 +
src/couch_views/src/couch_views.erl | 21 +-
src/couch_views/src/couch_views_sup.erl | 8 +
src/couch_views/test/couch_views_indexer_test.erl | 18 +-
src/fabric/src/fabric2_db.erl | 4 +
src/fabric/src/fabric2_index.erl | 222 ++++++++++++++++
src/fabric/src/fabric2_sup.erl | 8 +
src/fabric/test/fabric2_index_tests.erl | 304 ++++++++++++++++++++++
11 files changed, 620 insertions(+), 3 deletions(-)
create mode 100644 src/fabric/src/fabric2_index.erl
create mode 100644 src/fabric/test/fabric2_index_tests.erl
[couchdb] 01/03: Optimize resubmitting pending jobs
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch prototype/fdb-layer
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 17ce741c7cfcb14df7a8f24d38a59a7bf302dd5d
Author: Nick Vatamaniuc <va...@apache.org>
AuthorDate: Sat Mar 7 23:53:22 2020 -0500
Optimize resubmitting pending jobs
Previously even when the scheduled time was the same the job was still deleted
and re-inserted into the pending queue. Now we perform the re-enqueing
operation only if the scheduled time has changed. So if the whole operation is
run in its own transaction, the transaction will now become a read-only
transaction.
This optimization should come in handy with the indexing auto-builder, for
example, where multiple nodes might try to re-enqueue the job and then only the
first would succeed, and the rest will perform a quick one row read and not do
any writes.
---
src/couch_jobs/src/couch_jobs_fdb.erl | 11 +++++++++++
src/couch_jobs/test/couch_jobs_tests.erl | 11 +++++++++++
2 files changed, 22 insertions(+)
diff --git a/src/couch_jobs/src/couch_jobs_fdb.erl b/src/couch_jobs/src/couch_jobs_fdb.erl
index e59387e..8c1ab7a 100644
--- a/src/couch_jobs/src/couch_jobs_fdb.erl
+++ b/src/couch_jobs/src/couch_jobs_fdb.erl
@@ -225,6 +225,17 @@ resubmit(#{jtx := true} = JTx0, #{job := true} = Job, NewSTime) ->
data => Data
},
{ok, Job1};
+ pending when STime == OldSTime ->
+ % If pending and scheduled time doesn't change avoid generating
+ % un-necessary writes by removing and re-adding the jobs into the
+ % pending queue.
+ Job1 = Job#{
+ stime => STime,
+ seq => ?PENDING_SEQ,
+ state => pending,
+ data => Data
+ },
+ {ok, Job1};
pending ->
JV1 = JV#jv{seq = ?PENDING_SEQ, stime = STime},
set_job_val(Tx, Key, JV1),
diff --git a/src/couch_jobs/test/couch_jobs_tests.erl b/src/couch_jobs/test/couch_jobs_tests.erl
index 62a75c8..9d8e2df 100644
--- a/src/couch_jobs/test/couch_jobs_tests.erl
+++ b/src/couch_jobs/test/couch_jobs_tests.erl
@@ -35,6 +35,7 @@ couch_jobs_basic_test_() ->
[
fun add_remove_pending/1,
fun add_remove_errors/1,
+ fun add_with_the_same_scheduled_time/1,
fun get_job_data_and_state/1,
fun resubmit_as_job_creator/1,
fun type_timeouts_and_server/1,
@@ -159,6 +160,16 @@ add_remove_errors(#{t1 := T, j1 := J}) ->
end).
+add_with_the_same_scheduled_time(#{t1 := T, j1 := J}) ->
+ ?_test(begin
+ ?assertEqual(ok, couch_jobs:add(?TX, T, J, #{})),
+ fabric2_fdb:transactional(fun(Tx) ->
+ ?assertEqual(ok, couch_jobs:add(Tx, T, J, #{})),
+ ?assert(erlfdb:is_read_only(Tx))
+ end)
+ end).
+
+
resubmit_as_job_creator(#{t1 := T, j1 := J}) ->
?_test(begin
Data = #{<<"x">> => 42},
[couchdb] 02/03: Implement a simple index auto-updater
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch prototype/fdb-layer
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 02ca72ba34eb1768a631f12e34022464cf70278f
Author: Nick Vatamaniuc <va...@apache.org>
AuthorDate: Sun Mar 8 00:03:15 2020 -0500
Implement a simple index auto-updater
The main logic is as follows:
- After doc update transaction(s) are completed in the `fabric2_db` module,
call the `fabric2_index:db_updated(Db)` function.
- `fabric2_index:db_updated(Db)` inserts a `{DbName, Timestamp}` tuple into
one of the sharded ets tables. The tuple is inserted using `ets:insert_new`,
which ensures only the first entry succeeds and others will be ignored,
until that entry is cleared.
- Each ets table in `fabric2_index` has a simple monitor process that
periodically scans that table. If it finds databases which have been
updated, it notifies all the indices which have registered with
`fabric2_index` to build indices.
There are config settings to disable index auto-updating and to adjust the
delay interval, and the resolution. The interval specifies how long to wait
since the first time the db was modified. The resolution interval specifies how
often to check the ets tables.
Just like in the original ken, design documents can have an `"autoupdate":
false` option to disable auto-updating that design document only.
---
rel/files/eunit.ini | 6 +-
rel/overlay/etc/default.ini | 10 ++
src/fabric/src/fabric2_db.erl | 4 +
src/fabric/src/fabric2_index.erl | 222 +++++++++++++++++++++++
src/fabric/src/fabric2_sup.erl | 8 +
src/fabric/test/fabric2_index_tests.erl | 304 ++++++++++++++++++++++++++++++++
6 files changed, 553 insertions(+), 1 deletion(-)
diff --git a/rel/files/eunit.ini b/rel/files/eunit.ini
index 361ea66..2b73ab3 100644
--- a/rel/files/eunit.ini
+++ b/rel/files/eunit.ini
@@ -35,4 +35,8 @@ level = info
[replicator]
; disable jitter to reduce test run times
-startup_jitter = 0
\ No newline at end of file
+startup_jitter = 0
+
+[fabric]
+; disable index auto-updater to avoid interfering with some of the tests
+index_updater_enabled = false
diff --git a/rel/overlay/etc/default.ini b/rel/overlay/etc/default.ini
index 4c978b2..749cdd2 100644
--- a/rel/overlay/etc/default.ini
+++ b/rel/overlay/etc/default.ini
@@ -226,6 +226,16 @@ port = 6984
; should have a matching directory prefix in order to read and write the same
; data. Changes to this value take effect only on node start-up.
;fdb_directory = couchdb
+;
+; Enable or disable index auto-updater
+;index_autoupdater_enabled = true
+;
+; How long to wait from the first db update event until index building is
+; triggered.
+;index_autoupdater_delay_msec = 60000
+;
+; How often to check if databases may need their indices updated.
+;index_autoupdater_resolution_msec = 10000
; [rexi]
; buffer_count = 2000
diff --git a/src/fabric/src/fabric2_db.erl b/src/fabric/src/fabric2_db.erl
index b0f7849..791282f 100644
--- a/src/fabric/src/fabric2_db.erl
+++ b/src/fabric/src/fabric2_db.erl
@@ -733,6 +733,10 @@ update_docs(Db, Docs0, Options) ->
end)
end, Docs1)
end,
+
+ % Notify index builder
+ fabric2_index:db_updated(name(Db)),
+
% Convert errors
Resps1 = lists:map(fun(Resp) ->
case Resp of
diff --git a/src/fabric/src/fabric2_index.erl b/src/fabric/src/fabric2_index.erl
new file mode 100644
index 0000000..9382105
--- /dev/null
+++ b/src/fabric/src/fabric2_index.erl
@@ -0,0 +1,222 @@
+% 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(fabric2_index).
+
+
+-behaviour(gen_server).
+
+
+-export([
+ register_index/1,
+ db_updated/1,
+ start_link/0
+]).
+
+-export([
+ init/1,
+ terminate/2,
+ handle_call/3,
+ handle_cast/2,
+ handle_info/2,
+ code_change/3
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+
+
+-callback build_indices(Db :: map(), DDocs :: list(#doc{})) ->
+ [{ok, JobId::binary()} | {error, any()}].
+
+
+-define(SHARDS, 32).
+-define(DEFAULT_DELAY_MSEC, 60000).
+-define(DEFAULT_RESOLUTION_MSEC, 10000).
+
+
+register_index(Mod) when is_atom(Mod) ->
+ Indices = lists:usort([Mod | registrations()]),
+ application:set_env(fabric, indices, Indices).
+
+
+db_updated(DbName) when is_binary(DbName) ->
+ Table = table(erlang:phash2(DbName) rem ?SHARDS),
+ ets:insert_new(Table, {DbName, now_msec()}).
+
+
+start_link() ->
+ gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+
+init(_) ->
+ lists:foreach(fun(T) ->
+ spawn_link(fun() -> process_loop(T) end)
+ end, create_tables()),
+ {ok, nil}.
+
+
+terminate(_M, _St) ->
+ ok.
+
+
+handle_call(Msg, _From, St) ->
+ {stop, {bad_call, Msg}, {bad_call, Msg}, St}.
+
+
+handle_cast(Msg, St) ->
+ {stop, {bad_cast, Msg}, St}.
+
+
+handle_info(Msg, St) ->
+ {stop, {bad_info, Msg}, St}.
+
+
+code_change(_OldVsn, St, _Extra) ->
+ {ok, St}.
+
+
+create_tables() ->
+ Opts = [
+ named_table,
+ public,
+ {write_concurrency, true},
+ {read_concurrency, true}
+ ],
+ Tables = [table(N) || N <- lists:seq(0, ?SHARDS - 1)],
+ [ets:new(T, Opts) || T <- Tables].
+
+
+table(Id) when is_integer(Id), Id >= 0 andalso Id < ?SHARDS ->
+ list_to_atom("fabric2_index_" ++ integer_to_list(Id)).
+
+
+process_loop(Table) ->
+ Now = now_msec(),
+ Delay = delay_msec(),
+ Since = Now - Delay,
+ case is_enabled() of
+ true ->
+ process_updates(Table, Since),
+ clean_stale(Table, Since);
+ false ->
+ clean_stale(Table, Now)
+ end,
+ Resolution = resolution_msec(),
+ Jitter = rand:uniform(1 + Resolution div 2),
+ timer:sleep(Resolution + Jitter),
+ process_loop(Table).
+
+
+clean_stale(Table, Since) ->
+ Head = {'_', '$1'},
+ Guard = {'<', '$1', Since},
+ % Monotonic is not strictly monotonic, so we process items using `=<` but
+ % clean with `<` in case there was an update with the same timestamp after
+ % we started processing already at that timestamp.
+ ets:select_delete(Table, [{Head, [Guard], [true]}]).
+
+
+process_updates(Table, Since) ->
+ Head = {'$1', '$2'},
+ Guard = {'=<', '$2', Since},
+ case ets:select(Table, [{Head, [Guard], ['$1']}], 25) of
+ '$end_of_table' -> ok;
+ {Match, Cont} -> process_updates_iter(Match, Cont)
+ end.
+
+
+process_updates_iter([], Cont) ->
+ case ets:select(Cont) of
+ '$end_of_table' -> ok;
+ {Match, Cont1} -> process_updates_iter(Match, Cont1)
+ end;
+
+process_updates_iter([Db | Rest], Cont) ->
+ try
+ process_db(Db)
+ catch
+ error:database_does_not_exist ->
+ ok;
+ Tag:Reason ->
+ Stack = erlang:get_stacktrace(),
+ LogMsg = "~p failed to build indices for `~s` ~p:~p ~p",
+ couch_log:error(LogMsg, [?MODULE, Db, Tag, Reason, Stack])
+ end,
+ process_updates_iter(Rest, Cont).
+
+
+build_indices(_Db, []) ->
+ [];
+
+build_indices(Db, DDocs) ->
+ lists:flatmap(fun(Mod) ->
+ Mod:build_indices(Db, DDocs)
+ end, registrations()).
+
+
+registrations() ->
+ application:get_env(fabric, indices, []).
+
+
+process_db(DbName) when is_binary(DbName) ->
+ {ok, Db} = fabric2_db:open(DbName, [?ADMIN_CTX]),
+ fabric2_fdb:transactional(Db, fun(TxDb) ->
+ DDocs1 = get_design_docs(TxDb),
+ DDocs2 = lists:filter(fun should_update/1, DDocs1),
+ DDocs3 = shuffle(DDocs2),
+ build_indices(TxDb, DDocs3)
+ end).
+
+
+get_design_docs(Db) ->
+ Callback = fun
+ ({meta, _}, Acc) -> {ok, Acc};
+ (complete, Acc) -> {ok, Acc};
+ ({row, Row}, Acc) -> {ok, [get_doc(Db, Row) | Acc]}
+ end,
+ {ok, DDocs} = fabric2_db:fold_design_docs(Db, Callback, [], []),
+ DDocs.
+
+
+get_doc(Db, Row) ->
+ {_, DocId} = lists:keyfind(id, 1, Row),
+ {ok, #doc{deleted = false} = Doc} = fabric2_db:open_doc(Db, DocId, []),
+ Doc.
+
+
+should_update(#doc{body = {Props}}) ->
+ couch_util:get_value(<<"autoupdate">>, Props, true).
+
+
+shuffle(Items) ->
+ Tagged = [{rand:uniform(), I} || I <- Items],
+ Sorted = lists:sort(Tagged),
+ [I || {_T, I} <- Sorted].
+
+
+now_msec() ->
+ erlang:monotonic_time(millisecond).
+
+
+is_enabled() ->
+ config:get_boolean("fabric", "index_updater_enabled", true).
+
+
+delay_msec() ->
+ config:get_integer("fabric", "index_updater_delay_msec",
+ ?DEFAULT_DELAY_MSEC).
+
+
+resolution_msec() ->
+ config:get_integer("fabric", "index_updater_resolution_msec",
+ ?DEFAULT_RESOLUTION_MSEC).
diff --git a/src/fabric/src/fabric2_sup.erl b/src/fabric/src/fabric2_sup.erl
index 2510b13..e8201b4 100644
--- a/src/fabric/src/fabric2_sup.erl
+++ b/src/fabric/src/fabric2_sup.erl
@@ -47,6 +47,14 @@ init([]) ->
5000,
worker,
[fabric2_server]
+ },
+ {
+ fabric2_index,
+ {fabric2_index, start_link, []},
+ permanent,
+ 5000,
+ worker,
+ [fabric2_index]
}
],
ChildrenWithEpi = couch_epi:register_service(fabric2_epi, Children),
diff --git a/src/fabric/test/fabric2_index_tests.erl b/src/fabric/test/fabric2_index_tests.erl
new file mode 100644
index 0000000..3fc8a5b
--- /dev/null
+++ b/src/fabric/test/fabric2_index_tests.erl
@@ -0,0 +1,304 @@
+% 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(fabric2_index_tests).
+
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+-include("fabric2_test.hrl").
+
+
+% Should match fabric2_index define
+-define(SHARDS, 32).
+
+
+index_test_() ->
+ {
+ "Test fabric indexing module",
+ {
+ setup,
+ fun setup/0,
+ fun cleanup/1,
+ with([
+ ?TDEF(register_index_works),
+ ?TDEF(single_update),
+ ?TDEF(multiple_updates),
+ ?TDEF(skip_db_if_no_ddocs),
+ ?TDEF(ignore_deleted_dbs),
+ ?TDEF(check_gen_server_messages)
+ ])
+ }
+ }.
+
+
+index_process_cleanup_test_() ->
+ {
+ "Test fabric process cleanup in indexing module",
+ {
+ foreach,
+ fun setup/0,
+ fun cleanup/1,
+ [
+ ?TDEF_FE(updater_processes_start),
+ ?TDEF_FE(updater_processes_stop),
+ ?TDEF_FE(indexing_can_be_disabled),
+ ?TDEF_FE(handle_indexer_blowing_up)
+ ]
+ }
+ }.
+
+
+setup() ->
+ meck:new(config, [passthrough]),
+ meck:expect(config, get_integer, fun
+ ("fabric", "index_updater_delay_msec", _) -> 200;
+ ("fabric", "index_updater_resolution_msec", _) -> 100;
+
+ (_, _, Default) -> Default
+ end),
+ meck:expect(config, get_boolean, fun
+ ("fabric", "index_updater_enabled", _) -> true;
+ (_, _, Default) -> Default
+ end),
+
+ Indices = application:get_env(fabric, indices, []),
+
+ Ctx = test_util:start_couch([fabric]),
+
+ % Db1 has a valid design doc, a deleted one and one with "autoupdate":false
+ {ok, Db1} = fabric2_db:create(?tempdb(), [?ADMIN_CTX]),
+ {_, _} = create_doc(Db1, <<"_design/doc1">>),
+
+ DDocId2 = <<"_design/doc2">>,
+ {DDocId2, {Pos, Rev}} = create_doc(Db1, DDocId2),
+ Delete2 = #doc{id = DDocId2, revs = {Pos, [Rev]}, deleted = true},
+ {ok, _} = fabric2_db:update_doc(Db1, Delete2),
+
+ NoAutoUpdate = {[{<<"autoupdate">>, false}]},
+ {_, _} = create_doc(Db1, <<"_design/doc3">>, NoAutoUpdate),
+
+ % Db2 doesn't have any desig documents
+ {ok, Db2} = fabric2_db:create(?tempdb(), [?ADMIN_CTX]),
+
+ #{db1 => Db1, db2 => Db2, ctx => Ctx, indices => Indices}.
+
+
+cleanup(#{db1 := Db1, db2 := Db2, ctx := Ctx, indices := Indices}) ->
+ catch fabric2_db:delete(fabric2_db:name(Db1), []),
+ catch fabric2_db:delete(fabric2_db:name(Db2), []),
+
+ test_util:stop_couch(Ctx),
+ application:set_env(fabric, indices, Indices),
+
+ meck:unload().
+
+
+register_index_works(_) ->
+ reset_callbacks(),
+
+ Mod1 = fabric2_test_callback1,
+ fabric2_index:register_index(Mod1),
+ Indices1 = application:get_env(fabric, indices, []),
+ ?assertEqual([Mod1], Indices1),
+
+ Mod2 = fabric2_test_callback2,
+ fabric2_index:register_index(Mod2),
+ Indices2 = application:get_env(fabric, indices, []),
+ ?assertEqual(lists:sort([Mod1, Mod2]), lists:sort(Indices2)).
+
+
+single_update(#{db1 := Db}) ->
+ reset_callbacks(),
+
+ Mod = fabric2_test_callback3,
+ setup_callback(Mod),
+ create_doc(Db),
+
+ meck:wait(Mod, build_indices, 2, 2000),
+ ?assertEqual(1, meck:num_calls(Mod, build_indices, 2)).
+
+
+multiple_updates(#{db1 := Db}) ->
+ reset_callbacks(),
+
+ Mod = fabric2_test_callback4,
+ setup_callback(Mod),
+ create_docs(Db, 10),
+
+ % should be called at least once
+ meck:wait(Mod, build_indices, 2, 2000),
+
+ % Maybe called another time or two at most
+ timer:sleep(500),
+ ?assert(meck:num_calls(Mod, build_indices, 2) =< 3).
+
+
+skip_db_if_no_ddocs(#{db2 := Db}) ->
+ reset_callbacks(),
+
+ Mod = fabric2_test_callback5,
+ setup_callback(Mod),
+ create_doc(Db),
+
+ timer:sleep(500),
+ ?assertEqual(0, meck:num_calls(Mod, build_indices, 2)).
+
+
+ignore_deleted_dbs(#{}) ->
+ reset_callbacks(),
+
+ Mod = fabric2_test_callback6,
+ setup_callback(Mod),
+ lists:foreach(fun(_) ->
+ RandomDbName = fabric2_util:uuid(),
+ fabric2_index:db_updated(RandomDbName)
+ end, lists:seq(1, 10000)),
+
+ test_util:wait(fun() ->
+ case table_sizes() =:= 0 of
+ true -> ok;
+ false -> wait
+ end
+ end, 5000).
+
+
+check_gen_server_messages(#{}) ->
+ CallExpect = {stop, {bad_call, foo}, {bad_call, foo}, baz},
+ CastExpect = {stop, {bad_cast, foo}, bar},
+ InfoExpect = {stop, {bad_info, foo}, bar},
+ ?assertEqual(CallExpect, fabric2_index:handle_call(foo, bar, baz)),
+ ?assertEqual(CastExpect, fabric2_index:handle_cast(foo, bar)),
+ ?assertEqual(InfoExpect, fabric2_index:handle_info(foo, bar)),
+ ?assertEqual(ok, fabric2_index:terminate(shutdown, nil)),
+ ?assertEqual({ok, nil}, fabric2_index:code_change(v0, nil, extra)).
+
+
+updater_processes_start(#{}) ->
+ Pid = whereis(fabric2_index),
+ ?assert(is_process_alive(Pid)),
+ lists:map(fun(N) ->
+ ?assertEqual(tid(N), ets:info(tid(N), name))
+ end, lists:seq(0, ?SHARDS - 1)).
+
+
+updater_processes_stop(#{}) ->
+ Refs = lists:map(fun(N) ->
+ Pid = ets:info(tid(N), owner),
+ ?assert(is_process_alive(Pid)),
+ monitor(process, Pid)
+ end, lists:seq(0, ?SHARDS - 1)),
+
+ % We stop but don't restart fabric after this as we're running in a foreach
+ % test list where app restart happens after each test.
+ application:stop(fabric),
+
+ lists:foreach(fun(Ref) ->
+ receive
+ {'DOWN', Ref, _, _, _} -> ok
+ after 3000 ->
+ ?assert(false)
+ end
+ end, Refs).
+
+
+indexing_can_be_disabled(#{db1 := Db}) ->
+ Mod = fabric2_test_callback7,
+ setup_callback(Mod),
+
+ meck:expect(config, get_boolean, fun
+ ("fabric", "index_updater_enabled", _) -> false;
+ (_, _, Default) -> Default
+ end),
+
+ create_doc(Db),
+ timer:sleep(500),
+ ?assertEqual(0, meck:num_calls(Mod, build_indices, 2)),
+
+ meck:expect(config, get_boolean, fun
+ ("fabric", "index_updater_enabled", _) -> true;
+ (_, _, Default) -> Default
+ end),
+
+ create_doc(Db),
+ meck:wait(Mod, build_indices, 2, 2000).
+
+
+handle_indexer_blowing_up(#{db1 := Db}) ->
+ Mod = fabric2_test_callback8,
+ setup_callback(Mod),
+ meck:expect(Mod, build_indices, fun(_, _) -> error(bad_index) end),
+
+ MainPid = whereis(fabric2_index),
+ WPids1 = [ets:info(tid(N), owner) || N <- lists:seq(0, ?SHARDS - 1)],
+
+ create_doc(Db),
+ meck:wait(Mod, build_indices, 2, 2000),
+
+ ?assert(is_process_alive(MainPid)),
+
+ WPids2 = [ets:info(tid(N), owner) || N <- lists:seq(0, ?SHARDS - 1)],
+ ?assertEqual(lists:sort(WPids1), lists:sort(WPids2)),
+ ?assert(lists:all(fun(Pid) -> is_process_alive(Pid) end, WPids2)).
+
+
+% Utility functions
+
+setup_callback(Mod) ->
+ catch meck:unload(Mod),
+ meck:new(Mod, [non_strict]),
+ meck:expect(Mod, build_indices, 2, []),
+ fabric2_index:register_index(Mod).
+
+
+reset_callbacks() ->
+ Mods = application:get_env(fabric, indices, []),
+ application:set_env(fabric, indices, []),
+ lists:foreach(fun(M) ->
+ catch meck:reset(M),
+ catch meck:unload(M)
+ end, Mods).
+
+
+tid(Id) when is_integer(Id) ->
+ TableName = "fabric2_index_" ++ integer_to_list(Id),
+ list_to_existing_atom(TableName).
+
+
+table_sizes() ->
+ Sizes = [ets:info(tid(N), size) || N <- lists:seq(0, ?SHARDS - 1)],
+ lists:sum(Sizes).
+
+
+create_docs(Db, Count) ->
+ lists:map(fun(_) ->
+ {DocId, _RevStr} = create_doc(Db),
+ DocId
+ end, lists:seq(1, Count)).
+
+
+create_doc(Db) ->
+ create_doc(Db, fabric2_util:uuid()).
+
+
+create_doc(Db, DocId) ->
+ create_doc(Db, DocId, {[]}).
+
+
+create_doc(Db, DocId, Body) ->
+ Doc = #doc{
+ id = DocId,
+ body = Body
+ },
+ {ok, {Pos, Rev}} = fabric2_db:update_doc(Db, Doc, []),
+ {DocId, {Pos, Rev}}.
[couchdb] 03/03: Enable index auto-updating for couch_views
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch prototype/fdb-layer
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 132cf7a9ba797a1cf228c39ccf6448a683a934d0
Author: Nick Vatamaniuc <va...@apache.org>
AuthorDate: Sun Mar 8 00:38:48 2020 -0500
Enable index auto-updating for couch_views
* Register with the fabric2_index module
* Provide a build_indices/2 callback
* In the callback attempt to parse an `#mrst{}` indexing context and then
trigger an indexing job. We don't wait for the job to finish but instead
rely on `couch_job`'s max worker limit to keep concurrency in check.
---
src/couch_views/src/couch_views.erl | 21 ++++++++++++++++++++-
src/couch_views/src/couch_views_sup.erl | 8 ++++++++
src/couch_views/test/couch_views_indexer_test.erl | 18 +++++++++++++++++-
3 files changed, 45 insertions(+), 2 deletions(-)
diff --git a/src/couch_views/src/couch_views.erl b/src/couch_views/src/couch_views.erl
index 322415b..58cfb24 100644
--- a/src/couch_views/src/couch_views.erl
+++ b/src/couch_views/src/couch_views.erl
@@ -12,8 +12,15 @@
-module(couch_views).
+
+-behavior(fabric2_index).
+
+
-export([
- query/6
+ query/6,
+
+ % fabric2_index behavior
+ build_indices/2
]).
@@ -55,6 +62,18 @@ query(Db, DDoc, ViewName, Callback, Acc0, Args0) ->
end.
+build_indices(#{} = Db, DDocs) when is_list(DDocs) ->
+ DbName = fabric2_db:name(Db),
+ lists:filtermap(fun(DDoc) ->
+ try couch_views_util:ddoc_to_mrst(DbName, DDoc) of
+ {ok, #mrst{} = Mrst} ->
+ {true, couch_views_jobs:build_view_async(Db, Mrst)}
+ catch _:_ ->
+ false
+ end
+ end, DDocs).
+
+
read_view(Db, Mrst, ViewName, Callback, Acc0, Args) ->
fabric2_fdb:transactional(Db, fun(TxDb) ->
try
diff --git a/src/couch_views/src/couch_views_sup.erl b/src/couch_views/src/couch_views_sup.erl
index 7a72a1f..2a40f0a 100644
--- a/src/couch_views/src/couch_views_sup.erl
+++ b/src/couch_views/src/couch_views_sup.erl
@@ -28,6 +28,7 @@
start_link() ->
+ ok = register_views_index(),
Arg = case fabric2_node_types:is_type(view_indexing) of
true -> normal;
false -> builds_disabled
@@ -50,6 +51,13 @@ init(builds_disabled) ->
{ok, {flags(), []}}.
+register_views_index() ->
+ case fabric2_node_types:is_type(api_frontend) of
+ true -> fabric2_index:register_index(couch_views);
+ false -> ok
+ end.
+
+
flags() ->
#{
strategy => one_for_one,
diff --git a/src/couch_views/test/couch_views_indexer_test.erl b/src/couch_views/test/couch_views_indexer_test.erl
index cd5b2b0..02a12e7 100644
--- a/src/couch_views/test/couch_views_indexer_test.erl
+++ b/src/couch_views/test/couch_views_indexer_test.erl
@@ -41,7 +41,8 @@ indexer_test_() ->
?TDEF_FE(multipe_identical_keys_from_same_doc),
?TDEF_FE(fewer_multipe_identical_keys_from_same_doc),
?TDEF_FE(handle_size_key_limits),
- ?TDEF_FE(handle_size_value_limits)
+ ?TDEF_FE(handle_size_value_limits),
+ ?TDEF_FE(index_autoupdater_callback)
]
}
}
@@ -536,6 +537,21 @@ handle_size_value_limits(Db) ->
], Out1).
+index_autoupdater_callback(Db) ->
+ DDoc = create_ddoc(),
+ Doc1 = doc(0),
+ {ok, _} = fabric2_db:update_doc(Db, DDoc, []),
+ {ok, _} = fabric2_db:update_doc(Db, Doc1, []),
+
+ DbSeq = fabric2_db:get_update_seq(Db),
+
+ Result = couch_views:build_indices(Db, [DDoc]),
+ ?assertMatch([{ok, <<_/binary>>}], Result),
+ [{ok, JobId}] = Result,
+
+ ?assertEqual(ok, couch_views_jobs:wait_for_job(JobId, DbSeq)).
+
+
row(Id, Key, Value) ->
{row, [
{id, Id},