You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by rn...@apache.org on 2021/11/30 17:51:52 UTC
[couchdb] branch sharded_couch_index_server created (now 22f4be7)
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a change to branch sharded_couch_index_server
in repository https://gitbox.apache.org/repos/asf/couchdb.git.
at 22f4be7 Add sharding to couch_index_server
This branch includes the following new commits:
new 67b4bfc [unrelated] improve erlang_ls.config
new 22f4be7 Add sharding to couch_index_server
The 2 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.
[couchdb] 02/02: Add sharding to couch_index_server
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch sharded_couch_index_server
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 22f4be730f6f8b970a3683472a83abac676bcf72
Author: Robert Newson <rn...@apache.org>
AuthorDate: Tue Nov 30 15:39:58 2021 +0000
Add sharding to couch_index_server
---
src/couch/src/couch_secondary_sup.erl | 11 +-
src/couch_index/src/couch_index_server.erl | 184 ++++++++++++++++++-----------
2 files changed, 126 insertions(+), 69 deletions(-)
diff --git a/src/couch/src/couch_secondary_sup.erl b/src/couch/src/couch_secondary_sup.erl
index bb78215..87d1113 100644
--- a/src/couch/src/couch_secondary_sup.erl
+++ b/src/couch/src/couch_secondary_sup.erl
@@ -27,11 +27,10 @@ init([]) ->
dynamic}
],
Daemons = [
- {index_server, {couch_index_server, start_link, []}},
{query_servers, {couch_proc_manager, start_link, []}},
{vhosts, {couch_httpd_vhost, start_link, []}},
{uuids, {couch_uuids, start, []}}
- ],
+ ] ++ couch_index_servers(),
MaybeHttp = case http_enabled() of
true -> [{httpd, {couch_httpd, start_link, []}}];
@@ -70,3 +69,11 @@ https_enabled() ->
LegacySSLEnabled = LegacySSL =:= "{chttpd, start_link, [https]}",
SSLEnabled orelse LegacySSLEnabled.
+
+couch_index_servers() ->
+ N = couch_index_server:num_servers(),
+ [couch_index_server(I) || I <- lists:seq(1, N)].
+
+couch_index_server(N) ->
+ Name = couch_index_server:couch_index_server(N),
+ {Name, {couch_index_server, start_link, [N]}}.
diff --git a/src/couch_index/src/couch_index_server.erl b/src/couch_index/src/couch_index_server.erl
index 6bebff2..bd4e453 100644
--- a/src/couch_index/src/couch_index_server.erl
+++ b/src/couch_index/src/couch_index_server.erl
@@ -16,10 +16,11 @@
-vsn(2).
--export([start_link/0, validate/2, get_index/4, get_index/3, get_index/2]).
+-export([start_link/1, validate/2, get_index/4, get_index/3, get_index/2]).
-export([init/1, terminate/2, code_change/3]).
-export([handle_call/3, handle_cast/2, handle_info/2]).
+-export([num_servers/0, couch_index_server/1]).
% Exported for callbacks
-export([
@@ -30,15 +31,18 @@
-include_lib("couch/include/couch_db.hrl").
--define(BY_SIG, couchdb_indexes_by_sig).
--define(BY_PID, couchdb_indexes_by_pid).
--define(BY_DB, couchdb_indexes_by_db).
-define(RELISTEN_DELAY, 5000).
--record(st, {root_dir}).
+-record(st, {
+ root_dir,
+ index_server,
+ by_sig,
+ by_pid,
+ by_db
+}).
-start_link() ->
- gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+start_link(N) ->
+ gen_server:start_link({local, couch_index_server(N)}, ?MODULE, [N], []).
validate(Db, DDoc) ->
@@ -103,88 +107,95 @@ get_index(Module, Db, DDoc, _Fun) ->
get_index(Module, IdxState) ->
DbName = Module:get(db_name, IdxState),
Sig = Module:get(signature, IdxState),
- case ets:lookup(?BY_SIG, {DbName, Sig}) of
+ case ets:lookup(by_sig(DbName), {DbName, Sig}) of
[{_, Pid}] when is_pid(Pid) ->
DDocId = Module:get(idx_name, IdxState),
- case ets:match_object(?BY_DB, {DbName, {DDocId, Sig}}) of
+ case ets:match_object(by_db(DbName), {DbName, {DDocId, Sig}}) of
[] ->
Args = [Pid, DbName, DDocId, Sig],
- gen_server:cast(?MODULE, {add_to_ets, Args});
+ gen_server:cast(couch_index_server(DbName), {add_to_ets, Args});
_ -> ok
end,
{ok, Pid};
_ ->
Args = {Module, IdxState, DbName, Sig},
- gen_server:call(?MODULE, {get_index, Args}, infinity)
+ gen_server:call(couch_index_server(DbName), {get_index, Args}, infinity)
end.
-init([]) ->
+init([N]) ->
process_flag(trap_exit, true),
- ok = config:listen_for_changes(?MODULE, couch_index_util:root_dir()),
- ets:new(?BY_SIG, [protected, set, named_table]),
- ets:new(?BY_PID, [private, set, named_table]),
- ets:new(?BY_DB, [protected, bag, named_table]),
- couch_event:link_listener(?MODULE, handle_db_event, nil, [all_dbs]),
+ ets:new(by_sig(N), [protected, set, named_table]),
+ ets:new(by_pid(N), [private, set, named_table]),
+ ets:new(by_db(N), [protected, bag, named_table]),
RootDir = couch_index_util:root_dir(),
couch_file:init_delete_dir(RootDir),
- {ok, #st{root_dir=RootDir}}.
+ St = #st{
+ root_dir=RootDir,
+ index_server=couch_index_server(N),
+ by_sig=by_sig(N),
+ by_pid=by_pid(N),
+ by_db=by_db(N)
+ },
+ ok = config:listen_for_changes(?MODULE, St),
+ couch_event:link_listener(?MODULE, handle_db_event, St, [all_dbs]),
+ {ok, St}.
-terminate(_Reason, _State) ->
- Pids = [Pid || {Pid, _} <- ets:tab2list(?BY_PID)],
+terminate(_Reason, State) ->
+ Pids = [Pid || {Pid, _} <- ets:tab2list(State#st.by_pid)],
lists:map(fun couch_util:shutdown_sync/1, Pids),
ok.
handle_call({get_index, {_Mod, _IdxState, DbName, Sig}=Args}, From, State) ->
- case ets:lookup(?BY_SIG, {DbName, Sig}) of
+ case ets:lookup(State#st.by_sig, {DbName, Sig}) of
[] ->
spawn_link(fun() -> new_index(Args) end),
- ets:insert(?BY_SIG, {{DbName, Sig}, [From]}),
+ ets:insert(State#st.by_sig, {{DbName, Sig}, [From]}),
{noreply, State};
[{_, Waiters}] when is_list(Waiters) ->
- ets:insert(?BY_SIG, {{DbName, Sig}, [From | Waiters]}),
+ ets:insert(State#st.by_sig, {{DbName, Sig}, [From | Waiters]}),
{noreply, State};
[{_, Pid}] when is_pid(Pid) ->
{reply, {ok, Pid}, State}
end;
handle_call({async_open, {DbName, DDocId, Sig}, {ok, Pid}}, _From, State) ->
- [{_, Waiters}] = ets:lookup(?BY_SIG, {DbName, Sig}),
+ [{_, Waiters}] = ets:lookup(State#st.by_sig, {DbName, Sig}),
[gen_server:reply(From, {ok, Pid}) || From <- Waiters],
link(Pid),
add_to_ets(DbName, Sig, DDocId, Pid),
{reply, ok, State};
handle_call({async_error, {DbName, _DDocId, Sig}, Error}, _From, State) ->
- [{_, Waiters}] = ets:lookup(?BY_SIG, {DbName, Sig}),
+ [{_, Waiters}] = ets:lookup(State#st.by_sig, {DbName, Sig}),
[gen_server:reply(From, Error) || From <- Waiters],
- ets:delete(?BY_SIG, {DbName, Sig}),
+ ets:delete(State#st.by_sig, {DbName, Sig}),
{reply, ok, State};
handle_call({reset_indexes, DbName}, _From, State) ->
- reset_indexes(DbName, State#st.root_dir),
+ reset_indexes(DbName, State),
{reply, ok, State}.
handle_cast({reset_indexes, DbName}, State) ->
- reset_indexes(DbName, State#st.root_dir),
+ reset_indexes(DbName, State),
{noreply, State};
handle_cast({add_to_ets, [Pid, DbName, DDocId, Sig]}, State) ->
% check if Pid still exists
- case ets:lookup(?BY_PID, Pid) of
+ case ets:lookup(State#st.by_pid, Pid) of
[{Pid, {DbName, Sig}}] when is_pid(Pid) ->
- ets:insert(?BY_DB, {DbName, {DDocId, Sig}});
+ ets:insert(State#st.by_db, {DbName, {DDocId, Sig}});
_ -> ok
end,
{noreply, State};
handle_cast({rem_from_ets, [DbName, DDocId, Sig]}, State) ->
- ets:delete_object(?BY_DB, {DbName, {DDocId, Sig}}),
+ ets:delete_object(State#st.by_db, {DbName, {DDocId, Sig}}),
{noreply, State}.
handle_info({'EXIT', Pid, Reason}, Server) ->
- case ets:lookup(?BY_PID, Pid) of
+ case ets:lookup(Server#st.by_pid, Pid) of
[{Pid, {DbName, Sig}}] ->
DDocIds = [DDocId || {_, {DDocId, _}}
- <- ets:match_object(?BY_DB, {DbName, {'$1', Sig}})],
+ <- ets:match_object(Server#st.by_db, {DbName, {'$1', Sig}})],
rem_from_ets(DbName, Sig, DDocIds, Pid);
[] when Reason /= normal ->
exit(Reason);
@@ -196,7 +207,7 @@ handle_info(restart_config_listener, State) ->
ok = config:listen_for_changes(?MODULE, couch_index_util:root_dir()),
{noreply, State};
handle_info(Msg, State) ->
- couch_log:warning("~p did not expect ~p", [?MODULE, Msg]),
+ couch_log:warning("~p did not expect ~p", [State#st.index_server, Msg]),
{noreply, State}.
@@ -204,23 +215,23 @@ code_change(_OldVsn, State, _Extra) ->
{ok, State}.
-handle_config_change("couchdb", "index_dir", RootDir, _, RootDir) ->
- {ok, RootDir};
-handle_config_change("couchdb", "view_index_dir", RootDir, _, RootDir) ->
- {ok, RootDir};
-handle_config_change("couchdb", "index_dir", _, _, _) ->
- exit(whereis(couch_index_server), config_change),
+handle_config_change("couchdb", "index_dir", RootDir, _, #st{root_dir = RootDir} = St) ->
+ {ok, St};
+handle_config_change("couchdb", "view_index_dir", RootDir, _, #st{root_dir = RootDir} = St) ->
+ {ok, St};
+handle_config_change("couchdb", "index_dir", _, _, St) ->
+ exit(whereis(St#st.index_server), config_change),
remove_handler;
-handle_config_change("couchdb", "view_index_dir", _, _, _) ->
- exit(whereis(couch_index_server), config_change),
+handle_config_change("couchdb", "view_index_dir", _, _, St) ->
+ exit(whereis(St#st.index_server), config_change),
remove_handler;
-handle_config_change(_, _, _, _, RootDir) ->
- {ok, RootDir}.
+handle_config_change(_, _, _, _, St) ->
+ {ok, St}.
handle_config_terminate(_, stop, _) ->
ok;
-handle_config_terminate(_Server, _Reason, _State) ->
- erlang:send_after(?RELISTEN_DELAY, whereis(?MODULE), restart_config_listener),
+handle_config_terminate(_Server, _Reason, State) ->
+ erlang:send_after(?RELISTEN_DELAY, whereis(State#st.index_server), restart_config_listener),
{ok, couch_index_util:root_dir()}.
new_index({Mod, IdxState, DbName, Sig}) ->
@@ -228,21 +239,21 @@ new_index({Mod, IdxState, DbName, Sig}) ->
case couch_index:start_link({Mod, IdxState}) of
{ok, Pid} ->
ok = gen_server:call(
- ?MODULE, {async_open, {DbName, DDocId, Sig}, {ok, Pid}}),
+ couch_index_server(DbName), {async_open, {DbName, DDocId, Sig}, {ok, Pid}}),
unlink(Pid);
Error ->
ok = gen_server:call(
- ?MODULE, {async_error, {DbName, DDocId, Sig}, Error})
+ couch_index_server(DbName), {async_error, {DbName, DDocId, Sig}, Error})
end.
-reset_indexes(DbName, Root) ->
+reset_indexes(DbName, #st{} = State) ->
% shutdown all the updaters and clear the files, the db got changed
SigDDocIds = lists:foldl(fun({_, {DDocId, Sig}}, DDict) ->
dict:append(Sig, DDocId, DDict)
- end, dict:new(), ets:lookup(?BY_DB, DbName)),
+ end, dict:new(), ets:lookup(State#st.by_db, DbName)),
Fun = fun({Sig, DDocIds}) ->
- [{_, Pid}] = ets:lookup(?BY_SIG, {DbName, Sig}),
+ [{_, Pid}] = ets:lookup(State#st.by_sig, {DbName, Sig}),
unlink(Pid),
gen_server:cast(Pid, delete),
receive
@@ -256,28 +267,28 @@ reset_indexes(DbName, Root) ->
end,
lists:foreach(Fun, dict:to_list(SigDDocIds)),
Path = couch_index_util:index_dir("", DbName),
- couch_file:nuke_dir(Root, Path).
+ couch_file:nuke_dir(State#st.root_dir, Path).
add_to_ets(DbName, Sig, DDocId, Pid) ->
- ets:insert(?BY_SIG, {{DbName, Sig}, Pid}),
- ets:insert(?BY_PID, {Pid, {DbName, Sig}}),
- ets:insert(?BY_DB, {DbName, {DDocId, Sig}}).
+ ets:insert(by_sig(DbName), {{DbName, Sig}, Pid}),
+ ets:insert(by_pid(DbName), {Pid, {DbName, Sig}}),
+ ets:insert(by_db(DbName), {DbName, {DDocId, Sig}}).
rem_from_ets(DbName, Sig, DDocIds, Pid) ->
- ets:delete(?BY_SIG, {DbName, Sig}),
- ets:delete(?BY_PID, Pid),
+ ets:delete(by_sig(DbName), {DbName, Sig}),
+ ets:delete(by_pid(DbName), Pid),
lists:foreach(fun(DDocId) ->
- ets:delete_object(?BY_DB, {DbName, {DDocId, Sig}})
+ ets:delete_object(by_db(DbName), {DbName, {DDocId, Sig}})
end, DDocIds).
handle_db_event(DbName, created, St) ->
- gen_server:cast(?MODULE, {reset_indexes, DbName}),
+ gen_server:cast(St#st.index_server, {reset_indexes, DbName}),
{ok, St};
handle_db_event(DbName, deleted, St) ->
- gen_server:cast(?MODULE, {reset_indexes, DbName}),
+ gen_server:cast(St#st.index_server, {reset_indexes, DbName}),
{ok, St};
handle_db_event(<<"shards/", _/binary>> = DbName, {ddoc_updated,
DDocId}, St) ->
@@ -292,31 +303,70 @@ handle_db_event(<<"shards/", _/binary>> = DbName, {ddoc_updated,
lists:foreach(fun(DbShard) ->
lists:foreach(fun({_DbShard, {_DDocId, Sig}}) ->
% check if there are other ddocs with the same Sig for the same db
- SigDDocs = ets:match_object(?BY_DB, {DbShard, {'$1', Sig}}),
+ SigDDocs = ets:match_object(St#st.by_db, {DbShard, {'$1', Sig}}),
if length(SigDDocs) > 1 ->
- % remove records from ?BY_DB for this DDoc
+ % remove records from by_db for this DDoc
Args = [DbShard, DDocId, Sig],
- gen_server:cast(?MODULE, {rem_from_ets, Args});
+ gen_server:cast(St#st.index_server, {rem_from_ets, Args});
true ->
% single DDoc with this Sig - close couch_index processes
- case ets:lookup(?BY_SIG, {DbShard, Sig}) of
+ case ets:lookup(St#st.by_sig, {DbShard, Sig}) of
[{_, IndexPid}] -> (catch
gen_server:cast(IndexPid, {ddoc_updated, DDocResult}));
[] -> []
end
end
- end, ets:match_object(?BY_DB, {DbShard, {DDocId, '$1'}}))
+ end, ets:match_object(St#st.by_db, {DbShard, {DDocId, '$1'}}))
end, DbShards),
{ok, St};
handle_db_event(DbName, {ddoc_updated, DDocId}, St) ->
lists:foreach(fun({_DbName, {_DDocId, Sig}}) ->
- case ets:lookup(?BY_SIG, {DbName, Sig}) of
+ case ets:lookup(St#st.by_sig, {DbName, Sig}) of
[{_, IndexPid}] ->
(catch gen_server:cast(IndexPid, ddoc_updated));
[] ->
ok
end
- end, ets:match_object(?BY_DB, {DbName, {DDocId, '$1'}})),
+ end, ets:match_object(St#st.by_db, {DbName, {DDocId, '$1'}})),
{ok, St};
handle_db_event(_DbName, _Event, St) ->
{ok, St}.
+
+
+num_servers() ->
+ erlang:system_info(schedulers).
+
+couch_index_server(Bin) when is_binary(Bin) ->
+ couch_index_server(to_n(Bin));
+
+couch_index_server(N) when is_integer(N), N > 0 ->
+ name("index_server", N).
+
+
+by_sig(Bin) when is_binary(Bin) ->
+ by_sig(to_n(Bin));
+
+by_sig(N) when is_integer(N), N > 0 ->
+ name("couchdb_indexes_by_sig", N).
+
+
+by_pid(Bin) when is_binary(Bin) ->
+ by_pid(to_n(Bin));
+
+by_pid(N) ->
+ name("couchdb_indexes_by_pid", N).
+
+
+by_db(Bin) when is_binary(Bin) ->
+ by_db(to_n(Bin));
+
+by_db(N) ->
+ name("couchdb_indexes_by_db", N).
+
+
+to_n(Bin) when is_binary(Bin) ->
+ 1 + erlang:phash2(Bin, num_servers()).
+
+
+name(BaseName, N) when is_integer(N), N > 0 ->
+ list_to_atom(BaseName ++ "_" ++ integer_to_list(N)).
[couchdb] 01/02: [unrelated] improve erlang_ls.config
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch sharded_couch_index_server
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 67b4bfccc3fc62ddcd344f3f28915552d648d8b1
Author: Robert Newson <rn...@apache.org>
AuthorDate: Tue Sep 21 12:37:07 2021 +0100
[unrelated] improve erlang_ls.config
---
erlang_ls.config | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/erlang_ls.config b/erlang_ls.config
index 3d2fc9e..94483cf 100644
--- a/erlang_ls.config
+++ b/erlang_ls.config
@@ -1,2 +1,5 @@
+apps_dirs:
+ - "src/*"
include_dirs:
- - "src/"
+ - "src"
+ - "src/*/include"