diff options
Diffstat (limited to 'src/fabric/src/fabric_db_create.erl')
| -rw-r--r-- | src/fabric/src/fabric_db_create.erl | 227 |
1 files changed, 115 insertions, 112 deletions
diff --git a/src/fabric/src/fabric_db_create.erl b/src/fabric/src/fabric_db_create.erl index 1048cb0bb..38770aea4 100644 --- a/src/fabric/src/fabric_db_create.erl +++ b/src/fabric/src/fabric_db_create.erl @@ -17,61 +17,60 @@ -include_lib("mem3/include/mem3.hrl"). -include_lib("couch/include/couch_db.hrl"). - %% @doc Create a new database, and all its partition files across the cluster %% Options is proplist with user_ctx, n, q, validate_name go(DbName, Options) -> case validate_dbname(DbName, Options) of - ok -> - couch_partition:validate_dbname(DbName, Options), - case db_exists(DbName) of - true -> - {error, file_exists}; - false -> - {Shards, Doc} = generate_shard_map(DbName, Options), - CreateShardResult = create_shard_files(Shards, Options), - case CreateShardResult of - enametoolong -> - {error, {database_name_too_long, DbName}}; - _ -> - case {CreateShardResult, create_shard_db_doc(Doc)} of - {ok, {ok, Status}} -> - Status; - {ok, {error, conflict} = ShardDocError} -> - % Check if it is just a race to create the shard doc - case db_exists(DbName) of - true -> {error, file_exists}; - false -> ShardDocError - end; - {file_exists, {ok, _}} -> + ok -> + couch_partition:validate_dbname(DbName, Options), + case db_exists(DbName) of + true -> {error, file_exists}; - {_, Error} -> - Error - end - end - end; - Error -> - Error + false -> + {Shards, Doc} = generate_shard_map(DbName, Options), + CreateShardResult = create_shard_files(Shards, Options), + case CreateShardResult of + enametoolong -> + {error, {database_name_too_long, DbName}}; + _ -> + case {CreateShardResult, create_shard_db_doc(Doc)} of + {ok, {ok, Status}} -> + Status; + {ok, {error, conflict} = ShardDocError} -> + % Check if it is just a race to create the shard doc + case db_exists(DbName) of + true -> {error, file_exists}; + false -> ShardDocError + end; + {file_exists, {ok, _}} -> + {error, file_exists}; + {_, Error} -> + Error + end + end + end; + Error -> + Error end. validate_dbname(DbName, Options) -> case couch_util:get_value(validate_name, Options, true) of - false -> - ok; - true -> - couch_db:validate_dbname(DbName) + false -> + ok; + true -> + couch_db:validate_dbname(DbName) end. generate_shard_map(DbName, Options) -> {MegaSecs, Secs, _} = os:timestamp(), - Suffix = "." ++ integer_to_list(MegaSecs*1000000 + Secs), - Shards = mem3:choose_shards(DbName, [{shard_suffix,Suffix} | Options]), + Suffix = "." ++ integer_to_list(MegaSecs * 1000000 + Secs), + Shards = mem3:choose_shards(DbName, [{shard_suffix, Suffix} | Options]), case mem3_util:open_db_doc(DbName) of - {ok, Doc} -> - % the DB already exists, and may have a different Suffix - ok; - {not_found, _} -> - Doc = make_document(Shards, Suffix, Options) + {ok, Doc} -> + % the DB already exists, and may have a different Suffix + ok; + {not_found, _} -> + Doc = make_document(Shards, Suffix, Options) end, {Shards, Doc}. @@ -79,56 +78,53 @@ create_shard_files(Shards, Options) -> Workers = fabric_util:submit_jobs(Shards, create_db, [Options]), RexiMon = fabric_util:create_monitors(Shards), try fabric_util:recv(Workers, #shard.ref, fun handle_message/3, Workers) of - {error, file_exists} -> - file_exists; - {error, enametoolong} -> - enametoolong; - {timeout, DefunctWorkers} -> - fabric_util:log_timeout(DefunctWorkers, "create_db"), - {error, timeout}; - _ -> - ok + {error, file_exists} -> + file_exists; + {error, enametoolong} -> + enametoolong; + {timeout, DefunctWorkers} -> + fabric_util:log_timeout(DefunctWorkers, "create_db"), + {error, timeout}; + _ -> + ok after rexi_monitor:stop(RexiMon) end. handle_message({error, enametoolong}, _, _) -> {error, enametoolong}; - handle_message(file_exists, _, _) -> {error, file_exists}; - handle_message({rexi_DOWN, _, {_, Node}, _}, _, Workers) -> case lists:filter(fun(S) -> S#shard.node =/= Node end, Workers) of - [] -> - {stop, ok}; - RemainingWorkers -> - {ok, RemainingWorkers} + [] -> + {stop, ok}; + RemainingWorkers -> + {ok, RemainingWorkers} end; - handle_message(_, Worker, Workers) -> case lists:delete(Worker, Workers) of - [] -> - {stop, ok}; - RemainingWorkers -> - {ok, RemainingWorkers} + [] -> + {stop, ok}; + RemainingWorkers -> + {ok, RemainingWorkers} end. create_shard_db_doc(Doc) -> - Shards = [#shard{node=N} || N <- mem3:nodes()], + Shards = [#shard{node = N} || N <- mem3:nodes()], RexiMon = fabric_util:create_monitors(Shards), Workers = fabric_util:submit_jobs(Shards, create_shard_db_doc, [Doc]), Acc0 = {length(Shards), fabric_dict:init(Workers, nil)}, try fabric_util:recv(Workers, #shard.ref, fun handle_db_update/3, Acc0) of - {timeout, {_, WorkersDict}} -> - DefunctWorkers = fabric_util:remove_done_workers(WorkersDict, nil), - fabric_util:log_timeout( - DefunctWorkers, - "create_shard_db_doc" - ), - {error, timeout}; - Else -> - Else + {timeout, {_, WorkersDict}} -> + DefunctWorkers = fabric_util:remove_done_workers(WorkersDict, nil), + fabric_util:log_timeout( + DefunctWorkers, + "create_shard_db_doc" + ), + {error, timeout}; + Else -> + Else after rexi_monitor:stop(RexiMon) end. @@ -136,57 +132,68 @@ create_shard_db_doc(Doc) -> handle_db_update({rexi_DOWN, _, {_, Node}, _}, _Worker, {W, Counters}) -> New = fabric_dict:filter(fun(S, _) -> S#shard.node =/= Node end, Counters), maybe_stop(W, New); - handle_db_update({rexi_EXIT, _Reason}, Worker, {W, Counters}) -> maybe_stop(W, fabric_dict:erase(Worker, Counters)); - handle_db_update(conflict, _, _) -> % just fail when we get any conflicts {error, conflict}; - handle_db_update(Msg, Worker, {W, Counters}) -> maybe_stop(W, fabric_dict:store(Worker, Msg, Counters)). maybe_stop(W, Counters) -> case fabric_dict:any(nil, Counters) of - true -> - {ok, {W, Counters}}; - false -> - case lists:sum([1 || {_, ok} <- Counters]) of - NumOk when NumOk >= (W div 2 +1) -> - {stop, ok}; - NumOk when NumOk > 0 -> - {stop, accepted}; - _ -> - {error, internal_server_error} - end + true -> + {ok, {W, Counters}}; + false -> + case lists:sum([1 || {_, ok} <- Counters]) of + NumOk when NumOk >= (W div 2 + 1) -> + {stop, ok}; + NumOk when NumOk > 0 -> + {stop, accepted}; + _ -> + {error, internal_server_error} + end end. -make_document([#shard{dbname=DbName}|_] = Shards, Suffix, Options) -> +make_document([#shard{dbname = DbName} | _] = Shards, Suffix, Options) -> {RawOut, ByNodeOut, ByRangeOut} = - lists:foldl(fun(#shard{node=N, range=[B,E]}, {Raw, ByNode, ByRange}) -> - Range = ?l2b([couch_util:to_hex(<<B:32/integer>>), "-", - couch_util:to_hex(<<E:32/integer>>)]), - Node = couch_util:to_binary(N), - {[[<<"add">>, Range, Node] | Raw], orddict:append(Node, Range, ByNode), - orddict:append(Range, Node, ByRange)} - end, {[], [], []}, Shards), - EngineProp = case couch_util:get_value(engine, Options) of - E when is_binary(E) -> [{<<"engine">>, E}]; - _ -> [] - end, - DbProps = case couch_util:get_value(props, Options) of - Props when is_list(Props) -> [{<<"props">>, {Props}}]; - _ -> [] - end, + lists:foldl( + fun(#shard{node = N, range = [B, E]}, {Raw, ByNode, ByRange}) -> + Range = ?l2b([ + couch_util:to_hex(<<B:32/integer>>), + "-", + couch_util:to_hex(<<E:32/integer>>) + ]), + Node = couch_util:to_binary(N), + { + [[<<"add">>, Range, Node] | Raw], + orddict:append(Node, Range, ByNode), + orddict:append(Range, Node, ByRange) + } + end, + {[], [], []}, + Shards + ), + EngineProp = + case couch_util:get_value(engine, Options) of + E when is_binary(E) -> [{<<"engine">>, E}]; + _ -> [] + end, + DbProps = + case couch_util:get_value(props, Options) of + Props when is_list(Props) -> [{<<"props">>, {Props}}]; + _ -> [] + end, #doc{ id = DbName, - body = {[ - {<<"shard_suffix">>, Suffix}, - {<<"changelog">>, lists:sort(RawOut)}, - {<<"by_node">>, {[{K,lists:sort(V)} || {K,V} <- ByNodeOut]}}, - {<<"by_range">>, {[{K,lists:sort(V)} || {K,V} <- ByRangeOut]}} - ] ++ EngineProp ++ DbProps} + body = { + [ + {<<"shard_suffix">>, Suffix}, + {<<"changelog">>, lists:sort(RawOut)}, + {<<"by_node">>, {[{K, lists:sort(V)} || {K, V} <- ByNodeOut]}}, + {<<"by_range">>, {[{K, lists:sort(V)} || {K, V} <- ByRangeOut]}} + ] ++ EngineProp ++ DbProps + } }. db_exists(DbName) -> is_list(catch mem3:shards(DbName)). @@ -205,24 +212,20 @@ db_exists_test_() -> ] }. - setup_all() -> meck:new(mem3). - teardown_all(_) -> meck:unload(). - db_exists_for_existing_db() -> Mock = fun(DbName) when is_binary(DbName) -> - [#shard{dbname = DbName, range = [0,100]}] + [#shard{dbname = DbName, range = [0, 100]}] end, ok = meck:expect(mem3, shards, Mock), ?assertEqual(true, db_exists(<<"foobar">>)), ?assertEqual(true, meck:validate(mem3)). - db_exists_for_missing_db() -> Mock = fun(DbName) -> erlang:error(database_does_not_exist, DbName) |