You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by da...@apache.org on 2014/02/05 00:44:46 UTC
[33/44] couchdb commit: updated
refs/heads/1843-feature-bigcouch-multi-repo to 2385a37
Remove src/mem3
Project: http://git-wip-us.apache.org/repos/asf/couchdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb/commit/4cac46af
Tree: http://git-wip-us.apache.org/repos/asf/couchdb/tree/4cac46af
Diff: http://git-wip-us.apache.org/repos/asf/couchdb/diff/4cac46af
Branch: refs/heads/1843-feature-bigcouch-multi-repo
Commit: 4cac46af3dc40900b7ceb19c78db075bf30a3325
Parents: 3822d8f
Author: Paul J. Davis <pa...@gmail.com>
Authored: Tue Feb 4 17:41:40 2014 -0600
Committer: Paul J. Davis <pa...@gmail.com>
Committed: Tue Feb 4 17:41:40 2014 -0600
----------------------------------------------------------------------
src/mem3/README.md | 33 ---
src/mem3/include/mem3.hrl | 42 ----
src/mem3/src/mem3.app.src | 50 -----
src/mem3/src/mem3.erl | 240 ---------------------
src/mem3/src/mem3_app.erl | 21 --
src/mem3/src/mem3_httpd.erl | 51 -----
src/mem3/src/mem3_nodes.erl | 149 -------------
src/mem3/src/mem3_rep.erl | 223 --------------------
src/mem3/src/mem3_shards.erl | 329 -----------------------------
src/mem3/src/mem3_sup.erl | 34 ---
src/mem3/src/mem3_sync.erl | 344 -------------------------------
src/mem3/src/mem3_sync_event.erl | 85 --------
src/mem3/src/mem3_sync_nodes.erl | 114 ----------
src/mem3/src/mem3_sync_security.erl | 105 ----------
src/mem3/src/mem3_util.erl | 196 ------------------
src/mem3/test/01-config-default.ini | 2 -
src/mem3/test/mem3_util_test.erl | 152 --------------
17 files changed, 2170 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/README.md
----------------------------------------------------------------------
diff --git a/src/mem3/README.md b/src/mem3/README.md
deleted file mode 100644
index ba6e826..0000000
--- a/src/mem3/README.md
+++ /dev/null
@@ -1,33 +0,0 @@
-## mem3
-
-Mem3 is the node membership application for clustered [CouchDB][1]. It is used in [BigCouch][2] and tracks two very important things for the cluster:
-
- 1. member nodes
- 2. node/shards mappings for each database
-
-Both the nodes and shards are tracked in node-local couch databases. Shards are heavily used, so an ETS cache is also maintained for low-latency lookups. The nodes and shards are synchronized via continuous CouchDB replication, which serves as 'gossip' in Dynamo parlance. The shards ETS cache is kept in sync based on membership and database event listeners.
-
-A very important point to make here is that BigCouch does not necessarily divide up each database into equal shards across the nodes of a cluster. For instance, in a 20-node cluster, you may have the need to create a small database with very few documents. For efficiency reasons, you may create your database with Q=4 and keep the default of N=3. This means you only have 12 shards total, so 8 nodes will hold none of the data for this database. Given this feature, we even shard use out across the cluster by altering the 'start' node for the database's shards.
-
-Splitting and merging shards is an immature feature of the system, and will require attention in the near-term. We believe we can implement both functions and perform them while the database remains online.
-
-### Getting Started
-
-Mem3 requires R13B03 or higher and can be built with [rebar][6], which comes bundled in the repository. Rebar needs to be able to find the `couch_db.hrl` header file; one way to accomplish this is to set ERL_LIBS to point to the apps
-subdirectory of a bigcouch checkout, e.g.
-
- ERL_LIBS="/usr/local/src/bigcouch/apps" ./rebar compile
-
-### License
-[Apache 2.0][3]
-
-### Contact
- * [http://cloudant.com][4]
- * [info@cloudant.com][5]
-
-[1]: http://couchdb.apache.org
-[2]: http://github.com/cloudant/bigcouch
-[3]: http://www.apache.org/licenses/LICENSE-2.0.html
-[4]: http://cloudant.com
-[5]: mailto:info@cloudant.com
-[6]: http://github.com/basho/rebar
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/include/mem3.hrl
----------------------------------------------------------------------
diff --git a/src/mem3/include/mem3.hrl b/src/mem3/include/mem3.hrl
deleted file mode 100644
index cb39e78..0000000
--- a/src/mem3/include/mem3.hrl
+++ /dev/null
@@ -1,42 +0,0 @@
-% 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.
-
-% type specification hacked to suppress dialyzer warning re: match spec
--record(shard, {
- name :: binary() | '_',
- node :: node() | '_',
- dbname :: binary(),
- range :: [non_neg_integer() | '$1' | '$2'],
- ref :: reference() | 'undefined' | '_'
-}).
-
-%% types
--type join_type() :: init | join | replace | leave.
--type join_order() :: non_neg_integer().
--type options() :: list().
--type mem_node() :: {join_order(), node(), options()}.
--type mem_node_list() :: [mem_node()].
--type arg_options() :: {test, boolean()}.
--type args() :: [] | [arg_options()].
--type test() :: undefined | node().
--type epoch() :: float().
--type clock() :: {node(), epoch()}.
--type vector_clock() :: [clock()].
--type ping_node() :: node() | nil.
--type gossip_fun() :: call | cast.
-
--type part() :: #shard{}.
--type fullmap() :: [part()].
--type ref_part_map() :: {reference(), part()}.
--type tref() :: reference().
--type np() :: {node(), part()}.
--type beg_acc() :: [integer()].
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3.app.src
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3.app.src b/src/mem3/src/mem3.app.src
deleted file mode 100644
index 616e4bd..0000000
--- a/src/mem3/src/mem3.app.src
+++ /dev/null
@@ -1,50 +0,0 @@
-% 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.
-
-{application, mem3, [
- {description, "CouchDB Cluster Membership"},
- {vsn, git},
- {modules, [
- mem3,
- mem3_app,
- mem3_httpd,
- mem3_nodes,
- mem3_rep,
- mem3_shards,
- mem3_sup,
- mem3_sync,
- mem3_sync_event,
- mem3_sync_nodes,
- mem3_sync_security,
- mem3_util
- ]},
- {mod, {mem3_app, []}},
- {registered, [
- mem3_events,
- mem3_nodes,
- mem3_shards,
- mem3_sync,
- mem3_sync_nodes,
- mem3_sup
- ]},
- {applications, [
- kernel,
- stdlib,
- config,
- sasl,
- crypto,
- mochiweb,
- couch,
- rexi,
- twig
- ]}
-]}.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3.erl
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3.erl b/src/mem3/src/mem3.erl
deleted file mode 100644
index c9b4793..0000000
--- a/src/mem3/src/mem3.erl
+++ /dev/null
@@ -1,240 +0,0 @@
-% 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(mem3).
-
--export([start/0, stop/0, restart/0, nodes/0, node_info/2, shards/1, shards/2,
- choose_shards/2, n/1, dbname/1, ushards/1]).
--export([get_shard/3, local_shards/1, fold_shards/2]).
--export([sync_security/0, sync_security/1]).
--export([compare_nodelists/0, compare_shards/1]).
--export([quorum/1, group_by_proximity/1]).
--export([live_shards/2]).
-
--include_lib("mem3/include/mem3.hrl").
--include_lib("couch/include/couch_db.hrl").
-
-start() ->
- application:start(mem3).
-
-stop() ->
- application:stop(mem3).
-
-restart() ->
- stop(),
- start().
-
-%% @doc Detailed report of cluster-wide membership state. Queries the state
-%% on all member nodes and builds a dictionary with unique states as the
-%% key and the nodes holding that state as the value. Also reports member
-%% nodes which fail to respond and nodes which are connected but are not
-%% cluster members. Useful for debugging.
--spec compare_nodelists() -> [{{cluster_nodes, [node()]} | bad_nodes
- | non_member_nodes, [node()]}].
-compare_nodelists() ->
- Nodes = mem3:nodes(),
- AllNodes = erlang:nodes([this, visible]),
- {Replies, BadNodes} = gen_server:multi_call(Nodes, mem3_nodes, get_nodelist),
- Dict = lists:foldl(fun({Node, Nodelist}, D) ->
- orddict:append({cluster_nodes, Nodelist}, Node, D)
- end, orddict:new(), Replies),
- [{non_member_nodes, AllNodes -- Nodes}, {bad_nodes, BadNodes} | Dict].
-
--spec compare_shards(DbName::iodata()) -> [{bad_nodes | [#shard{}], [node()]}].
-compare_shards(DbName) when is_list(DbName) ->
- compare_shards(list_to_binary(DbName));
-compare_shards(DbName) ->
- Nodes = mem3:nodes(),
- {Replies, BadNodes} = rpc:multicall(mem3, shards, [DbName]),
- GoodNodes = [N || N <- Nodes, not lists:member(N, BadNodes)],
- Dict = lists:foldl(fun({Shards, Node}, D) ->
- orddict:append(Shards, Node, D)
- end, orddict:new(), lists:zip(Replies, GoodNodes)),
- [{bad_nodes, BadNodes} | Dict].
-
--spec n(DbName::iodata()) -> integer().
-n(DbName) ->
- length(mem3:shards(DbName, <<"foo">>)).
-
--spec nodes() -> [node()].
-nodes() ->
- mem3_nodes:get_nodelist().
-
-node_info(Node, Key) ->
- mem3_nodes:get_node_info(Node, Key).
-
--spec shards(DbName::iodata()) -> [#shard{}].
-shards(DbName) when is_list(DbName) ->
- shards(list_to_binary(DbName));
-shards(DbName) ->
- ShardDbName =
- list_to_binary(config:get("mem3", "shard_db", "dbs")),
- case DbName of
- ShardDbName ->
- %% shard_db is treated as a single sharded db to support calls to db_info
- %% and view_all_docs
- [#shard{
- node = node(),
- name = ShardDbName,
- dbname = ShardDbName,
- range = [0, 2 bsl 31]}];
- _ ->
- mem3_shards:for_db(DbName)
- end.
-
--spec shards(DbName::iodata(), DocId::binary()) -> [#shard{}].
-shards(DbName, DocId) when is_list(DbName) ->
- shards(list_to_binary(DbName), DocId);
-shards(DbName, DocId) when is_list(DocId) ->
- shards(DbName, list_to_binary(DocId));
-shards(DbName, DocId) ->
- mem3_shards:for_docid(DbName, DocId).
-
--spec ushards(DbName::iodata()) -> [#shard{}].
-ushards(DbName) ->
- Nodes = [node()|erlang:nodes()],
- ZoneMap = zone_map(Nodes),
- ushards(DbName, live_shards(DbName, Nodes), ZoneMap).
-
-ushards(DbName, Shards0, ZoneMap) ->
- {L,S,D} = group_by_proximity(Shards0, ZoneMap),
- % Prefer shards in the local zone over shards in a different zone,
- % but sort each zone separately to ensure a consistent choice between
- % nodes in the same zone.
- Shards = choose_ushards(DbName, L ++ S) ++ choose_ushards(DbName, D),
- lists:ukeysort(#shard.range, Shards).
-
-get_shard(DbName, Node, Range) ->
- mem3_shards:get(DbName, Node, Range).
-
-local_shards(DbName) ->
- mem3_shards:local(DbName).
-
-fold_shards(Fun, Acc) ->
- mem3_shards:fold(Fun, Acc).
-
-sync_security() ->
- mem3_sync_security:go().
-
-sync_security(Db) ->
- mem3_sync_security:go(dbname(Db)).
-
--spec choose_shards(DbName::iodata(), Options::list()) -> [#shard{}].
-choose_shards(DbName, Options) when is_list(DbName) ->
- choose_shards(list_to_binary(DbName), Options);
-choose_shards(DbName, Options) ->
- try shards(DbName)
- catch error:E when E==database_does_not_exist; E==badarg ->
- Nodes = mem3:nodes(),
- case get_placement(Options) of
- undefined ->
- choose_shards(DbName, Nodes, Options);
- Placement ->
- lists:flatmap(fun({Zone, N}) ->
- NodesInZone = nodes_in_zone(Nodes, Zone),
- Options1 = lists:keymerge(1, [{n,N}], Options),
- choose_shards(DbName, NodesInZone, Options1)
- end, Placement)
- end
- end.
-
-choose_shards(DbName, Nodes, Options) ->
- NodeCount = length(Nodes),
- Suffix = couch_util:get_value(shard_suffix, Options, ""),
- N = mem3_util:n_val(couch_util:get_value(n, Options), NodeCount),
- if N =:= 0 -> erlang:error(no_nodes_in_zone);
- true -> ok
- end,
- Q = mem3_util:to_integer(couch_util:get_value(q, Options,
- config:get("cluster", "q", "8"))),
- %% rotate to a random entry in the nodelist for even distribution
- {A, B} = lists:split(crypto:rand_uniform(1,length(Nodes)+1), Nodes),
- RotatedNodes = B ++ A,
- mem3_util:create_partition_map(DbName, N, Q, RotatedNodes, Suffix).
-
-get_placement(Options) ->
- case couch_util:get_value(placement, Options) of
- undefined ->
- case config:get("cluster", "placement") of
- undefined ->
- undefined;
- PlacementStr ->
- decode_placement_string(PlacementStr)
- end;
- PlacementStr ->
- decode_placement_string(PlacementStr)
- end.
-
-decode_placement_string(PlacementStr) ->
- [begin
- [Zone, N] = string:tokens(Rule, ":"),
- {list_to_binary(Zone), list_to_integer(N)}
- end || Rule <- string:tokens(PlacementStr, ",")].
-
--spec dbname(#shard{} | iodata()) -> binary().
-dbname(#shard{dbname = DbName}) ->
- DbName;
-dbname(<<"shards/", _:8/binary, "-", _:8/binary, "/", DbName/binary>>) ->
- list_to_binary(filename:rootname(binary_to_list(DbName)));
-dbname(DbName) when is_list(DbName) ->
- dbname(list_to_binary(DbName));
-dbname(DbName) when is_binary(DbName) ->
- DbName;
-dbname(_) ->
- erlang:error(badarg).
-
-nodes_in_zone(Nodes, Zone) ->
- [Node || Node <- Nodes, Zone == mem3:node_info(Node, <<"zone">>)].
-
-live_shards(DbName, Nodes) ->
- [S || #shard{node=Node} = S <- shards(DbName), lists:member(Node, Nodes)].
-
-zone_map(Nodes) ->
- [{Node, node_info(Node, <<"zone">>)} || Node <- Nodes].
-
-group_by_proximity(Shards) ->
- Nodes = [N || #shard{node=N} <- lists:ukeysort(#shard.node, Shards)],
- group_by_proximity(Shards, zone_map(Nodes)).
-
-group_by_proximity(Shards, ZoneMap) ->
- {Local, Remote} = lists:partition(fun(S) -> S#shard.node =:= node() end,
- Shards),
- LocalZone = proplists:get_value(node(), ZoneMap),
- Fun = fun(S) -> proplists:get_value(S#shard.node, ZoneMap) =:= LocalZone end,
- {SameZone, DifferentZone} = lists:partition(Fun, Remote),
- {Local, SameZone, DifferentZone}.
-
-choose_ushards(DbName, Shards) ->
- Groups = group_by_range(rotate_list(DbName, lists:sort(Shards))),
- Fun = fun(Group, {N, Acc}) ->
- {N+1, [lists:nth(1 + N rem length(Group), Group) | Acc]} end,
- {_, Result} = lists:foldl(Fun, {0, []}, Groups),
- Result.
-
-rotate_list(_DbName, []) ->
- [];
-rotate_list(DbName, List) ->
- {H, T} = lists:split(erlang:crc32(DbName) rem length(List), List),
- T ++ H.
-
-group_by_range(Shards) ->
- Groups0 = lists:foldl(fun(#shard{range=Range}=Shard, Dict) ->
- orddict:append(Range, Shard, Dict) end, orddict:new(), Shards),
- {_, Groups} = lists:unzip(Groups0),
- Groups.
-
-% quorum functions
-
-quorum(#db{name=DbName}) ->
- quorum(DbName);
-quorum(DbName) ->
- n(DbName) div 2 + 1.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3_app.erl
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3_app.erl b/src/mem3/src/mem3_app.erl
deleted file mode 100644
index 3ddfbe6..0000000
--- a/src/mem3/src/mem3_app.erl
+++ /dev/null
@@ -1,21 +0,0 @@
-% 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(mem3_app).
--behaviour(application).
--export([start/2, stop/1]).
-
-start(_Type, []) ->
- mem3_sup:start_link().
-
-stop([]) ->
- ok.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3_httpd.erl
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3_httpd.erl b/src/mem3/src/mem3_httpd.erl
deleted file mode 100644
index 94196fa..0000000
--- a/src/mem3/src/mem3_httpd.erl
+++ /dev/null
@@ -1,51 +0,0 @@
-% 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(mem3_httpd).
-
--export([handle_membership_req/1]).
-
-%% includes
--include_lib("mem3/include/mem3.hrl").
--include_lib("couch/include/couch_db.hrl").
-
-
-handle_membership_req(#httpd{method='GET',
- path_parts=[<<"_membership">>]} = Req) ->
- ClusterNodes = try mem3:nodes()
- catch _:_ -> {ok,[]} end,
- couch_httpd:send_json(Req, {[
- {all_nodes, lists:sort([node()|nodes()])},
- {cluster_nodes, lists:sort(ClusterNodes)}
- ]});
-handle_membership_req(#httpd{method='GET',
- path_parts=[<<"_membership">>, <<"parts">>, DbName]} = Req) ->
- ClusterNodes = try mem3:nodes()
- catch _:_ -> {ok,[]} end,
- Shards = mem3:shards(DbName),
- JsonShards = json_shards(Shards, dict:new()),
- couch_httpd:send_json(Req, {[
- {all_nodes, lists:sort([node()|nodes()])},
- {cluster_nodes, lists:sort(ClusterNodes)},
- {partitions, JsonShards}
- ]}).
-
-%%
-%% internal
-%%
-
-json_shards([], AccIn) ->
- List = dict:to_list(AccIn),
- {lists:sort(List)};
-json_shards([#shard{node=Node, range=[B,_E]} | Rest], AccIn) ->
- HexBeg = couch_util:to_hex(<<B:32/integer>>),
- json_shards(Rest, dict:append(HexBeg, Node, AccIn)).
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3_nodes.erl
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3_nodes.erl b/src/mem3/src/mem3_nodes.erl
deleted file mode 100644
index 782a8b5..0000000
--- a/src/mem3/src/mem3_nodes.erl
+++ /dev/null
@@ -1,149 +0,0 @@
-% 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(mem3_nodes).
--behaviour(gen_server).
--export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
- code_change/3]).
-
--export([start_link/0, get_nodelist/0, get_node_info/2]).
-
--include_lib("mem3/include/mem3.hrl").
--include_lib("couch/include/couch_db.hrl").
-
--record(state, {changes_pid, update_seq}).
-
-start_link() ->
- gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
-
-get_nodelist() ->
- try
- lists:sort([N || {N,_} <- ets:tab2list(?MODULE)])
- catch error:badarg ->
- gen_server:call(?MODULE, get_nodelist)
- end.
-
-get_node_info(Node, Key) ->
- try
- couch_util:get_value(Key, ets:lookup_element(?MODULE, Node, 2))
- catch error:badarg ->
- gen_server:call(?MODULE, {get_node_info, Node, Key})
- end.
-
-init([]) ->
- ets:new(?MODULE, [named_table, {read_concurrency, true}]),
- UpdateSeq = initialize_nodelist(),
- {Pid, _} = spawn_monitor(fun() -> listen_for_changes(UpdateSeq) end),
- {ok, #state{changes_pid = Pid, update_seq = UpdateSeq}}.
-
-handle_call(get_nodelist, _From, State) ->
- {reply, lists:sort([N || {N,_} <- ets:tab2list(?MODULE)]), State};
-handle_call({get_node_info, Node, Key}, _From, State) ->
- Resp = try
- couch_util:get_value(Key, ets:lookup_element(?MODULE, Node, 2))
- catch error:badarg ->
- error
- end,
- {reply, Resp, State};
-handle_call({add_node, Node, NodeInfo}, _From, State) ->
- gen_event:notify(mem3_events, {add_node, Node}),
- ets:insert(?MODULE, {Node, NodeInfo}),
- {reply, ok, State};
-handle_call({remove_node, Node}, _From, State) ->
- gen_event:notify(mem3_events, {remove_node, Node}),
- ets:delete(?MODULE, Node),
- {reply, ok, State};
-handle_call(_Call, _From, State) ->
- {noreply, State}.
-
-handle_cast(_Msg, State) ->
- {noreply, State}.
-
-handle_info({'DOWN', _, _, Pid, Reason}, #state{changes_pid=Pid} = State) ->
- twig:log(notice, "~p changes listener died ~p", [?MODULE, Reason]),
- StartSeq = State#state.update_seq,
- Seq = case Reason of {seq, EndSeq} -> EndSeq; _ -> StartSeq end,
- erlang:send_after(5000, self(), start_listener),
- {noreply, State#state{update_seq = Seq}};
-handle_info(start_listener, #state{update_seq = Seq} = State) ->
- {NewPid, _} = spawn_monitor(fun() -> listen_for_changes(Seq) end),
- {noreply, State#state{changes_pid=NewPid}};
-handle_info(_Info, State) ->
- {noreply, State}.
-
-terminate(_Reason, _State) ->
- ok.
-
-code_change(_OldVsn, {state, ChangesPid, UpdateSeq, _}, _Extra) ->
- ets:new(?MODULE, [named_table, {read_concurrency, true}]),
- initialize_nodelist(),
- {ok, #state{changes_pid = ChangesPid, update_seq = UpdateSeq}};
-code_change(_OldVsn, State, _Extra) ->
- {ok, State}.
-
-%% internal functions
-
-initialize_nodelist() ->
- DbName = config:get("mem3", "node_db", "nodes"),
- {ok, Db} = mem3_util:ensure_exists(DbName),
- {ok, _, Db} = couch_btree:fold(Db#db.id_tree, fun first_fold/3, Db, []),
- % add self if not already present
- case ets:lookup(?MODULE, node()) of
- [_] ->
- ok;
- [] ->
- ets:insert(?MODULE, {node(), []}),
- Doc = #doc{id = couch_util:to_binary(node())},
- {ok, _} = couch_db:update_doc(Db, Doc, [])
- end,
- couch_db:close(Db),
- Db#db.update_seq.
-
-first_fold(#full_doc_info{id = <<"_design/", _/binary>>}, _, Acc) ->
- {ok, Acc};
-first_fold(#full_doc_info{deleted=true}, _, Acc) ->
- {ok, Acc};
-first_fold(#full_doc_info{id=Id}=DocInfo, _, Db) ->
- {ok, #doc{body={Props}}} = couch_db:open_doc(Db, DocInfo, [ejson_body]),
- ets:insert(?MODULE, {mem3_util:to_atom(Id), Props}),
- {ok, Db}.
-
-listen_for_changes(Since) ->
- DbName = config:get("mem3", "node_db", "nodes"),
- {ok, Db} = mem3_util:ensure_exists(DbName),
- Args = #changes_args{
- feed = "continuous",
- since = Since,
- heartbeat = true,
- include_docs = true
- },
- ChangesFun = couch_changes:handle_changes(Args, nil, Db),
- ChangesFun(fun changes_callback/2).
-
-changes_callback(start, _) ->
- {ok, nil};
-changes_callback({stop, EndSeq}, _) ->
- exit({seq, EndSeq});
-changes_callback({change, {Change}, _}, _) ->
- Node = couch_util:get_value(<<"id">>, Change),
- case Node of <<"_design/", _/binary>> -> ok; _ ->
- case mem3_util:is_deleted(Change) of
- false ->
- {Props} = couch_util:get_value(doc, Change),
- gen_server:call(?MODULE, {add_node, mem3_util:to_atom(Node), Props});
- true ->
- gen_server:call(?MODULE, {remove_node, mem3_util:to_atom(Node)})
- end
- end,
- {ok, couch_util:get_value(<<"seq">>, Change)};
-changes_callback(timeout, _) ->
- {ok, nil}.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3_rep.erl
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3_rep.erl b/src/mem3/src/mem3_rep.erl
deleted file mode 100644
index 373bc3f..0000000
--- a/src/mem3/src/mem3_rep.erl
+++ /dev/null
@@ -1,223 +0,0 @@
-% 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(mem3_rep).
-
--export([go/2, go/3, changes_enumerator/3, make_local_id/2]).
-
--include_lib("mem3/include/mem3.hrl").
--include_lib("couch/include/couch_db.hrl").
-
--define(CTX, #user_ctx{roles = [<<"_admin">>]}).
-
--record(acc, {
- batch_size,
- batch_count,
- revcount = 0,
- infos = [],
- seq,
- localid,
- source,
- target,
- filter
-}).
-
-go(Source, Target) ->
- go(Source, Target, []).
-
-go(DbName, Node, Opts) when is_binary(DbName), is_atom(Node) ->
- go(#shard{name=DbName, node=node()}, #shard{name=DbName, node=Node}, Opts);
-
-go(#shard{} = Source, #shard{} = Target, Opts) ->
- mem3_sync_security:maybe_sync(Source, Target),
- BatchSize = case proplists:get_value(batch_size, Opts) of
- BS when is_integer(BS), BS > 0 -> BS;
- _ -> 100
- end,
- BatchCount = case proplists:get_value(batch_count, Opts) of
- all -> all;
- BC when is_integer(BC), BC > 0 -> BC;
- _ -> 1
- end,
- Filter = proplists:get_value(filter, Opts),
- LocalId = make_local_id(Source, Target, Filter),
- Acc = #acc{
- batch_size = BatchSize,
- batch_count = BatchCount,
- localid = LocalId,
- source = Source,
- target = Target,
- filter = Filter
- },
- go(Acc).
-
-go(#acc{source=Source, batch_count=BC}=Acc) ->
- case couch_db:open(Source#shard.name, [{user_ctx,?CTX}]) of
- {ok, Db} ->
- Resp = try
- repl(Db, Acc)
- catch error:{not_found, no_db_file} ->
- {error, missing_target}
- after
- couch_db:close(Db)
- end,
- case Resp of
- {ok, P} when P > 0, BC == all ->
- go(Acc);
- {ok, P} when P > 0, BC > 1 ->
- go(Acc#acc{batch_count=BC-1});
- Else ->
- Else
- end;
- {not_found, no_db_file} ->
- {error, missing_source}
- end.
-
-repl(#db{name=DbName, seq_tree=Bt}=Db, #acc{localid=LocalId}=Acc0) ->
- erlang:put(io_priority, {internal_repl, DbName}),
- Seq = calculate_start_seq(Db, Acc0#acc.target, LocalId),
- Acc1 = Acc0#acc{source=Db, seq=Seq},
- Fun = fun ?MODULE:changes_enumerator/3,
- {ok, _, Acc2} = couch_btree:fold(Bt, Fun, Acc1, [{start_key, Seq + 1}]),
- {ok, #acc{seq = LastSeq}} = replicate_batch(Acc2),
- {ok, couch_db:count_changes_since(Db, LastSeq)}.
-
-make_local_id(#shard{}=Source, #shard{}=Target) ->
- make_local_id(Source, Target, undefined).
-
-make_local_id(#shard{node=SourceNode}, #shard{node=TargetNode}, Filter) ->
- S = couch_util:encodeBase64Url(couch_util:md5(term_to_binary(SourceNode))),
- T = couch_util:encodeBase64Url(couch_util:md5(term_to_binary(TargetNode))),
- F = case is_function(Filter) of
- true ->
- {new_uniq, Hash} = erlang:fun_info(Filter, new_uniq),
- B = couch_util:encodeBase64Url(Hash),
- <<"-", B/binary>>;
- false ->
- <<>>
- end,
- <<"_local/shard-sync-", S/binary, "-", T/binary, F/binary>>.
-
-changes_enumerator(FDI, _, #acc{revcount=C, infos=Infos}=Acc0) ->
- #doc_info{
- high_seq=Seq,
- revs=Revs
- } = couch_doc:to_doc_info(FDI),
- {Count, NewInfos} = case filter_doc(Acc0#acc.filter, FDI) of
- keep -> {C + length(Revs), [FDI | Infos]};
- discard -> {C, Infos}
- end,
- Acc1 = Acc0#acc{
- seq=Seq,
- revcount=Count,
- infos=NewInfos
- },
- Go = if Count < Acc1#acc.batch_size -> ok; true -> stop end,
- {Go, Acc1}.
-
-filter_doc(Filter, FullDocInfo) when is_function(Filter) ->
- try Filter(FullDocInfo) of
- discard -> discard;
- _ -> keep
- catch _:_ ->
- keep
- end;
-filter_doc(_, _) ->
- keep.
-
-replicate_batch(#acc{target = #shard{node=Node, name=Name}} = Acc) ->
- case find_missing_revs(Acc) of
- [] ->
- ok;
- Missing ->
- ok = save_on_target(Node, Name, open_docs(Acc, Missing))
- end,
- update_locals(Acc),
- {ok, Acc#acc{revcount=0, infos=[]}}.
-
-find_missing_revs(Acc) ->
- #acc{target = #shard{node=Node, name=Name}, infos = Infos} = Acc,
- IdsRevs = lists:map(fun(FDI) ->
- #doc_info{id=Id, revs=RevInfos} = couch_doc:to_doc_info(FDI),
- {Id, [R || #rev_info{rev=R} <- RevInfos]}
- end, Infos),
- Options = [{io_priority, {internal_repl, Name}}, {user_ctx, ?CTX}],
- rexi_call(Node, {fabric_rpc, get_missing_revs, [Name, IdsRevs, Options]}).
-
-open_docs(#acc{source=Source, infos=Infos}, Missing) ->
- lists:flatmap(fun({Id, Revs, _}) ->
- FDI = lists:keyfind(Id, #full_doc_info.id, Infos),
- open_doc_revs(Source, FDI, Revs)
- end, Missing).
-
-save_on_target(Node, Name, Docs) ->
- Options = [replicated_changes, full_commit, {user_ctx, ?CTX},
- {io_priority, {internal_repl, Name}}],
- rexi_call(Node, {fabric_rpc, update_docs, [Name, Docs, Options]}),
- ok.
-
-update_locals(Acc) ->
- #acc{seq=Seq, source=Db, target=Target, localid=Id} = Acc,
- #shard{name=Name, node=Node} = Target,
- Doc = #doc{id = Id, body = {[
- {<<"seq">>, Seq},
- {<<"node">>, list_to_binary(atom_to_list(Node))},
- {<<"timestamp">>, list_to_binary(iso8601_timestamp())}
- ]}},
- {ok, _} = couch_db:update_doc(Db, Doc, []),
- Options = [{user_ctx, ?CTX}, {io_priority, {internal_repl, Name}}],
- rexi_call(Node, {fabric_rpc, update_docs, [Name, [Doc], Options]}).
-
-rexi_call(Node, MFA) ->
- Mon = rexi_monitor:start([{rexi_server, Node}]),
- Ref = rexi:cast(Node, self(), MFA, [sync]),
- try
- receive {Ref, {ok, Reply}} ->
- Reply;
- {Ref, Error} ->
- erlang:error(Error);
- {rexi_DOWN, Mon, _, Reason} ->
- erlang:error({rexi_DOWN, {Node, Reason}})
- after 600000 ->
- erlang:error(timeout)
- end
- after
- rexi_monitor:stop(Mon)
- end.
-
-calculate_start_seq(Db, #shard{node=Node, name=Name}, LocalId) ->
- case couch_db:open_doc(Db, LocalId, [ejson_body]) of
- {ok, #doc{body = {SProps}}} ->
- Opts = [{user_ctx, ?CTX}, {io_priority, {internal_repl, Name}}],
- try rexi_call(Node, {fabric_rpc, open_doc, [Name, LocalId, Opts]}) of
- #doc{body = {TProps}} ->
- SourceSeq = couch_util:get_value(<<"seq">>, SProps, 0),
- TargetSeq = couch_util:get_value(<<"seq">>, TProps, 0),
- erlang:min(SourceSeq, TargetSeq)
- catch error:{not_found, _} ->
- 0
- end;
- {not_found, _} ->
- 0
- end.
-
-open_doc_revs(Db, #full_doc_info{id=Id, rev_tree=RevTree}, Revs) ->
- {FoundRevs, _} = couch_key_tree:get_key_leafs(RevTree, Revs),
- lists:map(fun({#leaf{deleted=IsDel, ptr=SummaryPtr}, FoundRevPath}) ->
- couch_db:make_doc(Db, Id, IsDel, SummaryPtr, FoundRevPath)
- end, FoundRevs).
-
-iso8601_timestamp() ->
- {_,_,Micro} = Now = os:timestamp(),
- {{Year,Month,Date},{Hour,Minute,Second}} = calendar:now_to_datetime(Now),
- Format = "~4.10.0B-~2.10.0B-~2.10.0BT~2.10.0B:~2.10.0B:~2.10.0B.~6.10.0BZ",
- io_lib:format(Format, [Year, Month, Date, Hour, Minute, Second, Micro]).
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3_shards.erl
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3_shards.erl b/src/mem3/src/mem3_shards.erl
deleted file mode 100644
index 9949869..0000000
--- a/src/mem3/src/mem3_shards.erl
+++ /dev/null
@@ -1,329 +0,0 @@
-% 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(mem3_shards).
--behaviour(gen_server).
--behaviour(config_listener).
-
--export([init/1, terminate/2, code_change/3]).
--export([handle_call/3, handle_cast/2, handle_info/2]).
--export([handle_config_change/5]).
-
--export([start_link/0]).
--export([for_db/1, for_docid/2, get/3, local/1, fold/2]).
--export([set_max_size/1]).
-
--record(st, {
- max_size = 25000,
- cur_size = 0,
- changes_pid
-}).
-
--include_lib("mem3/include/mem3.hrl").
--include_lib("couch/include/couch_db.hrl").
-
--define(DBS, mem3_dbs).
--define(SHARDS, mem3_shards).
--define(ATIMES, mem3_atimes).
-
-start_link() ->
- gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
-
-for_db(DbName) ->
- try ets:lookup(?SHARDS, DbName) of
- [] ->
- load_shards_from_disk(DbName);
- Else ->
- gen_server:cast(?MODULE, {cache_hit, DbName}),
- Else
- catch error:badarg ->
- load_shards_from_disk(DbName)
- end.
-
-for_docid(DbName, DocId) ->
- HashKey = mem3_util:hash(DocId),
- Head = #shard{
- name = '_',
- node = '_',
- dbname = DbName,
- range = ['$1','$2'],
- ref = '_'
- },
- Conditions = [{'=<', '$1', HashKey}, {'=<', HashKey, '$2'}],
- try ets:select(?SHARDS, [{Head, Conditions, ['$_']}]) of
- [] ->
- load_shards_from_disk(DbName, DocId);
- Shards ->
- gen_server:cast(?MODULE, {cache_hit, DbName}),
- Shards
- catch error:badarg ->
- load_shards_from_disk(DbName, DocId)
- end.
-
-get(DbName, Node, Range) ->
- Res = lists:foldl(fun(#shard{node=N, range=R}=S, Acc) ->
- case {N, R} of
- {Node, Range} -> [S | Acc];
- _ -> Acc
- end
- end, [], for_db(DbName)),
- case Res of
- [] -> {error, not_found};
- [Shard] -> {ok, Shard};
- [_|_] -> {error, duplicates}
- end.
-
-local(DbName) ->
- Pred = fun(#shard{node=Node}) when Node == node() -> true; (_) -> false end,
- lists:filter(Pred, for_db(DbName)).
-
-fold(Fun, Acc) ->
- DbName = config:get("mem3", "shards_db", "dbs"),
- {ok, Db} = mem3_util:ensure_exists(DbName),
- FAcc = {Db, Fun, Acc},
- try
- {ok, _, LastAcc} = couch_db:enum_docs(Db, fun fold_fun/3, FAcc, []),
- {_Db, _UFun, UAcc} = LastAcc,
- UAcc
- after
- couch_db:close(Db)
- end.
-
-set_max_size(Size) when is_integer(Size), Size > 0 ->
- gen_server:call(?MODULE, {set_max_size, Size}).
-
-handle_config_change("mem3", "shard_cache_size", SizeList, _, _) ->
- Size = list_to_integer(SizeList),
- {ok, gen_server:call(?MODULE, {set_max_size, Size}, infinity)};
-handle_config_change("mem3", "shard_db", _DbName, _, _) ->
- {ok, gen_server:call(?MODULE, shard_db_changed, infinity)};
-handle_config_change(_, _, _, _, _) ->
- {ok, nil}.
-
-init([]) ->
- ets:new(?SHARDS, [bag, protected, named_table, {keypos,#shard.dbname}]),
- ets:new(?DBS, [set, protected, named_table]),
- ets:new(?ATIMES, [ordered_set, protected, named_table]),
- ok = config:listen_for_changes(?MODULE, nil),
- SizeList = config:get("mem3", "shard_cache_size", "25000"),
- {Pid, _} = spawn_monitor(fun() -> listen_for_changes(get_update_seq()) end),
- {ok, #st{
- max_size = list_to_integer(SizeList),
- cur_size = 0,
- changes_pid = Pid
- }}.
-
-handle_call({set_max_size, Size}, _From, St) ->
- {reply, ok, cache_free(St#st{max_size=Size})};
-handle_call(shard_db_changed, _From, St) ->
- exit(St#st.changes_pid, shard_db_changed),
- {reply, ok, St};
-handle_call(_Call, _From, St) ->
- {noreply, St}.
-
-handle_cast({cache_hit, DbName}, St) ->
- cache_hit(DbName),
- {noreply, St};
-handle_cast({cache_insert, DbName, Shards}, St) ->
- {noreply, cache_free(cache_insert(St, DbName, Shards))};
-handle_cast({cache_remove, DbName}, St) ->
- {noreply, cache_remove(St, DbName)};
-handle_cast(_Msg, St) ->
- {noreply, St}.
-
-handle_info({'DOWN', _, _, Pid, Reason}, #st{changes_pid=Pid}=St) ->
- {NewSt, Seq} = case Reason of
- {seq, EndSeq} ->
- {St, EndSeq};
- shard_db_changed ->
- {cache_clear(St), get_update_seq()};
- _ ->
- twig:log(notice, "~p changes listener died ~p", [?MODULE, Reason]),
- {St, get_update_seq()}
- end,
- erlang:send_after(5000, self(), {start_listener, Seq}),
- {noreply, NewSt#st{changes_pid=undefined}};
-handle_info({start_listener, Seq}, St) ->
- {NewPid, _} = spawn_monitor(fun() -> listen_for_changes(Seq) end),
- {noreply, St#st{changes_pid=NewPid}};
-handle_info({gen_event_EXIT, {config_listener, ?MODULE}, _Reason}, State) ->
- erlang:send_after(5000, self(), restart_config_listener),
- {noreply, State};
-handle_info(restart_config_listener, State) ->
- ok = config:listen_for_changes(?MODULE, nil),
- {noreply, State};
-handle_info(_Msg, St) ->
- {noreply, St}.
-
-terminate(_Reason, #st{changes_pid=Pid}) ->
- exit(Pid, kill),
- ok.
-
-code_change(_OldVsn, St, _Extra) ->
- {ok, St}.
-
-%% internal functions
-
-fold_fun(#full_doc_info{}=FDI, _, Acc) ->
- DI = couch_doc:to_doc_info(FDI),
- fold_fun(DI, nil, Acc);
-fold_fun(#doc_info{}=DI, _, {Db, UFun, UAcc}) ->
- case couch_db:open_doc(Db, DI, [ejson_body, conflicts]) of
- {ok, Doc} ->
- {Props} = Doc#doc.body,
- Shards = mem3_util:build_shards(Doc#doc.id, Props),
- NewUAcc = lists:foldl(UFun, UAcc, Shards),
- {ok, {Db, UFun, NewUAcc}};
- _ ->
- {ok, {Db, UFun, UAcc}}
- end.
-
-get_update_seq() ->
- DbName = config:get("mem3", "shards_db", "dbs"),
- {ok, Db} = mem3_util:ensure_exists(DbName),
- couch_db:close(Db),
- Db#db.update_seq.
-
-listen_for_changes(Since) ->
- DbName = config:get("mem3", "shards_db", "dbs"),
- {ok, Db} = mem3_util:ensure_exists(DbName),
- Args = #changes_args{
- feed = "continuous",
- since = Since,
- heartbeat = true,
- include_docs = true
- },
- ChangesFun = couch_changes:handle_changes(Args, Since, Db),
- ChangesFun(fun changes_callback/2).
-
-changes_callback(start, Acc) ->
- {ok, Acc};
-changes_callback({stop, EndSeq}, _) ->
- exit({seq, EndSeq});
-changes_callback({change, {Change}, _}, _) ->
- DbName = couch_util:get_value(<<"id">>, Change),
- case DbName of <<"_design/", _/binary>> -> ok; _Else ->
- case mem3_util:is_deleted(Change) of
- true ->
- gen_server:cast(?MODULE, {cache_remove, DbName});
- false ->
- case couch_util:get_value(doc, Change) of
- {error, Reason} ->
- twig:log(error, "missing partition table for ~s: ~p",
- [DbName, Reason]);
- {Doc} ->
- Shards = mem3_util:build_shards(DbName, Doc),
- gen_server:cast(?MODULE, {cache_insert, DbName, Shards}),
- [create_if_missing(Name) || #shard{name=Name, node=Node}
- <- Shards, Node =:= node()]
- end
- end
- end,
- {ok, couch_util:get_value(<<"seq">>, Change)};
-changes_callback(timeout, _) ->
- ok.
-
-load_shards_from_disk(DbName) when is_binary(DbName) ->
- X = ?l2b(config:get("mem3", "shard_db", "dbs")),
- {ok, Db} = mem3_util:ensure_exists(X),
- try
- load_shards_from_db(Db, DbName)
- after
- couch_db:close(Db)
- end.
-
-load_shards_from_db(#db{} = ShardDb, DbName) ->
- case couch_db:open_doc(ShardDb, DbName, [ejson_body]) of
- {ok, #doc{body = {Props}}} ->
- Shards = mem3_util:build_shards(DbName, Props),
- gen_server:cast(?MODULE, {cache_insert, DbName, Shards}),
- Shards;
- {not_found, _} ->
- erlang:error(database_does_not_exist, ?b2l(DbName))
- end.
-
-load_shards_from_disk(DbName, DocId)->
- Shards = load_shards_from_disk(DbName),
- HashKey = mem3_util:hash(DocId),
- [S || #shard{range = [B,E]} = S <- Shards, B =< HashKey, HashKey =< E].
-
-create_if_missing(Name) ->
- DbDir = config:get("couchdb", "database_dir"),
- Filename = filename:join(DbDir, ?b2l(Name) ++ ".couch"),
- case filelib:is_regular(Filename) of
- true ->
- ok;
- false ->
- Options = [{user_ctx, #user_ctx{roles=[<<"_admin">>]}}],
- case couch_server:create(Name, Options) of
- {ok, Db} ->
- couch_db:close(Db);
- Error ->
- twig:log(error, "~p tried to create ~s, got ~p",
- [?MODULE, Name, Error])
- end
- end.
-
-cache_insert(#st{cur_size=Cur}=St, DbName, Shards) ->
- NewATime = now(),
- true = ets:delete(?SHARDS, DbName),
- true = ets:insert(?SHARDS, Shards),
- case ets:lookup(?DBS, DbName) of
- [{DbName, ATime}] ->
- true = ets:delete(?ATIMES, ATime),
- true = ets:insert(?ATIMES, {NewATime, DbName}),
- true = ets:insert(?DBS, {DbName, NewATime}),
- St;
- [] ->
- true = ets:insert(?ATIMES, {NewATime, DbName}),
- true = ets:insert(?DBS, {DbName, NewATime}),
- St#st{cur_size=Cur + 1}
- end.
-
-cache_remove(#st{cur_size=Cur}=St, DbName) ->
- true = ets:delete(?SHARDS, DbName),
- case ets:lookup(?DBS, DbName) of
- [{DbName, ATime}] ->
- true = ets:delete(?DBS, DbName),
- true = ets:delete(?ATIMES, ATime),
- St#st{cur_size=Cur-1};
- [] ->
- St
- end.
-
-cache_hit(DbName) ->
- case ets:lookup(?DBS, DbName) of
- [{DbName, ATime}] ->
- NewATime = now(),
- true = ets:delete(?ATIMES, ATime),
- true = ets:insert(?ATIMES, {NewATime, DbName}),
- true = ets:insert(?DBS, {DbName, NewATime});
- [] ->
- ok
- end.
-
-cache_free(#st{max_size=Max, cur_size=Cur}=St) when Max =< Cur ->
- ATime = ets:first(?ATIMES),
- [{ATime, DbName}] = ets:lookup(?ATIMES, ATime),
- true = ets:delete(?ATIMES, ATime),
- true = ets:delete(?DBS, DbName),
- true = ets:delete(?SHARDS, DbName),
- cache_free(St#st{cur_size=Cur-1});
-cache_free(St) ->
- St.
-
-cache_clear(St) ->
- true = ets:delete_all_objects(?DBS),
- true = ets:delete_all_objects(?SHARDS),
- true = ets:delete_all_objects(?ATIMES),
- St#st{cur_size=0}.
-
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3_sup.erl
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3_sup.erl b/src/mem3/src/mem3_sup.erl
deleted file mode 100644
index 6ff688b..0000000
--- a/src/mem3/src/mem3_sup.erl
+++ /dev/null
@@ -1,34 +0,0 @@
-% 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(mem3_sup).
--behaviour(supervisor).
--export([start_link/0, init/1]).
-
-start_link() ->
- supervisor:start_link({local, ?MODULE}, ?MODULE, []).
-
-init(_Args) ->
- Children = [
- child(mem3_events),
- child(mem3_nodes),
- child(mem3_sync_nodes), % Order important?
- child(mem3_sync),
- child(mem3_shards)
- ],
- {ok, {{one_for_one,10,1}, Children}}.
-
-child(mem3_events) ->
- MFA = {gen_event, start_link, [{local, mem3_events}]},
- {mem3_events, MFA, permanent, 1000, worker, dynamic};
-child(Child) ->
- {Child, {Child, start_link, []}, permanent, 1000, worker, [Child]}.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3_sync.erl
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3_sync.erl b/src/mem3/src/mem3_sync.erl
deleted file mode 100644
index e47f6fa..0000000
--- a/src/mem3/src/mem3_sync.erl
+++ /dev/null
@@ -1,344 +0,0 @@
-% 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(mem3_sync).
--behaviour(gen_server).
--export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
- code_change/3]).
-
--export([start_link/0, get_active/0, get_queue/0, push/1, push/2,
- remove_node/1, initial_sync/1, get_backlog/0]).
-
--import(queue, [in/2, out/1, to_list/1, join/2, from_list/1, is_empty/1]).
-
--include_lib("mem3/include/mem3.hrl").
--include_lib("couch/include/couch_db.hrl").
-
--record(state, {
- active = [],
- count = 0,
- limit,
- dict = dict:new(),
- waiting = queue:new(),
- update_notifier
-}).
-
--record(job, {name, node, count=nil, pid=nil}).
-
-start_link() ->
- gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
-
-get_active() ->
- gen_server:call(?MODULE, get_active).
-
-get_queue() ->
- gen_server:call(?MODULE, get_queue).
-
-get_backlog() ->
- gen_server:call(?MODULE, get_backlog).
-
-push(#shard{name = Name}, Target) ->
- push(Name, Target);
-push(Name, #shard{node=Node}) ->
- push(Name, Node);
-push(Name, Node) ->
- push(#job{name = Name, node = Node}).
-
-push(#job{node = Node} = Job) when Node =/= node() ->
- gen_server:cast(?MODULE, {push, Job});
-push(_) ->
- ok.
-
-remove_node(Node) ->
- gen_server:cast(?MODULE, {remove_node, Node}).
-
-init([]) ->
- process_flag(trap_exit, true),
- Concurrency = config:get("mem3", "sync_concurrency", "10"),
- gen_event:add_handler(mem3_events, mem3_sync_event, []),
- {ok, Pid} = start_update_notifier(),
- initial_sync(),
- {ok, #state{limit = list_to_integer(Concurrency), update_notifier=Pid}}.
-
-handle_call({push, Job}, From, State) ->
- handle_cast({push, Job#job{pid = From}}, State);
-
-handle_call(get_active, _From, State) ->
- {reply, State#state.active, State};
-
-handle_call(get_queue, _From, State) ->
- {reply, to_list(State#state.waiting), State};
-
-handle_call(get_backlog, _From, #state{active=A, waiting=WQ} = State) ->
- CA = lists:sum([C || #job{count=C} <- A, is_integer(C)]),
- CW = lists:sum([C || #job{count=C} <- to_list(WQ), is_integer(C)]),
- {reply, CA+CW, State}.
-
-handle_cast({push, DbName, Node}, State) ->
- handle_cast({push, #job{name = DbName, node = Node}}, State);
-
-handle_cast({push, Job}, #state{count=Count, limit=Limit} = State)
- when Count >= Limit ->
- {noreply, add_to_queue(State, Job)};
-
-handle_cast({push, Job}, State) ->
- #state{active = L, count = C} = State,
- #job{name = DbName, node = Node} = Job,
- case is_running(DbName, Node, L) of
- true ->
- {noreply, add_to_queue(State, Job)};
- false ->
- Pid = start_push_replication(Job),
- {noreply, State#state{active=[Job#job{pid=Pid}|L], count=C+1}}
- end;
-
-handle_cast({remove_node, Node}, #state{waiting = W0} = State) ->
- {Alive, Dead} = lists:partition(fun(#job{node=N}) -> N =/= Node end, to_list(W0)),
- Dict = remove_entries(State#state.dict, Dead),
- [exit(Pid, die_now) || #job{node=N, pid=Pid} <- State#state.active,
- N =:= Node],
- {noreply, State#state{dict = Dict, waiting = from_list(Alive)}};
-
-handle_cast({remove_shard, Shard}, #state{waiting = W0} = State) ->
- {Alive, Dead} = lists:partition(fun(#job{name=S}) ->
- S =/= Shard end, to_list(W0)),
- Dict = remove_entries(State#state.dict, Dead),
- [exit(Pid, die_now) || #job{name=S, pid=Pid} <- State#state.active,
- S =:= Shard],
- {noreply, State#state{dict = Dict, waiting = from_list(Alive)}}.
-
-handle_info({'EXIT', Pid, _}, #state{update_notifier=Pid} = State) ->
- {ok, NewPid} = start_update_notifier(),
- {noreply, State#state{update_notifier=NewPid}};
-
-handle_info({'EXIT', Active, normal}, State) ->
- handle_replication_exit(State, Active);
-
-handle_info({'EXIT', Active, die_now}, State) ->
- % we forced this one ourselves, do not retry
- handle_replication_exit(State, Active);
-
-handle_info({'EXIT', Active, {{not_found, no_db_file}, _Stack}}, State) ->
- % target doesn't exist, do not retry
- handle_replication_exit(State, Active);
-
-handle_info({'EXIT', Active, Reason}, State) ->
- NewState = case lists:keyfind(Active, #job.pid, State#state.active) of
- #job{name=OldDbName, node=OldNode} = Job ->
- twig:log(warn, "~s ~s ~s ~w", [?MODULE, OldDbName, OldNode, Reason]),
- case Reason of {pending_changes, Count} ->
- maybe_resubmit(State, Job#job{pid = nil, count = Count});
- _ ->
- try mem3:shards(mem3:dbname(Job#job.name)) of _ ->
- timer:apply_after(5000, ?MODULE, push, [Job#job{pid=nil}])
- catch error:database_does_not_exist ->
- % no need to retry
- ok
- end,
- State
- end;
- false -> State end,
- handle_replication_exit(NewState, Active);
-
-handle_info(Msg, State) ->
- twig:log(notice, "unexpected msg at replication manager ~p", [Msg]),
- {noreply, State}.
-
-terminate(_Reason, State) ->
- [exit(Pid, shutdown) || #job{pid=Pid} <- State#state.active],
- ok.
-
-code_change(_, #state{waiting = WaitingList} = State, _) when is_list(WaitingList) ->
- {ok, State#state{waiting = from_list(WaitingList)}};
-
-code_change(_, State, _) ->
- {ok, State}.
-
-maybe_resubmit(State, #job{name=DbName, node=Node} = Job) ->
- case lists:member(DbName, local_dbs()) of
- true ->
- case find_next_node() of
- Node ->
- add_to_queue(State, Job);
- _ ->
- State % don't resubmit b/c we have a new replication target
- end;
- false ->
- add_to_queue(State, Job)
- end.
-
-handle_replication_exit(State, Pid) ->
- #state{active=Active, limit=Limit, dict=D, waiting=Waiting} = State,
- Active1 = lists:keydelete(Pid, #job.pid, Active),
- case is_empty(Waiting) of
- true ->
- {noreply, State#state{active=Active1, count=length(Active1)}};
- _ ->
- Count = length(Active1),
- NewState = if Count < Limit ->
- case next_replication(Active1, Waiting, queue:new()) of
- nil -> % all waiting replications are also active
- State#state{active = Active1, count = Count};
- {#job{name=DbName, node=Node} = Job, StillWaiting} ->
- NewPid = start_push_replication(Job),
- State#state{
- active = [Job#job{pid = NewPid} | Active1],
- count = Count+1,
- dict = dict:erase({DbName,Node}, D),
- waiting = StillWaiting
- }
- end;
- true ->
- State#state{active = Active1, count=Count}
- end,
- {noreply, NewState}
- end.
-
-start_push_replication(#job{name=Name, node=Node, pid=From}) ->
- if From =/= nil -> gen_server:reply(From, ok); true -> ok end,
- spawn_link(fun() ->
- case mem3_rep:go(Name, maybe_redirect(Node)) of
- {ok, Pending} when Pending > 0 ->
- exit({pending_changes, Pending});
- _ ->
- ok
- end
- end).
-
-add_to_queue(State, #job{name=DbName, node=Node, pid=From} = Job) ->
- #state{dict=D, waiting=WQ} = State,
- case dict:is_key({DbName, Node}, D) of
- true ->
- if From =/= nil -> gen_server:reply(From, ok); true -> ok end,
- State;
- false ->
- twig:log(debug, "adding ~s -> ~p to mem3_sync queue", [DbName, Node]),
- State#state{
- dict = dict:store({DbName,Node}, ok, D),
- waiting = in(Job, WQ)
- }
- end.
-
-sync_nodes_and_dbs() ->
- Node = find_next_node(),
- [push(Db, Node) || Db <- local_dbs()].
-
-initial_sync() ->
- [net_kernel:connect_node(Node) || Node <- mem3:nodes()],
- mem3_sync_nodes:add(nodes()).
-
-initial_sync(Live) ->
- sync_nodes_and_dbs(),
- Acc = {node(), Live, []},
- {_, _, Shards} = mem3_shards:fold(fun initial_sync_fold/2, Acc),
- submit_replication_tasks(node(), Live, Shards).
-
-initial_sync_fold(#shard{dbname = Db} = Shard, {LocalNode, Live, AccShards}) ->
- case AccShards of
- [#shard{dbname = AccDb} | _] when Db =/= AccDb ->
- submit_replication_tasks(LocalNode, Live, AccShards),
- {LocalNode, Live, [Shard]};
- _ ->
- {LocalNode, Live, [Shard|AccShards]}
- end.
-
-submit_replication_tasks(LocalNode, Live, Shards) ->
- SplitFun = fun(#shard{node = Node}) -> Node =:= LocalNode end,
- {Local, Remote} = lists:partition(SplitFun, Shards),
- lists:foreach(fun(#shard{name = ShardName}) ->
- [sync_push(ShardName, N) || #shard{node=N, name=Name} <- Remote,
- Name =:= ShardName, lists:member(N, Live)]
- end, Local).
-
-sync_push(ShardName, N) ->
- gen_server:call(mem3_sync, {push, #job{name=ShardName, node=N}}, infinity).
-
-start_update_notifier() ->
- Db1 = nodes_db(),
- Db2 = shards_db(),
- Db3 = users_db(),
- couch_db_update_notifier:start_link(fun
- ({updated, Db}) when Db == Db1 ->
- Nodes = mem3:nodes(),
- Live = nodes(),
- [?MODULE:push(Db1, N) || N <- Nodes, lists:member(N, Live)];
- ({updated, Db}) when Db == Db2; Db == Db3 ->
- ?MODULE:push(Db, find_next_node());
- ({updated, <<"shards/", _/binary>> = ShardName}) ->
- % TODO deal with split/merged partitions by comparing keyranges
- try mem3:shards(mem3:dbname(ShardName)) of
- Shards ->
- Targets = [S || #shard{node=N, name=Name} = S <- Shards,
- N =/= node(), Name =:= ShardName],
- Live = nodes(),
- [?MODULE:push(ShardName,N) || #shard{node=N} <- Targets,
- lists:member(N, Live)]
- catch error:database_does_not_exist ->
- ok
- end;
- ({deleted, <<"shards/", _:18/binary, _/binary>> = ShardName}) ->
- gen_server:cast(?MODULE, {remove_shard, ShardName});
- (_) -> ok end).
-
-find_next_node() ->
- LiveNodes = [node()|nodes()],
- AllNodes0 = lists:sort(mem3:nodes()),
- AllNodes1 = [X || X <- AllNodes0, lists:member(X, LiveNodes)],
- AllNodes = AllNodes1 ++ [hd(AllNodes1)],
- [_Self, Next| _] = lists:dropwhile(fun(N) -> N =/= node() end, AllNodes),
- Next.
-
-%% @doc Finds the next {DbName,Node} pair in the list of waiting replications
-%% which does not correspond to an already running replication
--spec next_replication([#job{}], [#job{}], [#job{}]) -> {#job{}, [#job{}]} | nil.
-next_replication(Active, Waiting, WaitingAndRunning) ->
- case is_empty(Waiting) of
- true ->
- nil;
- false ->
- {{value, #job{name=S, node=N} = Job}, RemQ} = out(Waiting),
- case is_running(S,N,Active) of
- true ->
- next_replication(Active, RemQ, in(Job, WaitingAndRunning));
- false ->
- {Job, join(RemQ, WaitingAndRunning)}
- end
- end.
-
-is_running(DbName, Node, ActiveList) ->
- [] =/= [true || #job{name=S, node=N} <- ActiveList, S=:=DbName, N=:=Node].
-
-remove_entries(Dict, Entries) ->
- lists:foldl(fun(#job{name=S, node=N}, D) ->
- dict:erase({S, N}, D)
- end, Dict, Entries).
-
-local_dbs() ->
- [nodes_db(), shards_db(), users_db()].
-
-nodes_db() ->
- ?l2b(config:get("mem3", "node_db", "nodes")).
-
-shards_db() ->
- ?l2b(config:get("mem3", "shard_db", "dbs")).
-
-users_db() ->
- ?l2b(config:get("couch_httpd_auth", "authentication_db", "_users")).
-
-maybe_redirect(Node) ->
- case config:get("mem3.redirects", atom_to_list(Node)) of
- undefined ->
- Node;
- Redirect ->
- twig:log(debug, "Redirecting push from ~p to ~p", [Node, Redirect]),
- list_to_existing_atom(Redirect)
- end.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3_sync_event.erl
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3_sync_event.erl b/src/mem3/src/mem3_sync_event.erl
deleted file mode 100644
index 7a20b0b..0000000
--- a/src/mem3/src/mem3_sync_event.erl
+++ /dev/null
@@ -1,85 +0,0 @@
-% 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(mem3_sync_event).
--behaviour(gen_event).
-
--export([init/1, handle_event/2, handle_call/2, handle_info/2, terminate/2,
- code_change/3]).
-
-init(_) ->
- net_kernel:monitor_nodes(true),
- {ok, nil}.
-
-handle_event({add_node, Node}, State) when Node =/= node() ->
- net_kernel:connect_node(Node),
- mem3_sync_nodes:add([Node]),
- {ok, State};
-
-handle_event({remove_node, Node}, State) ->
- mem3_sync:remove_node(Node),
- {ok, State};
-
-handle_event(_Event, State) ->
- {ok, State}.
-
-handle_call(_Request, State) ->
- {ok, ok, State}.
-
-handle_info({nodeup, Node}, State) ->
- Nodes0 = lists:usort(drain_nodeups([Node])),
- Nodes = lists:filter(fun(N) -> lists:member(N, mem3:nodes()) end, Nodes0),
- wait_for_rexi(Nodes, 5),
- {ok, State};
-
-handle_info({nodedown, Node}, State) ->
- mem3_sync:remove_node(Node),
- {ok, State};
-
-handle_info(_Info, State) ->
- {ok, State}.
-
-terminate(_Reason, _State) ->
- ok.
-
-code_change(_OldVsn, State, _Extra) ->
- {ok, State}.
-
-drain_nodeups(Acc) ->
- receive
- {nodeup, Node} ->
- drain_nodeups([Node | Acc])
- after 0 ->
- Acc
- end.
-
-wait_for_rexi([], _Retries) ->
- ok;
-wait_for_rexi(Waiting, Retries) ->
- % Hack around rpc:multicall/4 so that we can
- % be sure which nodes gave which response
- Msg = {call, erlang, whereis, [rexi_server], group_leader()},
- {Resp, _Bad} = gen_server:multi_call(Waiting, rex, Msg, 1000),
- Up = [N || {N, P} <- Resp, is_pid(P)],
- NotUp = Waiting -- Up,
- case length(Up) > 0 of
- true ->
- mem3_sync_nodes:add(Up);
- false -> ok
- end,
- case length(NotUp) > 0 andalso Retries > 0 of
- true ->
- timer:sleep(1000),
- wait_for_rexi(NotUp, Retries-1);
- false ->
- ok
- end.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3_sync_nodes.erl
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3_sync_nodes.erl b/src/mem3/src/mem3_sync_nodes.erl
deleted file mode 100644
index e07fd44..0000000
--- a/src/mem3/src/mem3_sync_nodes.erl
+++ /dev/null
@@ -1,114 +0,0 @@
-% 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(mem3_sync_nodes).
--behaviour(gen_server).
-
-
--export([start_link/0]).
--export([add/1]).
-
--export([init/1, terminate/2, code_change/3]).
--export([handle_call/3, handle_cast/2, handle_info/2]).
-
--export([monitor_sync/1]).
-
-
--record(st, {
- tid
-}).
-
-
--record(job, {
- nodes,
- pid,
- retry
-}).
-
-
-start_link() ->
- gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
-
-
-add(Nodes) ->
- gen_server:cast(?MODULE, {add, Nodes}).
-
-
-init([]) ->
- {ok, #st{
- tid = ets:new(?MODULE, [set, protected, {keypos, #job.nodes}])
- }}.
-
-
-terminate(_Reason, St) ->
- [exit(Pid, kill) || #job{pid=Pid} <- ets:tab2list(St#st.tid)],
- ok.
-
-
-handle_call(Msg, _From, St) ->
- {stop, {invalid_call, Msg}, invalid_call, St}.
-
-
-handle_cast({add, Nodes}, #st{tid=Tid}=St) ->
- case ets:lookup(Tid, Nodes) of
- [] ->
- Pid = start_sync(Nodes),
- ets:insert(Tid, #job{nodes=Nodes, pid=Pid, retry=false});
- [#job{retry=false}=Job] ->
- ets:insert(Tid, Job#job{retry=true});
- _ ->
- ok
- end,
- {noreply, St};
-
-handle_cast(Msg, St) ->
- {stop, {invalid_cast, Msg}, St}.
-
-
-handle_info({'DOWN', _, _, _, {sync_done, Nodes}}, #st{tid=Tid}=St) ->
- case ets:lookup(Tid, Nodes) of
- [#job{retry=true}=Job] ->
- Pid = start_sync(Nodes),
- ets:insert(Tid, Job#job{pid=Pid, retry=false});
- _ ->
- ets:delete(Tid, Nodes)
- end,
- {noreply, St};
-
-handle_info({'DOWN', _, _, _, {sync_error, Nodes}}, #st{tid=Tid}=St) ->
- Pid = start_sync(Nodes),
- ets:insert(Tid, #job{nodes=Nodes, pid=Pid, retry=false}),
- {noreply, St};
-
-handle_info(Msg, St) ->
- {stop, {invalid_info, Msg}, St}.
-
-
-code_change(_OldVsn, St, _Extra) ->
- {ok, St}.
-
-
-start_sync(Nodes) ->
- {Pid, _} = spawn_monitor(?MODULE, monitor_sync, [Nodes]),
- Pid.
-
-
-monitor_sync(Nodes) ->
- process_flag(trap_exit, true),
- Pid = spawn_link(mem3_sync, initial_sync, [Nodes]),
- receive
- {'EXIT', Pid, normal} ->
- exit({sync_done, Nodes});
- _ ->
- exit({sync_error, Nodes})
- end.
-
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3_sync_security.erl
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3_sync_security.erl b/src/mem3/src/mem3_sync_security.erl
deleted file mode 100644
index da112aa..0000000
--- a/src/mem3/src/mem3_sync_security.erl
+++ /dev/null
@@ -1,105 +0,0 @@
-% 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(mem3_sync_security).
-
--export([maybe_sync/2, maybe_sync_int/2]).
--export([go/0, go/1]).
-
--include_lib("mem3/include/mem3.hrl").
-
-
-maybe_sync(#shard{}=Src, #shard{}=Dst) ->
- case is_local(Src#shard.name) of
- false ->
- erlang:spawn(?MODULE, maybe_sync_int, [Src, Dst]);
- true ->
- ok
- end.
-
-maybe_sync_int(#shard{name=Name}=Src, Dst) ->
- DbName = mem3:dbname(Name),
- case fabric:get_all_security(DbName, [{shards, [Src, Dst]}]) of
- {ok, WorkerObjs} ->
- Objs = [Obj || {_Worker, Obj} <- WorkerObjs],
- case length(lists:usort(Objs)) of
- 1 -> ok;
- 2 -> go(DbName)
- end;
- Else ->
- Args = [DbName, Else],
- twig:log(err, "Error checking security objects for ~s :: ~p", Args)
- end.
-
-go() ->
- {ok, Dbs} = fabric:all_dbs(),
- lists:foreach(fun handle_db/1, Dbs).
-
-go(DbName) when is_binary(DbName) ->
- handle_db(DbName).
-
-handle_db(DbName) ->
- ShardCount = length(mem3:shards(DbName)),
- case get_all_security(DbName) of
- {ok, SecObjs} ->
- case is_ok(SecObjs, ShardCount) of
- ok ->
- ok;
- {fixable, SecObj} ->
- twig:log(info, "Sync security object for ~p: ~p", [DbName, SecObj]),
- case fabric:set_security(DbName, SecObj) of
- ok -> ok;
- Error ->
- twig:log(err, "Error setting security object in ~p: ~p",
- [DbName, Error])
- end;
- broken ->
- twig:log(err, "Bad security object in ~p: ~p", [DbName, SecObjs])
- end;
- Error ->
- twig:log(err, "Error getting security objects for ~p: ~p", [
- DbName, Error])
- end.
-
-get_all_security(DbName) ->
- case fabric:get_all_security(DbName) of
- {ok, SecObjs} ->
- SecObjsDict = lists:foldl(fun({_, SO}, Acc) ->
- dict:update_counter(SO, 1, Acc)
- end, dict:new(), SecObjs),
- {ok, dict:to_list(SecObjsDict)};
- Error ->
- Error
- end.
-
-is_ok([_], _) ->
- % One security object is the happy case
- ok;
-is_ok([_, _] = SecObjs0, ShardCount) ->
- % Figure out if we have a simple majority of security objects
- % and if so, use that as the correct value. Otherwise we abort
- % and rely on human intervention.
- {Count, SecObj} = lists:max([{C, O} || {O, C} <- SecObjs0]),
- case Count >= ((ShardCount div 2) + 1) of
- true -> {fixable, SecObj};
- false -> broken
- end;
-is_ok(_, _) ->
- % Anything else requires human intervention
- broken.
-
-
-is_local(<<"shards/", _/binary>>) ->
- false;
-is_local(_) ->
- true.
-
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/src/mem3_util.erl
----------------------------------------------------------------------
diff --git a/src/mem3/src/mem3_util.erl b/src/mem3/src/mem3_util.erl
deleted file mode 100644
index 4460df6..0000000
--- a/src/mem3/src/mem3_util.erl
+++ /dev/null
@@ -1,196 +0,0 @@
-% 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(mem3_util).
-
--export([hash/1, name_shard/2, create_partition_map/5, build_shards/2,
- n_val/2, to_atom/1, to_integer/1, write_db_doc/1, delete_db_doc/1,
- shard_info/1, ensure_exists/1, open_db_doc/1]).
--export([owner/2, is_deleted/1]).
-
--export([create_partition_map/4, name_shard/1]).
--deprecated({create_partition_map, 4, eventually}).
--deprecated({name_shard, 1, eventually}).
-
--define(RINGTOP, 2 bsl 31). % CRC32 space
-
--include_lib("mem3/include/mem3.hrl").
--include_lib("couch/include/couch_db.hrl").
-
-hash(Item) when is_binary(Item) ->
- erlang:crc32(Item);
-hash(Item) ->
- erlang:crc32(term_to_binary(Item)).
-
-name_shard(Shard) ->
- name_shard(Shard, "").
-
-name_shard(#shard{dbname = DbName, range=[B,E]} = Shard, Suffix) ->
- Name = ["shards/", couch_util:to_hex(<<B:32/integer>>), "-",
- couch_util:to_hex(<<E:32/integer>>), "/", DbName, Suffix],
- Shard#shard{name = ?l2b(Name)}.
-
-create_partition_map(DbName, N, Q, Nodes) ->
- create_partition_map(DbName, N, Q, Nodes, "").
-
-create_partition_map(DbName, N, Q, Nodes, Suffix) ->
- UniqueShards = make_key_ranges((?RINGTOP) div Q, 0, []),
- Shards0 = lists:flatten([lists:duplicate(N, S) || S <- UniqueShards]),
- Shards1 = attach_nodes(Shards0, [], Nodes, []),
- [name_shard(S#shard{dbname=DbName}, Suffix) || S <- Shards1].
-
-make_key_ranges(_, CurrentPos, Acc) when CurrentPos >= ?RINGTOP ->
- Acc;
-make_key_ranges(Increment, Start, Acc) ->
- case Start + 2*Increment of
- X when X > ?RINGTOP ->
- End = ?RINGTOP - 1;
- _ ->
- End = Start + Increment - 1
- end,
- make_key_ranges(Increment, End+1, [#shard{range=[Start, End]} | Acc]).
-
-attach_nodes([], Acc, _, _) ->
- lists:reverse(Acc);
-attach_nodes(Shards, Acc, [], UsedNodes) ->
- attach_nodes(Shards, Acc, lists:reverse(UsedNodes), []);
-attach_nodes([S | Rest], Acc, [Node | Nodes], UsedNodes) ->
- attach_nodes(Rest, [S#shard{node=Node} | Acc], Nodes, [Node | UsedNodes]).
-
-open_db_doc(DocId) ->
- DbName = ?l2b(config:get("mem3", "shard_db", "dbs")),
- {ok, Db} = couch_db:open(DbName, []),
- try couch_db:open_doc(Db, DocId, [ejson_body]) after couch_db:close(Db) end.
-
-write_db_doc(Doc) ->
- DbName = ?l2b(config:get("mem3", "shard_db", "dbs")),
- write_db_doc(DbName, Doc, true).
-
-write_db_doc(DbName, #doc{id=Id, body=Body} = Doc, ShouldMutate) ->
- {ok, Db} = couch_db:open(DbName, []),
- try couch_db:open_doc(Db, Id, [ejson_body]) of
- {ok, #doc{body = Body}} ->
- % the doc is already in the desired state, we're done here
- ok;
- {not_found, _} when ShouldMutate ->
- try couch_db:update_doc(Db, Doc, []) of
- {ok, _} ->
- ok
- catch conflict ->
- % check to see if this was a replication race or a different edit
- write_db_doc(DbName, Doc, false)
- end;
- _ ->
- % the doc already exists in a different state
- conflict
- after
- couch_db:close(Db)
- end.
-
-delete_db_doc(DocId) ->
- gen_server:cast(mem3_shards, {cache_remove, DocId}),
- DbName = ?l2b(config:get("mem3", "shard_db", "dbs")),
- delete_db_doc(DbName, DocId, true).
-
-delete_db_doc(DbName, DocId, ShouldMutate) ->
- {ok, Db} = couch_db:open(DbName, []),
- {ok, Revs} = couch_db:open_doc_revs(Db, DocId, all, []),
- try [Doc#doc{deleted=true} || {ok, #doc{deleted=false}=Doc} <- Revs] of
- [] ->
- not_found;
- Docs when ShouldMutate ->
- try couch_db:update_docs(Db, Docs, []) of
- {ok, _} ->
- ok
- catch conflict ->
- % check to see if this was a replication race or if leafs survived
- delete_db_doc(DbName, DocId, false)
- end;
- _ ->
- % we have live leafs that we aren't allowed to delete. let's bail
- conflict
- after
- couch_db:close(Db)
- end.
-
-build_shards(DbName, DocProps) ->
- {ByNode} = couch_util:get_value(<<"by_node">>, DocProps, {[]}),
- Suffix = couch_util:get_value(<<"shard_suffix">>, DocProps, ""),
- lists:flatmap(fun({Node, Ranges}) ->
- lists:map(fun(Range) ->
- [B,E] = string:tokens(?b2l(Range), "-"),
- Beg = httpd_util:hexlist_to_integer(B),
- End = httpd_util:hexlist_to_integer(E),
- name_shard(#shard{
- dbname = DbName,
- node = to_atom(Node),
- range = [Beg, End]
- }, Suffix)
- end, Ranges)
- end, ByNode).
-
-to_atom(Node) when is_binary(Node) ->
- list_to_atom(binary_to_list(Node));
-to_atom(Node) when is_atom(Node) ->
- Node.
-
-to_integer(N) when is_integer(N) ->
- N;
-to_integer(N) when is_binary(N) ->
- list_to_integer(binary_to_list(N));
-to_integer(N) when is_list(N) ->
- list_to_integer(N).
-
-n_val(undefined, NodeCount) ->
- n_val(config:get("cluster", "n", "3"), NodeCount);
-n_val(N, NodeCount) when is_list(N) ->
- n_val(list_to_integer(N), NodeCount);
-n_val(N, NodeCount) when is_integer(NodeCount), N > NodeCount ->
- twig:log(error, "Request to create N=~p DB but only ~p node(s)", [N, NodeCount]),
- NodeCount;
-n_val(N, _) when N < 1 ->
- 1;
-n_val(N, _) ->
- N.
-
-shard_info(DbName) ->
- [{n, mem3:n(DbName)},
- {q, length(mem3:shards(DbName)) div mem3:n(DbName)}].
-
-ensure_exists(DbName) when is_list(DbName) ->
- ensure_exists(list_to_binary(DbName));
-ensure_exists(DbName) ->
- Options = [{user_ctx, #user_ctx{roles=[<<"_admin">>]}}],
- case couch_db:open(DbName, Options) of
- {ok, Db} ->
- {ok, Db};
- _ ->
- couch_server:create(DbName, Options)
- end.
-
-
-owner(DbName, DocId) ->
- Shards = mem3:shards(DbName, DocId),
- Nodes = [node()|nodes()],
- LiveShards = [S || #shard{node=Node} = S <- Shards, lists:member(Node, Nodes)],
- [#shard{node=Node}] = lists:usort(fun(#shard{name=A}, #shard{name=B}) ->
- A =< B end, LiveShards),
- node() =:= Node.
-
-is_deleted(Change) ->
- case couch_util:get_value(<<"deleted">>, Change) of
- undefined ->
- % keep backwards compatibility for a while
- couch_util:get_value(deleted, Change, false);
- Else ->
- Else
- end.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/test/01-config-default.ini
----------------------------------------------------------------------
diff --git a/src/mem3/test/01-config-default.ini b/src/mem3/test/01-config-default.ini
deleted file mode 100644
index 757f783..0000000
--- a/src/mem3/test/01-config-default.ini
+++ /dev/null
@@ -1,2 +0,0 @@
-[cluster]
-n=3
http://git-wip-us.apache.org/repos/asf/couchdb/blob/4cac46af/src/mem3/test/mem3_util_test.erl
----------------------------------------------------------------------
diff --git a/src/mem3/test/mem3_util_test.erl b/src/mem3/test/mem3_util_test.erl
deleted file mode 100644
index e289282..0000000
--- a/src/mem3/test/mem3_util_test.erl
+++ /dev/null
@@ -1,152 +0,0 @@
-% 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(mem3_util_test).
-
--include("mem3.hrl").
--include_lib("eunit/include/eunit.hrl").
-
-hash_test() ->
- ?assertEqual(1624516141,mem3_util:hash(0)),
- ?assertEqual(3816901808,mem3_util:hash("0")),
- ?assertEqual(3523407757,mem3_util:hash(<<0>>)),
- ?assertEqual(4108050209,mem3_util:hash(<<"0">>)),
- ?assertEqual(3094724072,mem3_util:hash(zero)),
- ok.
-
-name_shard_test() ->
- Shard1 = #shard{},
- ?assertError(function_clause, mem3_util:name_shard(Shard1, ".1234")),
-
- Shard2 = #shard{dbname = <<"testdb">>, range = [0,100]},
- #shard{name=Name2} = mem3_util:name_shard(Shard2, ".1234"),
- ?assertEqual(<<"shards/00000000-00000064/testdb.1234">>, Name2),
-
- ok.
-
-create_partition_map_test() ->
- {DbName1, N1, Q1, Nodes1} = {<<"testdb1">>, 3, 4, [a,b,c,d]},
- Map1 = mem3_util:create_partition_map(DbName1, N1, Q1, Nodes1),
- ?assertEqual(12, length(Map1)),
-
- {DbName2, N2, Q2, Nodes2} = {<<"testdb2">>, 1, 1, [a,b,c,d]},
- [#shard{name=Name2,node=Node2}] = Map2 =
- mem3_util:create_partition_map(DbName2, N2, Q2, Nodes2, ".1234"),
- ?assertEqual(1, length(Map2)),
- ?assertEqual(<<"shards/00000000-ffffffff/testdb2.1234">>, Name2),
- ?assertEqual(a, Node2),
- ok.
-
-build_shards_test() ->
- DocProps1 =
- [{<<"changelog">>,
- [[<<"add">>,<<"00000000-1fffffff">>,
- <<"bigcouch@node.local">>],
- [<<"add">>,<<"20000000-3fffffff">>,
- <<"bigcouch@node.local">>],
- [<<"add">>,<<"40000000-5fffffff">>,
- <<"bigcouch@node.local">>],
- [<<"add">>,<<"60000000-7fffffff">>,
- <<"bigcouch@node.local">>],
- [<<"add">>,<<"80000000-9fffffff">>,
- <<"bigcouch@node.local">>],
- [<<"add">>,<<"a0000000-bfffffff">>,
- <<"bigcouch@node.local">>],
- [<<"add">>,<<"c0000000-dfffffff">>,
- <<"bigcouch@node.local">>],
- [<<"add">>,<<"e0000000-ffffffff">>,
- <<"bigcouch@node.local">>]]},
- {<<"by_node">>,
- {[{<<"bigcouch@node.local">>,
- [<<"00000000-1fffffff">>,<<"20000000-3fffffff">>,
- <<"40000000-5fffffff">>,<<"60000000-7fffffff">>,
- <<"80000000-9fffffff">>,<<"a0000000-bfffffff">>,
- <<"c0000000-dfffffff">>,<<"e0000000-ffffffff">>]}]}},
- {<<"by_range">>,
- {[{<<"00000000-1fffffff">>,[<<"bigcouch@node.local">>]},
- {<<"20000000-3fffffff">>,[<<"bigcouch@node.local">>]},
- {<<"40000000-5fffffff">>,[<<"bigcouch@node.local">>]},
- {<<"60000000-7fffffff">>,[<<"bigcouch@node.local">>]},
- {<<"80000000-9fffffff">>,[<<"bigcouch@node.local">>]},
- {<<"a0000000-bfffffff">>,[<<"bigcouch@node.local">>]},
- {<<"c0000000-dfffffff">>,[<<"bigcouch@node.local">>]},
- {<<"e0000000-ffffffff">>,[<<"bigcouch@node.local">>]}]}}],
- Shards1 = mem3_util:build_shards(<<"testdb1">>, DocProps1),
- ExpectedShards1 =
- [{shard,<<"shards/00000000-1fffffff/testdb1">>,
- 'bigcouch@node.local',<<"testdb1">>,
- [0,536870911],
- undefined},
- {shard,<<"shards/20000000-3fffffff/testdb1">>,
- 'bigcouch@node.local',<<"testdb1">>,
- [536870912,1073741823],
- undefined},
- {shard,<<"shards/40000000-5fffffff/testdb1">>,
- 'bigcouch@node.local',<<"testdb1">>,
- [1073741824,1610612735],
- undefined},
- {shard,<<"shards/60000000-7fffffff/testdb1">>,
- 'bigcouch@node.local',<<"testdb1">>,
- [1610612736,2147483647],
- undefined},
- {shard,<<"shards/80000000-9fffffff/testdb1">>,
- 'bigcouch@node.local',<<"testdb1">>,
- [2147483648,2684354559],
- undefined},
- {shard,<<"shards/a0000000-bfffffff/testdb1">>,
- 'bigcouch@node.local',<<"testdb1">>,
- [2684354560,3221225471],
- undefined},
- {shard,<<"shards/c0000000-dfffffff/testdb1">>,
- 'bigcouch@node.local',<<"testdb1">>,
- [3221225472,3758096383],
- undefined},
- {shard,<<"shards/e0000000-ffffffff/testdb1">>,
- 'bigcouch@node.local',<<"testdb1">>,
- [3758096384,4294967295],
- undefined}],
- ?assertEqual(ExpectedShards1, Shards1),
- ok.
-
-
-%% n_val tests
-
-nval_test() ->
- ?assertEqual(2, mem3_util:n_val(2,4)),
- ?assertEqual(1, mem3_util:n_val(-1,4)),
- ?assertEqual(4, mem3_util:n_val(6,4)),
- ok.
-
-config_01_setup() ->
- Ini = filename:join([code:lib_dir(mem3, test), "01-config-default.ini"]),
- {ok, Pid} = config:start_link([Ini]),
- Pid.
-
-config_teardown(_Pid) ->
- config:stop().
-
-n_val_test_() ->
- {"n_val tests",
- [
- {setup,
- fun config_01_setup/0,
- fun config_teardown/1,
- fun(Pid) ->
- {with, Pid, [
- fun n_val_1/1
- ]}
- end}
- ]
- }.
-
-n_val_1(_Pid) ->
- ?assertEqual(3, mem3_util:n_val(undefined, 4)).