diff --git a/eqc/bg_manager_eqc.erl b/eqc/bg_manager_eqc.erl index e9a905cd..d648c5aa 100644 --- a/eqc/bg_manager_eqc.erl +++ b/eqc/bg_manager_eqc.erl @@ -1,6 +1,7 @@ +%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- %% ------------------------------------------------------------------- %% -%% Copyright (c) 2013 Basho Technologies, Inc. All Rights Reserved. +%% Copyright (c) 2013-2014 Basho Technologies, Inc. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file @@ -16,6 +17,8 @@ %% specific language governing permissions and limitations %% under the License. %% +%% ------------------------------------------------------------------- +%% %% QuickCheck may fail for both properties in this module %% The commands are not atomic and we cannot faithfully model %% the side effect of locking and freeing resources when the @@ -25,11 +28,13 @@ %% by mocking the call to release_resource. %% That however, would require a eqc_compnent specification %% instead of the current eqc_statem. - +%% -module(bg_manager_eqc). -ifdef(EQC). +-compile([export_all, nowarn_export_all]). + -include("include/riak_core_bg_manager.hrl"). -include_lib("eqc/include/eqc.hrl"). -include_lib("eqc/include/eqc_statem.hrl"). @@ -37,8 +42,6 @@ -define(QC_OUT(P), eqc:on_output(fun(Str, Args) -> io:format(user, Str, Args) end, P)). --compile([export_all, nowarn_export_all]). - -type bg_eqc_type() :: atom(). -type bg_eqc_limit() :: non_neg_integer(). @@ -857,11 +860,11 @@ bg_manager_monitors(Pid) -> prop_bgmgr() -> ?SETUP(fun() -> - error_logger:tty(false), - fun() -> - error_logger:tty(true) - end - end, + Level = riak_core_test_util:logger_silence(), + fun() -> + logger:set_handler_config(default, level, Level) + end + end, ?FORALL(Cmds, commands(?MODULE), ?SOMETIMES(2, aggregate(command_names(Cmds), @@ -909,11 +912,11 @@ prop_bgmgr() -> prop_bgmgr_parallel() -> ?SETUP(fun() -> - error_logger:tty(false), - fun() -> - error_logger:tty(true) - end - end, + Level = riak_core_test_util:logger_silence(), + fun() -> + logger:set_handler_config(default, level, Level) + end + end, ?FORALL(Cmds, parallel_commands(?MODULE, (initial_state())#state{exclude = [bypass]}), ?SOMETIMES(2, aggregate(command_names(Cmds), diff --git a/eqc/bprops_eqc.erl b/eqc/bprops_eqc.erl index 03392a36..fbd0296f 100644 --- a/eqc/bprops_eqc.erl +++ b/eqc/bprops_eqc.erl @@ -1,6 +1,7 @@ +%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- %% ------------------------------------------------------------------- %% -%% Copyright (c) 2016 Basho Technologies, Inc. All Rights Reserved. +%% Copyright (c) 2016 Basho Technologies, Inc. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file @@ -17,6 +18,7 @@ %% under the License. %% %% ------------------------------------------------------------------- +%% -module(bprops_eqc). %% @@ -32,11 +34,12 @@ %% -ifdef(EQC). --include_lib("eqc/include/eqc.hrl"). --include_lib("eqc/include/eqc_statem.hrl"). -compile([export_all, nowarn_export_all]). +-include_lib("eqc/include/eqc.hrl"). +-include_lib("eqc/include/eqc_statem.hrl"). + -type bucket_name() :: binary(). -type orddict() :: orddict:orddict(). @@ -249,7 +252,7 @@ prop_buckets() -> ). setup_cleanup() -> - error_logger:tty(false), + Level = riak_core_test_util:logger_silence(), meck:new(riak_core_capability, []), meck:expect( riak_core_capability, get, @@ -258,8 +261,8 @@ setup_cleanup() -> end ), fun() -> - error_logger:tty(true), - meck:unload(riak_core_capability) + meck:unload(riak_core_capability), + logger:set_handler_config(default, level, Level) end. %% diff --git a/eqc/btypes_eqc.erl b/eqc/btypes_eqc.erl index b7b675bd..6a767783 100644 --- a/eqc/btypes_eqc.erl +++ b/eqc/btypes_eqc.erl @@ -1,7 +1,7 @@ +%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- %% ------------------------------------------------------------------- %% -%% -%% Copyright (c) 2013 Basho Technologies, Inc. All Rights Reserved. +%% Copyright (c) 2013-2016 Basho Technologies, Inc. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file @@ -18,15 +18,17 @@ %% under the License. %% %% ------------------------------------------------------------------- +%% -module(btypes_eqc). -ifdef(EQC). + +-compile([export_all, nowarn_export_all]). + -include_lib("eqc/include/eqc.hrl"). -include_lib("eqc/include/eqc_statem.hrl"). -include_lib("eunit/include/eunit.hrl"). --compile([export_all, nowarn_export_all]). - -type type_name() :: binary(). -type type_active_status() :: boolean(). -type type_prop_name() :: binary(). @@ -360,7 +362,7 @@ prop_btype_invariant() -> ). setup_cleanup() -> - error_logger:tty(false), + Level = riak_core_test_util:logger_silence(), meck:new(riak_core_capability, []), meck:expect( riak_core_capability, get, @@ -369,8 +371,8 @@ setup_cleanup() -> end ), fun() -> - error_logger:tty(true), - meck:unload(riak_core_capability) + meck:unload(riak_core_capability), + logger:set_handler_config(default, level, Level) end. stop_pid(_Tag, Other) when not is_pid(Other) -> diff --git a/eqc/chash_eqc.erl b/eqc/chash_eqc.erl index c2d2d43b..5a37c9be 100644 --- a/eqc/chash_eqc.erl +++ b/eqc/chash_eqc.erl @@ -1,8 +1,7 @@ +%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- %% ------------------------------------------------------------------- %% -%% chash_eqc: QuickCheck tests for the chash module. -%% -%% Copyright (c) 2007-2011 Basho Technologies, Inc. All Rights Reserved. +%% Copyright (c) 2011-2012 Basho Technologies, Inc. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file @@ -19,12 +18,13 @@ %% under the License. %% %% ------------------------------------------------------------------- - +%% %% @doc QuickCheck tests for the chash module - +%% -module(chash_eqc). -ifdef(EQC). + -include_lib("eqc/include/eqc.hrl"). -include_lib("eunit/include/eunit.hrl"). @@ -67,18 +67,10 @@ eqc_test_() -> setup() -> %% Remove the logger noise. - application:load(sasl), - error_logger:tty(false), - %% Uncomment the following lines to send log output to files. - %% application:set_env(sasl, sasl_error_logger, {file, "chash_eqc_sasl.log"}), - %% error_logger:logfile({open, "chash_eqc.log"}), - - %% TODO: Perform any required setup - ok. - -cleanup(_) -> - %% TODO: Perform any required cleanup - ok. + riak_core_test_util:logger_silence(). + +cleanup(Level) -> + logger:set_handler_config(default, level, Level). %% ==================================================================== %% eqc property diff --git a/eqc/core_vnode_eqc.erl b/eqc/core_vnode_eqc.erl index 24292759..7d9c5776 100644 --- a/eqc/core_vnode_eqc.erl +++ b/eqc/core_vnode_eqc.erl @@ -1,8 +1,7 @@ +%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- %% ------------------------------------------------------------------- %% -%% core_vnode_eqc: QuickCheck tests for riak_core_vnode code -%% -%% Copyright (c) 2007-2010 Basho Technologies, Inc. All Rights Reserved. +%% Copyright (c) 2007-2014 Basho Technologies, Inc. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file @@ -19,16 +18,19 @@ %% under the License. %% %% ------------------------------------------------------------------- - +%% %% @doc QuickCheck tests for riak_core_vnode code - +%% -module(core_vnode_eqc). + -ifdef(EQC). + +-compile([export_all, nowarn_export_all]). + -include_lib("eqc/include/eqc.hrl"). -include_lib("eqc/include/eqc_fsm.hrl"). -include_lib("eunit/include/eunit.hrl"). -include("include/riak_core_vnode.hrl"). --compile([export_all, nowarn_export_all]). -define(POST_COND(PC, ErrTuple), case PC of @@ -49,9 +51,7 @@ async_work=[]}). % {Index, AsyncRef} async work submitted to each vnode setup_simple() -> - error_logger:tty(false), - application:set_env(sasl, sasl_error_logger, {file, "core_vnode_eqc_sasl.log"}), - error_logger:logfile({open, "core_vnode_eqc.log"}), + riak_core_test_util:logger_redirect(?MODULE), Vars = [{ring_creation_size, 8}, {ring_state_dir, ""}, diff --git a/eqc/worker_pool_pulse.erl b/eqc/worker_pool_pulse.erl index e6705315..fce53ee6 100644 --- a/eqc/worker_pool_pulse.erl +++ b/eqc/worker_pool_pulse.erl @@ -1,6 +1,7 @@ +%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- %% ------------------------------------------------------------------- %% -%% Copyright (c) 2013 Basho Technologies, Inc. All Rights Reserved. +%% Copyright (c) 2013 Basho Technologies, Inc. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file @@ -17,13 +18,14 @@ %% under the License. %% %% ------------------------------------------------------------------- - +%% %% @doc Test riak_core_vnode_worker_pool's interaction with poolboy %% under PULSE. This requires that riak_core, poolboy, and this module %% be compiled with the 'PULSE' macro defined. +%% -module(worker_pool_pulse). - -behaviour(riak_core_vnode_worker). + -ifdef(EQC). -include_lib("eqc/include/eqc.hrl"). -endif. @@ -123,11 +125,11 @@ all_work_gets_done(PoolSize, WorkList) -> length(Results) == length(WorkList). setup_and_teardown() -> - error_logger:tty(false), + Level = riak_core_test_util:logger_silence(), pulse:start(), fun() -> - pulse:stop(), - error_logger:tty(true) + pulse:stop(), + logger:set_handler_config(default, level, Level) end. -endif. diff --git a/rebar.config b/rebar.config index 16182180..2f990d84 100644 --- a/rebar.config +++ b/rebar.config @@ -1,13 +1,41 @@ +%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- +%% ------------------------------------------------------------------- +%% +%% Copyright (c) 2007-2017 Basho Technologies, Inc. +%% Copyright (c) 2019-2025 Workday, Inc. +%% +%% This file is provided to you 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. +%% +%% ------------------------------------------------------------------- {minimum_otp_vsn, "24.0"}. -{erl_first_files, ["src/gen_nb_server.erl", "src/riak_core_gen_server.erl", - "src/riak_core_stat_xform"]}. +{erl_first_files, [ + "src/gen_nb_server.erl", + "src/riak_core_gen_server.erl", + "src/riak_core_stat_xform.erl" +]}. {cover_enabled, true}. {erl_opts, [warnings_as_errors, debug_info]}. -{edoc_opts, [{preprocess, true}]}. +{edoc_opts, [ + {includes, ["include"]}, + {preprocess, true}, + {todo, true} +]}. {eunit_opts, [verbose]}. @@ -26,7 +54,9 @@ {basho_stats, {git, "https://github.com/OpenRiak/basho_stats.git", {branch, "openriak-3.4"}}} ]}. -{dialyzer, [{plt_apps, all_deps}]}. +{dialyzer, [ + {plt_apps, all_deps} +]}. {profiles, [ {test, [{deps, [{meck, {git, "https://github.com/OpenRiak/meck.git", {branch, "openriak-3.4"}}}]}, {erl_opts, [nowarn_export_all]}]}, diff --git a/src/riak_core.app.src b/src/riak_core.app.src index 6e14dbd6..e8562db1 100644 --- a/src/riak_core.app.src +++ b/src/riak_core.app.src @@ -9,7 +9,6 @@ {applications, [ kernel, stdlib, - sasl, crypto, ssl, runtime_tools, @@ -30,55 +29,55 @@ %% Cluster name {cluster_name, "default"}, - %% Default location for ring, cluster and other data files - {platform_data_dir, "data"}, + %% Default location for ring, cluster and other data files + {platform_data_dir, "data"}, - %% Default ring creation size. Make sure it is a power of 2, - %% e.g. 16, 32, 64, 128, 256, 512 etc - {ring_creation_size, 64}, + %% Default ring creation size. Make sure it is a power of 2, + %% e.g. 16, 32, 64, 128, 256, 512 etc + {ring_creation_size, 64}, - %% Default gossip interval (milliseconds) - {gossip_interval, 60000}, + %% Default gossip interval (milliseconds) + {gossip_interval, 60000}, - %% Target N value - {target_n_val, 4}, + %% Target N value + {target_n_val, 4}, %% Default claims functions - {wants_claim_fun, + {wants_claim_fun, {riak_core_membership_claim, default_wants_claim}}, - {choose_claim_fun, + {choose_claim_fun, {riak_core_membership_claim, default_choose_claim}}, - %% Vnode inactivity timeout (how often to check if fallback vnodes - %% should return their data) in ms. - {vnode_inactivity_timeout, 60000}, + %% Vnode inactivity timeout (how often to check if fallback vnodes + %% should return their data) in ms. + {vnode_inactivity_timeout, 60000}, - %% Number of VNodes allowed to do handoff concurrently. - {handoff_concurrency, 2}, + %% Number of VNodes allowed to do handoff concurrently. + {handoff_concurrency, 2}, - %% Disable Nagle on HTTP sockets - {disable_http_nagle, true}, + %% Disable Nagle on HTTP sockets + {disable_http_nagle, true}, - %% Handoff IP/port - {handoff_port, 8099}, - {handoff_ip, "0.0.0.0"}, + %% Handoff IP/port + {handoff_port, 8099}, + {handoff_ip, "0.0.0.0"}, - %% Disterl buffer sizes in bytes. - %% These sizes (3*128*1024 & 6*128*1024) were - %% derived from a limited amount of testing in a - %% 10GE environment, and may need tuning for your - %% network and workload. In particular they're likely - %% too small to be optimal for larger object sizes. - {dist_send_buf_size, 393216}, - {dist_recv_buf_size, 786432}, + %% Disterl buffer sizes in bytes. + %% These sizes (3*128*1024 & 6*128*1024) were + %% derived from a limited amount of testing in a + %% 10GE environment, and may need tuning for your + %% network and workload. In particular they're likely + %% too small to be optimal for larger object sizes. + {dist_send_buf_size, 393216}, + {dist_recv_buf_size, 786432}, - %% Exometer defaults - {exometer_defaults, - [ - {['_'], histogram, [{options, - [{histogram_module, exometer_slot_slide}, - {keep_high, 500}]} - ]} - ]} + %% Exometer defaults + {exometer_defaults, [ + {['_'], histogram, [ + {options, [ + {histogram_module, exometer_slot_slide}, + {keep_high, 500}]} + ]} ]} - ]}. + ]} +]}. diff --git a/src/riak_core_claimant.erl b/src/riak_core_claimant.erl index fe87e37d..e88eb58a 100644 --- a/src/riak_core_claimant.erl +++ b/src/riak_core_claimant.erl @@ -1,6 +1,7 @@ %% ------------------------------------------------------------------- %% -%% Copyright (c) 2012 Basho Technologies, Inc. All Rights Reserved. +%% Copyright (c) 2012-2015 Basho Technologies, Inc. +%% Copyright (c) 2023-2025 Workday, Inc. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file diff --git a/src/riak_core_coverage_plan.erl b/src/riak_core_coverage_plan.erl index aeebddd3..b25b1ffe 100644 --- a/src/riak_core_coverage_plan.erl +++ b/src/riak_core_coverage_plan.erl @@ -45,6 +45,10 @@ -type coverage_vnodes() :: [{index(), node()}]. -type vnode_filters() :: [{node(), [{index(), [index()]}]}]. -type coverage_plan() :: {coverage_vnodes(), vnode_filters()}. +-type position() :: non_neg_integer(). + % non_neg_integer() is 0 .. RingSize +-type vnode_covers() :: {position(), list(position())}. + %% =================================================================== %% Public API @@ -52,70 +56,103 @@ %% @doc Create a coverage plan to distribute work to a set %% covering VNodes around the ring. --spec create_plan(all | allup, pos_integer(), pos_integer(), - req_id(), atom()) -> - {error, term()} | coverage_plan(). +-spec create_plan( + all | allup, pos_integer(), pos_integer(), req_id(), atom() +) -> + {error, term()} | coverage_plan(). create_plan(VNodeSelector, NVal, PVC, ReqId, Service) -> {ok, CHBin} = riak_core_ring_manager:get_chash_bin(), PartitionCount = chashbin:num_partitions(CHBin), {ok, Ring} = riak_core_ring_manager:get_my_ring(), - %% Create a coverage plan with the requested primary - %% preference list VNode coverage. - %% Get a list of the VNodes owned by any unavailble nodes - Members = riak_core_ring:all_members(Ring), - NonCoverageNodes = [Node || Node <- Members, - riak_core_ring:get_member_meta(Ring, Node, participate_in_coverage) == false], - - DownVNodes = [Index || - {Index, _Node} - <- riak_core_apl:offline_owners(Service, CHBin, NonCoverageNodes)], - + NonCoverageNodes = + lists:filter( + fun(Node) -> + ParticipatingInCoverage = + riak_core_ring:get_member_meta( + Ring, + Node, + participate_in_coverage + ), + ParticipatingInCoverage == false + end, + riak_core_ring:all_members(Ring) + ), + % Nodes that have participate in coverage disabled, so are + % administratively blocked in participating + DownVNodes = + lists:map( + fun({Idx, _Node}) -> Idx end, + riak_core_apl:offline_owners(Service, CHBin, NonCoverageNodes) + ), + % Idx references for all primary partitions on nodes currently offline RingIndexInc = chash:ring_increment(PartitionCount), - UnavailableKeySpaces = [(DownVNode div RingIndexInc) || DownVNode <- DownVNodes], - %% Create function to map coverage keyspaces to - %% actual VNode indexes and determine which VNode - %% indexes should be filtered. - CoverageVNodeFun = - fun({Position, KeySpaces}, Acc) -> - %% Calculate the VNode index using the - %% ring position and the increment of - %% ring index values. - VNodeIndex = (Position rem PartitionCount) * RingIndexInc, - Node = chashbin:index_owner(VNodeIndex, CHBin), - CoverageVNode = {VNodeIndex, Node}, - case length(KeySpaces) < NVal of - true -> - %% Get the VNode index of each keyspace to - %% use to filter results from this VNode. - KeySpaceIndexes = [(((KeySpaceIndex+1) rem - PartitionCount) * RingIndexInc) || - KeySpaceIndex <- KeySpaces], - {CoverageVNode, [{VNodeIndex, KeySpaceIndexes} | Acc]}; - false -> - {CoverageVNode, Acc} - end - end, + % See https://github.com/OpenRiak/riak_core/issues/10 + % It easier to handle the vnodes as a list of small integers 0..(RS-1) + % But all Idx's are stored in the ring with lots of 0s + % + % Convert the list of DownVnodes into a list of small integers (i.e. + % positions) by dividing by the ring_increment + UnavailableKeySpaces = + lists:map( + fun(Idx) -> Idx div RingIndexInc end, + DownVNodes + ), + % Unavailable indexes as small integers - CoveragePlanFun = + CoverageResult = case application:get_env(riak_core, legacy_coverage_planner, false) of true -> % Safety net for refactoring, we can still go back to old - % function if necessary. Note 35 x performance degradation + % function if necessary. Note 100 x performance degradation % with this function with ring_size of 1024 - fun find_coverage/5; + find_coverage( + ReqId, + NVal, + PartitionCount, + UnavailableKeySpaces, + lists:min([PVC, NVal]) + ); false -> - fun initiate_plan/5 + initiate_plan( + ReqId, + NVal, + PartitionCount, + UnavailableKeySpaces, + lists:min([PVC, NVal]) + ) end, - %% The ReqId value serves as a tiebreaker in the - %% compare_next_vnode function and is used to distribute - %% work to different sets of VNodes. - CoverageResult = - CoveragePlanFun(ReqId, - NVal, - PartitionCount, - UnavailableKeySpaces, - lists:min([PVC, NVal])), + CoverageVNodeFun = + fun({Position, KeySpaces}, Acc) -> + VNodeIndex = expand_index(Position, PartitionCount, RingIndexInc), + Node = chashbin:index_owner(VNodeIndex, CHBin), + CoverageVNode = {VNodeIndex, Node}, + case length(KeySpaces) of + L when L < NVal -> + KeySpaceIndexes = + lists:map( + fun(KSI) -> + expand_index( + KSI + 1, + PartitionCount, + RingIndexInc + ) + end, + KeySpaces + ), + {CoverageVNode, [{VNodeIndex, KeySpaceIndexes} | Acc]}; + _ -> + {CoverageVNode, Acc} + end + end, + % A function to convert the positions (the small integers) into + % index/node pairings, and if the vnode is not required to cover all + % n_val partitions add the vnode to a list of filtered vnodes to return + % partial results + % + % There is some duplicate effort as chashbin:index_owner/2 will revert + % the index back to a position. + case CoverageResult of {ok, CoveragePlan} -> %% Assemble the data structures required for @@ -136,7 +173,18 @@ create_plan(VNodeSelector, NVal, PVC, ReqId, Service) -> %% Internal functions %% ==================================================================== --type vnode_covers() :: {non_neg_integer(), list(non_neg_integer())}. +%% Note +%% There is an issue with ELS if there is `rem` within an anonymous function, +%% and so some integer manipulation helper functions added to ease syntax +%% highlighting issues. +-spec expand_index(position(), pos_integer(), pos_integer()) -> index(). +expand_index(Position, RingSize, RingIncrement) -> + (Position rem RingSize) * RingIncrement. + +-spec lookback_position(pos_integer(), integer()) -> non_neg_integer(). +lookback_position(PartitionCount, Position) -> + % Handle negative positions when looking back. + (PartitionCount + Position) rem PartitionCount. %% @doc Produce a coverage plan %% The coverage plan should include all partitions at least PVC times @@ -148,15 +196,19 @@ create_plan(VNodeSelector, NVal, PVC, ReqId, Service) -> %% UnavailableVnodes - any primary vnodes not available, as either the node is %% down, or set not to participate_in_coverage %% PVC - Primary Vnode Count, in effect the r value for the query --spec initiate_plan(non_neg_integer(), - pos_integer(), - pos_integer(), - list(non_neg_integer()), - pos_integer()) -> - {ok, list(vnode_covers())} | - {insufficient_vnodes_available, - list(non_neg_integer()), - list(vnode_covers())}. +-spec initiate_plan( + non_neg_integer(), + pos_integer(), + pos_integer(), + list(position()), + pos_integer()) +-> + {ok, list(vnode_covers())} | + { + insufficient_vnodes_available, + any(), + list(vnode_covers()) + }. initiate_plan(ReqId, NVal, PartitionCount, UnavailableVnodes, PVC) -> % Order the vnodes for the fold. Will number each vnode in turn between % 0 and NVal - 1. Then sort by this Offset, so that by default we visit @@ -175,81 +227,155 @@ initiate_plan(ReqId, NVal, PartitionCount, UnavailableVnodes, PVC) -> % to always start at the front of the ring then those vnodes that cover the % tail of the ring will be involved in a disproportionate number of % queries. - {L1, L2} = - lists:split(ReqId rem PartitionCount, - lists:seq(0, PartitionCount - 1)), - % Use an array to hold a list for each offset, before flattening the array - % back to a list to rejoin together - A0 = array:new(NVal, {default, []}), - {A1, _} = - lists:foldl( - fun(I, {A, Offset}) -> - {array:set(Offset, [I|array:get(Offset, A)], A), - (Offset + 1) rem NVal} - end, - {A0, 0}, - L2 ++ L1 - ), - OrderedVnodes = lists:flatten(array:to_list(A1)), + ShuffledPositions = + get_shuffled_positions(ReqId rem PartitionCount, PartitionCount, NVal), % Setup an array for tracking which partition has "Wants" left, starting % with a value of PVC PartitionWants = array:new(PartitionCount, {default, PVC}), Countdown = PartitionCount * PVC, - % Subtract any Unavailable vnodes. Must only assign available primary - % vnodes a role in the coverage plan - AvailableVnodes = lists:subtract(OrderedVnodes, UnavailableVnodes), - - develop_plan(AvailableVnodes, NVal, PartitionWants, Countdown, []). - + % Subtract any Unavailable vnodes (actually positions). Must only assign + % available primary vnodes a role in the coverage plan + AvailablePositions = lists:subtract(ShuffledPositions, UnavailableVnodes), + + develop_plan(AvailablePositions, NVal, PartitionWants, Countdown, []). + +%% @doc +%% Order the vnodes to reduce the time to calculate the plan. +%% the vnodes are a list of positions 0 .. RingSize - 1 +%% For a given NVal, the most efficient plan will be every NValth position +%% e.g. if RingSize = 16, NVal = 3 +%% 0, 2, 5, 8, 11, 14 +%% However if we take the positions 0 .. RingSize - 1 as the input before +%% selecting then a third of the NVals would do almost all of the work - it is +%% required to balance evenly around the cluster. +%% It would be simpler to have just NVal plans, one for each offset - however +%% there is always a remainder when RingSize rem NVal is not 0. The filter +%% vnodes that run a partial query to pick up the remainder in this case would +%% be concentrated towards the end of the sequence, and this would be a further +%% imbalance. +%% So instead the initial position sequence 0 .. RingSize - 1 is split at a +%% random point - to full distribute work across all possible plans. +%% +%% On a node PartitionCount is constant, NVal will normally be fixed (or have +%% very limited variation), so the Split is the only variable. +%% +%% Typically there will be RingSize variations in this calculation, so rather +%% than repeat the calculation each time, each variation is stored as a +%% persistent term. +-spec get_shuffled_positions( + non_neg_integer(), pos_integer(), pos_integer()) +-> + list(position()). +get_shuffled_positions(Split, PartitionCount, NVal) -> + CachedResult = + persistent_term:get({?MODULE, Split, PartitionCount, NVal}, undefined), + case CachedResult of + undefined -> + {L1, L2} = + lists:split(Split, lists:seq(0, PartitionCount - 1)), + % Split the list and loop around at a mark. This means that + % the filtered vnodes will vary position. + A0 = array:new(NVal, {default, []}), + {A1, _} = + lists:foldl( + fun(I, {A, Offset}) -> + CurrentL = + case + array:get(Offset, A) of L when is_list(L) -> L + end, + { + array:set(Offset, [I|CurrentL], A), + (Offset + 1) rem NVal + } + end, + {A0, 0}, + L2 ++ L1 + ), + Vnodes = lists:flatten(array:to_list(A1)), + persistent_term:put( + {?MODULE, Split, PartitionCount, NVal}, + Vnodes + ), + Vnodes; + CachedResult -> + CachedResult + end. -develop_plan(_UnusedVnodes, _NVal, _PartitionWants, 0, VnodeCovers) -> - % Use the countdown to know when to stop, rather than having the cost of - % checking each entry in the PartitionWants array each loop - {ok, VnodeCovers}; +-spec develop_plan( + list(position()), + pos_integer(), + array:array(non_neg_integer()), + non_neg_integer(), + % A countdown of covered partitions, to stop the loop without + % the need to recount the covered partitions each loop + % Countdown should start at NVal * PartitionCount + list(vnode_covers())) +-> + {ok, list(vnode_covers())} | + {insufficient_vnodes_available, any(), list(vnode_covers())}. +develop_plan(_UnusedPositions, _NVal, _PartitionWants, 0, VnodeCovers) -> + {ok, lists:sort(VnodeCovers)}; develop_plan([], _NVal, _PartitionWants, _N, VnodeCovers) -> - % The previous function coverage_plan/7 returns "KeySpaces" as the second - % element, which is then ignored - so we don't bother calculating this here - {insufficient_vnodes_available, [], VnodeCovers}; -develop_plan([HeadVnode|RestVnodes], NVal, - PartitionWants, PartitionCountdown, - VnodeCovers) -> + { + insufficient_vnodes_available, + [], + % The previous function coverage_plan/7 returns "KeySpaces" as the + % second element, which is then ignored + % - so we don't bother calculating this here + lists:sort(VnodeCovers) + }; +develop_plan( + [HeadPos|RestPositions], + NVal, + PartitionWants, + PartitionCountdown, + VnodeCovers +) -> PartitionCount = array:size(PartitionWants), % Need to find what partitions are covered by this vnode LookBackFun = - fun(I) -> (PartitionCount + HeadVnode - I) rem PartitionCount end, + fun(I) -> lookback_position(PartitionCount, HeadPos - I) end, PartsCoveredByHeadNode = lists:sort(lists:map(LookBackFun, lists:seq(1, NVal))), % For these partitions covered by the vnode, are there any partitions with % non-zero wants PartsCoveredAndWanted = - lists:filter(fun(P) -> array:get(P, PartitionWants) > 0 end, - PartsCoveredByHeadNode), + lists:filter( + fun(P) -> array:get(P, PartitionWants) > 0 end, + PartsCoveredByHeadNode + ), % If there are partitions that are covered by the vnode and have wants, % then we should include this vnode in the coverage plan for these % partitions. Otherwise, skip the vnode. case length(PartsCoveredAndWanted) of L when L > 0 -> - % Add the vnode to the coverage plan - VnodeCovers0 = - lists:sort([{HeadVnode, PartsCoveredAndWanted}|VnodeCovers]), - % Update the wants, for each partition that has been added to the - % coverage plan - UpdateWantsFun = - fun(P, PWA) -> array:set(P, array:get(P, PWA) - 1, PWA) end, + % Reduce the wants PartitionWants0 = - lists:foldl(UpdateWantsFun, - PartitionWants, - PartsCoveredAndWanted), + lists:foldl( + fun(P, PWA) -> + array:set(P, array:get(P, PWA) - 1, PWA) + end, + PartitionWants, + PartsCoveredAndWanted + ), % Now loop, to find use of the remaining vnodes - develop_plan(RestVnodes, NVal, - PartitionWants0, PartitionCountdown - L, - VnodeCovers0); + develop_plan( + RestPositions, + NVal, + PartitionWants0, + PartitionCountdown - L, + [{HeadPos, PartsCoveredAndWanted}|VnodeCovers] + ); _L -> - develop_plan(RestVnodes, NVal, - PartitionWants, PartitionCountdown, - VnodeCovers) + develop_plan( + RestPositions, + NVal, + PartitionWants, + PartitionCountdown, + VnodeCovers + ) end. -spec find_coverage(non_neg_integer(), @@ -674,21 +800,60 @@ all_refactor_1024_ring_tester() -> PC1 = lists:foldl( fun({_I, L}, Acc) -> - lists:foldl(fun(P, IA) -> - array:set(P, - array:get(P, IA) + 1, - IA) - end, - Acc, - L) + lists:foldl( + fun(P, IA) -> + array:set(P, array:get(P, IA) + 1, IA) + end, + Acc, + L + ) end, PC, - VnodeCovers), + VnodeCovers + ), lists:foreach(fun(C) -> ?assertEqual(PVC, C) end, array:to_list(PC1)) end, lists:foreach(fun(I) -> TestFun(I) end, lists:seq(0, 1023)). +timed_ring_test_() -> + {timeout, 1200, fun timed_ring_tester/0}. + +timed_ring_tester() -> + io:format(user, "Timing tests:~n", []), + timed_ring_test(512, 1000), + timed_ring_test(1024, 1000), + timed_ring_test(2048, 300), + timed_ring_test(4096, 200), + timed_ring_test(8092, 200). + +timed_ring_test(RingSize, TestCount) -> + TT = + lists:sum( + lists:map( + fun(_I) -> + {T, ok} = + timer:tc( + fun() -> + bare_ring_tester(RingSize, fun initiate_plan/5) + end + ), + T + end, + lists:seq(1, TestCount) + ) + ), + io:format( + user, + "Average time to get coverage with ring_size=~w ~w~n", + [RingSize, TT div TestCount] + ). + +refactor_8192ring_test() -> + ring_tester(8192, fun initiate_plan/5). + +refactor_4096ring_test() -> + ring_tester(4096, fun initiate_plan/5). refactor_2048ring_test() -> ring_tester(2048, fun initiate_plan/5). @@ -742,16 +907,37 @@ compare_tester(RingSize) -> ?assertMatch(true, RFC =< (PFC + 1)). +bare_ring_tester(PartitionCount, CoverageFun) -> + ReqID = 100 * rand:uniform(10), + NVal = 3, + UnavailableKeySpaces = [], + PVC = 1, + {ok, _VnodeCovers0} = + CoverageFun( + ReqID, + NVal, + PartitionCount, + UnavailableKeySpaces, + PVC + ), + ok. + ring_tester(PartitionCount, CoverageFun) -> - ring_tester(PartitionCount, CoverageFun, 0). + ring_tester(PartitionCount, CoverageFun, 100 * rand:uniform(10)). ring_tester(PartitionCount, CoverageFun, ReqId) -> NVal = 3, UnavailableKeySpaces = [], PVC = 1, {Vnodes, CoveredKeySpaces} = - ring_tester(PartitionCount, CoverageFun, - ReqId, NVal, UnavailableKeySpaces, PVC), + ring_tester( + PartitionCount, + CoverageFun, + ReqId, + NVal, + UnavailableKeySpaces, + PVC + ), ExpVnodeCount = (PartitionCount div NVal) + 2, KeySpaces = lists:seq(0, PartitionCount - 1), ?assertMatch(KeySpaces, CoveredKeySpaces), @@ -760,9 +946,13 @@ ring_tester(PartitionCount, CoverageFun, ReqId) -> ring_tester(PartitionCount, CoverageFun, Offset, NVal, UnavailableKeySpaces, PVC) -> {ok, VnodeCovers0} = - CoverageFun(Offset, NVal, PartitionCount, - UnavailableKeySpaces, - PVC), + CoverageFun( + Offset, + NVal, + PartitionCount, + UnavailableKeySpaces, + PVC + ), AccFun = fun({A, L}, {VnodeAcc, CoverAcc}) -> {[A|VnodeAcc], CoverAcc ++ L} end, {Vnodes, Coverage} = lists:foldl(AccFun, {[], []}, VnodeCovers0), @@ -776,20 +966,26 @@ nonstandardring_tester(PartitionCount, CoverageFun) -> UnavailableKeySpaces = [OneDownVnode], PVC = 2, {ok, VnodeCovers} = - CoverageFun(Offset, NVal, PartitionCount, - UnavailableKeySpaces, - PVC), + CoverageFun( + Offset, + NVal, + PartitionCount, + UnavailableKeySpaces, + PVC + ), PC = array:new(PartitionCount, {default, 0}), PC1 = lists:foldl( fun({I, L}, Acc) -> ?assertNotEqual(OneDownVnode, I), - lists:foldl(fun(P, IA) -> - array:set(P, array:get(P, IA) + 1, IA) - end, - Acc, - L) + lists:foldl( + fun(P, IA) -> + array:set(P, array:get(P, IA) + 1, IA) + end, + Acc, + L + ) end, PC, VnodeCovers), diff --git a/src/riak_core_gen_server.erl b/src/riak_core_gen_server.erl index a98be82f..5e595a41 100644 --- a/src/riak_core_gen_server.erl +++ b/src/riak_core_gen_server.erl @@ -58,6 +58,9 @@ %% $Id$ %% -module(riak_core_gen_server). +%% +%% TODO: Either re-implement this module from current gen_server or eliminate it. +%% %%% --------------------------------------------------- %%% @@ -145,7 +148,7 @@ %% Internal exports -export([init_it/6, print_event/3]). --import(error_logger, [format/2]). +-include_lib("kernel/include/logger.hrl"). %%%========================================================================= %%% API @@ -895,34 +898,37 @@ terminate(Reason, Name, Msg, Mod, State, Debug) -> end end. -error_info(_Reason, application_controller, _Msg, _State, _Debug) -> - %% OTP-5811 Don't send an error report if it's the system process - %% application_controller which is terminating - let init take care - %% of it instead - ok; error_info(Reason, Name, Msg, State, Debug) -> - Reason1 = - case Reason of - {undef,[{M,F,A}|MFAs]} -> - case code:is_loaded(M) of - false -> - {'module could not be loaded',[{M,F,A}|MFAs]}; - _ -> - case erlang:function_exported(M, F, length(A)) of - true -> - Reason; - false -> - {'function not exported',[{M,F,A}|MFAs]} - end - end; - _ -> - Reason - end, - format("** Generic server ~p terminating \n" - "** Last message in was ~p~n" - "** When Server state == ~p~n" - "** Reason for termination == ~n** ~p~n", - [Name, Msg, State, Reason1]), + Reason1 = case Reason of + {undef, [{M, F, A} | _] = MFAs} -> + case code:is_loaded(M) of + false -> + {'module could not be loaded', MFAs}; + _ -> + case erlang:function_exported(M, F, length(A)) of + true -> + Reason; + _ -> + {'function not exported', MFAs} + end + end; + _ -> + Reason + end, + %% Report *similar to* current gen_server + Report = #{ + label => {?MODULE, terminate}, + message => <>, + name => Name, + last_message => Msg, + state => State, + reason => Reason1 + }, + Meta = #{ + domain => [riak, core], + error_logger => #{tag => error} + }, + ?LOG_ERROR(Report, Meta), sys:print_log(Debug), ok. @@ -958,8 +964,9 @@ dbg_options(Name, Opts) -> dbg_opts(Name, Opts) -> case catch sys:debug_options(Opts) of {'EXIT',_} -> - format("~p: ignoring erroneous debug options - ~p~n", - [Name, Opts]), + ?LOG_INFO( + "~0tp: ignoring erroneous debug options - ~0tp", + [Name, Opts]), []; Dbg -> Dbg diff --git a/src/riak_core_handoff_manager.erl b/src/riak_core_handoff_manager.erl index d1d7dff0..c38ad825 100644 --- a/src/riak_core_handoff_manager.erl +++ b/src/riak_core_handoff_manager.erl @@ -11,6 +11,8 @@ %% KIND, either express or implied. See the License for the %% specific language governing permissions and limitations %% under the License. +%% +%% ------------------------------------------------------------------- %% Copyright (c) 2007-2012 Basho Technologies, Inc. All Rights Reserved. -module(riak_core_handoff_manager). @@ -306,8 +308,8 @@ handle_info({'DOWN', Ref, process, _Pid, Reason}, State=#state{handoffs=HS}) -> case ShouldILog of true -> ?LOG_INFO( - "An ~w handoff of partition ~w ~w " - "was terminated for reason: ~w", + "An ~0tp handoff of partition ~0tp ~0tp " + "was terminated for reason: ~0tp", [Dir, M, I, Reason] ); false -> @@ -316,8 +318,8 @@ handle_info({'DOWN', Ref, process, _Pid, Reason}, State=#state{handoffs=HS}) -> true; _ -> ?LOG_ERROR( - "An ~w handoff of partition ~w ~w " - "was terminated for reason: ~w", + "An ~0tp handoff of partition ~0tp ~0tp " + "was terminated for reason: ~0tp", [Dir, M, I, Reason] ), true @@ -659,13 +661,16 @@ kill_xfer_i(ModSrcTarget, Reason, HS) -> src_node=SrcNode, transport_pid=TP } = Xfer, - Msg = "~p transfer of ~p from ~p ~p to ~p ~p killed for reason ~p", case Type of undefined -> ok; _ -> - ?LOG_INFO(Msg, [Type, Mod, SrcNode, SrcPartition, - TargetNode, TargetPartition, Reason]) + ?LOG_INFO( + "~0tp transfer of ~0tp from ~0tp ~0tp to" + " ~0tp ~0tp killed for reason ~tp", + [Type, Mod, SrcNode, SrcPartition, + TargetNode, TargetPartition, Reason] + ) end, exit(TP, {kill_xfer, Reason}), kill_xfer_i(ModSrcTarget, Reason, HS2) diff --git a/src/riak_core_ssl_util.erl b/src/riak_core_ssl_util.erl index aaac282f..094c2d99 100644 --- a/src/riak_core_ssl_util.erl +++ b/src/riak_core_ssl_util.erl @@ -48,7 +48,7 @@ ssl_handshake(Socket, SslOpts) -> openssl_suite(Cipher) -> ssl_cipher_format:suite_openssl_str_to_map(Cipher). openssl_suite_name(Cipher) -> - ssl_cipher_format:suite_map_to_openssl_str(ssl_cipher_format:suite_bin_to_map(Cipher)). + ssl_cipher_format:suite_map_to_openssl_str(Cipher). maybe_use_ssl(App) -> diff --git a/src/riak_core_stat.erl b/src/riak_core_stat.erl index 87ce81e5..70031cc0 100644 --- a/src/riak_core_stat.erl +++ b/src/riak_core_stat.erl @@ -1,6 +1,7 @@ %% ------------------------------------------------------------------- %% -%% Copyright (c) 2007-2011 Basho Technologies, Inc. All Rights Reserved. +%% Copyright (c) 2007-2011 Basho Technologies, Inc. +%% Copyright (c) 2018-2024 Workday, Inc. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file @@ -28,16 +29,26 @@ -behaviour(gen_server). %% API --export([start_link/0, get_stats/0, get_stats/1, update/1, - register_stats/0, vnodeq_stats/0, - register_stats/2, - register_vnode_stats/3, unregister_vnode_stats/2, - vnodeq_stats/1, - prefix/0]). +-export([ + get_stats/0, get_stats/1, + prefix/0, + register_stats/0, register_stats/2, + register_vnode_stats/3, unregister_vnode_stats/2, + start_link/0, + update/1 +]). + +%% exometer callbacks +-export([ + vnodeq_stats/0, vnodeq_stats/1 +]). %% gen_server callbacks --export([init/1, handle_call/3, handle_cast/2, handle_info/2, - terminate/2, code_change/3]). +-export([ + init/1, + handle_call/3, + handle_cast/2 +]). -include_lib("kernel/include/logger.hrl"). @@ -155,15 +166,6 @@ handle_cast({update, Arg}, State) -> handle_cast(_Req, State) -> {noreply, State}. -handle_info(_Info, State) -> - {noreply, State}. - -terminate(_Reason, _State) -> - ok. - -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - exometer_update(Name, Value) -> case exometer:update(Name, Value) of @@ -209,11 +211,11 @@ stats() -> nwp_stats() -> PoolNames = [vnode_pool, unregistered] ++ riak_core_node_worker_pool:pools(), - + [nwp_stat(Pool) || Pool <- PoolNames] ++ - + [nwpqt_stat(Pool) || Pool <- PoolNames] ++ - + [nwpwt_stat(Pool) || Pool <- PoolNames]. nwp_stat(Pool) -> @@ -232,25 +234,54 @@ nwpwt_stat(Pool) -> system_stats() -> [ - {cpu_stats, cpu, [{sample_interval, 5000}], [{nprocs, cpu_nprocs}, - {avg1 , cpu_avg1}, - {avg5 , cpu_avg5}, - {avg15 , cpu_avg15}]}, - {mem_stats, {function, memsup, get_memory_data, [], match, {total, allocated, '_'}}, - [], [{total, mem_total}, - {allocated, mem_allocated}]}, - {memory_stats, {function, erlang, memory, [], proplist, [total, processes, processes_used, - system, atom, atom_used, binary, - code, ets]}, - [], [{total , memory_total}, - {processes , memory_processes}, - {processes_used, memory_processes_used}, - {system , memory_system}, - {atom , memory_atom}, - {atom_used , memory_atom_used}, - {binary , memory_binary}, - {code , memory_code}, - {ets , memory_ets}]} + {cpu_stats, cpu, [{sample_interval, 5000}], [ + {nprocs, cpu_nprocs}, + {avg1, cpu_avg1}, + {avg5, cpu_avg5}, + {avg15, cpu_avg15} + ]}, + {mem_stats, + {function, memsup, get_memory_data, [], + match, {total, allocated, '_'} + }, [], [ + {total, mem_total}, + {allocated, mem_allocated} + ]}, + {memory_stats, + {function, erlang, memory, [], proplist, [ + total, processes, processes_used, system, + atom, atom_used, binary, code, ets + ]}, [], [ + {total, memory_total}, + {processes, memory_processes}, + {processes_used, memory_processes_used}, + {system, memory_system}, + {atom, memory_atom}, + {atom_used, memory_atom_used}, + {binary, memory_binary}, + {code, memory_code}, + {ets, memory_ets} + ]}, + {vm_stats, + {function, riak_core_vm_mon, vm_stats, [], match, { + atom_lim, atom_cnt, atom_pct, + ets_lim, ets_cnt, ets_pct, + port_lim, port_cnt, port_pct, + proc_lim, proc_cnt, proc_pct + }}, [], [ + {atom_lim, vm_atom_limit}, + {atom_cnt, vm_atom_count}, + {atom_pct, vm_atom_percent}, + {ets_lim, vm_ets_limit}, + {ets_cnt, vm_ets_count}, + {ets_pct, vm_ets_percent}, + {port_lim, vm_port_limit}, + {port_cnt, vm_port_count}, + {port_pct, vm_port_percent}, + {proc_lim, vm_proc_limit}, + {proc_cnt, vm_proc_count}, + {proc_pct, vm_proc_percent} + ]} ]. %% Provide aggregate stats for vnode queues. Compute instantaneously for now, diff --git a/src/riak_core_sysmon_handler.erl b/src/riak_core_sysmon_handler.erl index 1eefdfb6..d8ecc495 100644 --- a/src/riak_core_sysmon_handler.erl +++ b/src/riak_core_sysmon_handler.erl @@ -13,7 +13,8 @@ %% KIND, either express or implied. See the License for the %% specific language governing permissions and limitations %% under the License. - +%% +%% ------------------------------------------------------------------- %% @doc A custom event handler to the `riak_sysmon' application's %% `system_monitor' event manager. %% @@ -91,8 +92,10 @@ handle_event({monitor, PidOrPort, Type, Info}, State=#state{timer_ref=TimerRef}) %% Reset the inactivity timeout NewTimerRef = reset_timer(TimerRef), {Fmt, Args} = format_pretty_proc_or_port_info(PidOrPort, almost_current_function), - ?LOG_INFO("monitor ~w ~w "++ Fmt ++ " ~w", - [Type, PidOrPort] ++ Args ++ [Info]), + ?LOG_INFO( + "monitor ~0tp ~0tp "++ Fmt ++ " ~0tp", + [Type, PidOrPort] ++ Args ++ [Info] + ), {ok, State#state{timer_ref=NewTimerRef}}; handle_event(Event, State=#state{timer_ref=TimerRef}) -> NewTimerRef = reset_timer(TimerRef), @@ -182,7 +185,7 @@ format_pretty_proc_or_port_info(PidOrPort, Acf) -> Res end catch Class:Reason:Stacktrace -> - {"Pid ~w, ~W ~W at ~w\n", + {"Pid ~0tp, ~0tP ~0tP at ~tp", [PidOrPort, Class, 20, Reason, 20, Stacktrace]} end. diff --git a/src/riak_core_test_util.erl b/src/riak_core_test_util.erl index 7c5220c2..ff821783 100644 --- a/src/riak_core_test_util.erl +++ b/src/riak_core_test_util.erl @@ -1,8 +1,8 @@ +%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- %% ------------------------------------------------------------------- %% -%% riak_test_util: utilities for test scripts -%% -%% Copyright (c) 2007-2010 Basho Technologies, Inc. All Rights Reserved. +%% Copyright (c) 2007-2014 Basho Technologies, Inc. +%% Copyright (c) 2025 Workday, Inc. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file @@ -19,17 +19,231 @@ %% under the License. %% %% ------------------------------------------------------------------- +%% +%% @doc Utilities for test scripts. +%% +-module(riak_core_test_util). -%% @doc utilities for test scripts +%% Public API for use from other apps +-export([ + ensure_no_file/1, + get_test_dir/1, get_test_dir/2, + logger_filesync/0, + logger_redirect/1, logger_redirect/2, + logger_restore/0, + logger_silence/0 +]). --module(riak_core_test_util). +%% Private API for use only from riak_core +-ifdef(TEST). +-export([ + fake_ring/2, + setup_mockring1/0, + stop_pid/1, + wait_for_pid/1 +]). +-endif. % TEST + +-include_lib("stdlib/include/assert.hrl"). + +-type abs_path() :: nonempty_string(). +-type fs_path() :: abs_path() | rel_path(). +-type rel_path() :: file:name(). +-type test_name() :: atom() | nonempty_string(). + +%% Test data root when not running under Rebar3. +-define(TEST_DATA_ROOT, "/tmp"). +-define(TEST_DATA_DIR, "testdata"). + +%% Persistent term keys +-define(LOGSTATE_PKEY, {?MODULE, logger_state}). +-define(TESTROOT_PKEY, {?MODULE, testdata_root}). +-define(TESTDATA_PKEY(T), {?MODULE, testdata_dir, T}). + +-define(LOG_FILE(FilePath), #{ + type => file, + file => FilePath, + file_check => 50, + filesync_repeat_interval => 100 +}). +-define(LOG_FMT, {logger_formatter, #{ + legacy_header => false, + single_line => false, + time_designator => $\s, + template => [ + time, " [", level, "] ", {pid, [pid, "@"], []}, + {mfa, [mfa, ":"], []}, {line, [line, ":"], []}, + " ", msg, "\n" + ]} +}). +-define(LOG_CFG(FilePath), #{ + config => ?LOG_FILE(FilePath), + formatter => ?LOG_FMT +}). +-define(LOG_CFG(FilePath, Level), #{ + config => ?LOG_FILE(FilePath), + formatter => ?LOG_FMT, + level => Level +}). + +-spec ensure_no_file(Path :: fs_path()) -> ok. +%% @doc Ensures that the specified file/directory does not exist, +%% deleting it unconditionally if it does. +%% Note that this function should only be used on paths within transient +%% test data, as it does not report permission errors. +ensure_no_file(Path) -> + _ = filelib:is_file(Path) =:= false orelse + os:cmd(io_lib:format("/bin/rm -rf '~ts'", [Path])), + ok. + +-spec get_test_dir(TestName :: test_name()) -> abs_path(). +%% @doc Ensures the test directory for TestName exists. +%% @returns The Absolute path of the directory to use for the specified TestName. +%% @equiv get_test_dir(TestName, false) +get_test_dir(TestName) -> + get_test_dir(TestName, false). + +-spec get_test_dir( + TestName :: test_name(), EnsureEmpty :: boolean()) + -> abs_path(). +%% @doc Ensures the test directory for TestName exists and, if EnsureEmpty +%% is true, that it is empty. +%% @returns The Absolute path of the directory to use for the specified TestName. +get_test_dir(TestName, EnsureEmpty) -> + PKey = ?TESTDATA_PKEY(TestName), + TestDir = case persistent_term:get(PKey, undefined) of + undefined -> + Path = filename:join(get_testdata_path(), TestName), + persistent_term:put(PKey, Path), + Path; + Val -> + Val + end, + EnsureEmpty =/= true orelse ensure_no_file(TestDir), + ?assertMatch(ok, filelib:ensure_dir(filename:join(TestDir, "x"))), + TestDir. + +-spec get_testdata_path() -> abs_path(). +%% @hidden Gets the root under which all test-specific data directories live. +%% When running under Rebar3, the path is under the '_build' directory. +get_testdata_path() -> + PKey = ?TESTROOT_PKEY, + case persistent_term:get(PKey, undefined) of + undefined -> + {ok, CWD} = file:get_cwd(), + Build = filename:join(CWD, "_build"), + DataDir = case filelib:is_dir(Build) of + true -> + filename:join([Build, "test", ?TEST_DATA_DIR]); + _ -> + Default = filename:join(?TEST_DATA_ROOT, ?TEST_DATA_DIR), + io:format(user, + "~n*** ~ts not present~n*** Using ~ts~n", + [Build, Default]), + Default + end, + persistent_term:put(PKey, DataDir), + DataDir; + TDVal -> + TDVal + end. + +-spec logger_filesync() -> ok | {error, term()}. +%% @doc If the `default' log handler is directed to a file, invokes the +%% handler module's `filesync/1' function. +logger_filesync() -> + Handler = default, + case logger:get_handler_config(Handler) of + {ok, #{config := #{file := _}, module := Mod}} -> + ?assertMatch(ok, Mod:filesync(Handler)); + _ -> + ok + end. + +-spec logger_redirect(TestName :: test_name()) -> abs_path(). +%% @doc Redirects the `default' log handler to a file. +%% The file is named `TestName.log' in the directory returned by +%% {@link get_test_dir/1. get_test_dir(TestName)}. +logger_redirect(TestName) -> + logger_redirect(TestName, [TestName, ".log"]). + +-spec logger_redirect(TestName :: test_name(), LogFile :: rel_path()) + -> abs_path(). +%% @doc Redirects the `default' log handler to a file. +%% The file is named `LogFile' in the directory returned by +%% {@link get_test_dir/1. get_test_dir(TestName)}. +logger_redirect(TestName, LogFile) -> + TestDir = get_test_dir(TestName), + LogPath = filename:join(TestDir, LogFile), + Handler = default, + case logger:get_handler_config(Handler) of + {ok, #{config := #{file := LogPath}}} -> + %% already redirected to the target file + ok; + {ok, #{level := Level} = OldConf} -> + logger_replace(Handler, ?LOG_CFG(LogPath, Level), OldConf); + {error, {not_found, _Handler} = NFRec} -> + logger_replace(Handler, ?LOG_CFG(LogPath), NFRec) + end, + LogPath. + +-spec logger_restore() -> ok. +%% @doc Restores the `default' log handler that was replaced by +%% {@link logger_redirect/2}. +logger_restore() -> + case persistent_term:get(?LOGSTATE_PKEY, undefined) of + undefined -> + ok; + #{config := #{file := _}, id := ID, module := Mod} = OldConf -> + ?assertMatch(ok, Mod:filesync(ID)), + persistent_term:erase(?LOGSTATE_PKEY), + logger_replace(ID, OldConf, false); + #{id := ID} = OldConf -> + persistent_term:erase(?LOGSTATE_PKEY), + logger_replace(ID, OldConf, false); + {not_found, ID} -> + persistent_term:erase(?LOGSTATE_PKEY), + ?assertMatch(ok, logger:remove_handler(ID)) + end. + +-spec logger_silence() -> logger:level() | all | none. +%% @doc Silences the `default' log handler. +%% The handler's log level can be restored by invoking +%% ``` +%% logger:set_handler_config(default, level, Level) +%% ''' +%% where `Level' is the value returned from this function. +%% @returns The previous level, or `none' if no `default' handler is present. +logger_silence() -> + Handler = default, + case logger:get_handler_config(Handler) of + {ok, #{level := none = None}} -> + None; + {ok, #{level := Level}} -> + ?assertMatch(ok, logger:set_handler_config(Handler, level, none)), + Level; + _ -> + none + end. + +-spec logger_replace( + HandlerID :: logger:handler_id(), + HConfig :: logger:handler_config(), + StoreRec :: false | logger:handler_config() | tuple()) + -> ok. +%% @hidden The handler always needs to be replaced when changing +%% the output path. +%% If StoreRec is not `false' AND no prior state is stored, StoreRec +%% becomes the restorable state. +logger_replace(ID, HConfig, false) -> + ?assertMatch(ok, logger:remove_handler(ID)), + ?assertMatch(ok, logger:add_handler(ID, logger_std_h, HConfig)); +logger_replace(ID, HConfig, StoreRec) -> + persistent_term:get(?LOGSTATE_PKEY, undefined) =/= undefined + orelse persistent_term:put(?LOGSTATE_PKEY, StoreRec), + logger_replace(ID, HConfig, false). -ifdef(TEST). --export([setup_mockring1/0, - fake_ring/2, - stop_pid/1, - wait_for_pid/1]). --include_lib("eunit/include/eunit.hrl"). stop_pid(Other) when not is_pid(Other) -> ok; @@ -48,7 +262,6 @@ wait_for_pid(Pid) -> {error, didnotexit} end. - setup_mockring1() -> % requires a running riak_core_ring_manager, in test-mode is ok Ring0 = riak_core_ring:fresh(16,node()), diff --git a/src/riak_core_util.erl b/src/riak_core_util.erl index 82b9b766..018d742b 100644 --- a/src/riak_core_util.erl +++ b/src/riak_core_util.erl @@ -1,1077 +1,1087 @@ -%% ------------------------------------------------------------------- -%% -%% Copyright (c) 2007-2016 Basho Technologies, Inc. -%% -%% This file is provided to you 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. -%% -%% ------------------------------------------------------------------- - -%% @doc Various functions that are useful throughout Riak. --module(riak_core_util). - --export([moment/0, - make_tmp_dir/0, - replace_file/2, - compare_dates/2, - reload_all/1, - integer_to_list/2, - unique_id_62/0, - str_to_node/1, - chash_key/1, chash_key/2, - chash_std_keyfun/1, - chash_bucketonly_keyfun/1, - mkclientid/1, - start_app_deps/1, - build_tree/3, - orddict_delta/2, - safe_rpc/4, - safe_rpc/5, - rpc_every_member/4, - rpc_every_member_ann/4, - keydelete/2, - multi_keydelete/2, - multi_keydelete/3, - compose/1, - compose/2, - pmap/2, - pmap/3, - multi_rpc/4, - multi_rpc/5, - multi_rpc_ann/4, - multi_rpc_ann/5, - multicall_ann/4, - multicall_ann/5, - shuffle/1, - is_arch/1, - format_ip_and_port/2, - peername/2, - sockname/2, - sha/1, - md5/1, - make_fold_req/1, - make_fold_req/2, - make_fold_req/4, - make_newest_fold_req/1, - proxy_spawn/1, - proxy/2, - enable_job_class/1, - enable_job_class/2, - disable_job_class/1, - disable_job_class/2, - job_class_enabled/1, - job_class_enabled/2, - job_class_disabled_message/2, - report_job_request_disposition/6 - ]). - --include_lib("kernel/include/logger.hrl"). - --include("riak_core_vnode.hrl"). - --ifdef(TEST). --include_lib("eunit/include/eunit.hrl"). --export([counter_loop/1,incr_counter/1,decr_counter/1]). --endif. - -%% R14 Compatibility --compile({no_auto_import,[integer_to_list/2]}). - -%% =================================================================== -%% Public API -%% =================================================================== - -%% 719528 days from Jan 1, 0 to Jan 1, 1970 -%% *86400 seconds/day --define(SEC_TO_EPOCH, 62167219200). - -%% @spec moment() -> integer() -%% @doc Get the current "moment". Current implementation is the -%% number of seconds from year 0 to now, universal time, in -%% the gregorian calendar. - -moment() -> - {Mega, Sec, _Micro} = os:timestamp(), - (Mega * 1000000) + Sec + ?SEC_TO_EPOCH. - -%% @spec compare_dates(string(), string()) -> boolean() -%% @doc Compare two RFC1123 date strings or two now() tuples (or one -%% of each). Return true if date A is later than date B. -compare_dates(A={_,_,_}, B={_,_,_}) -> - %% assume 3-tuples are now() times - A > B; -compare_dates(A, B) when is_list(A) -> - %% assume lists are rfc1123 date strings - compare_dates(rfc1123_to_now(A), B); -compare_dates(A, B) when is_list(B) -> - compare_dates(A, rfc1123_to_now(B)). - -rfc1123_to_now(String) when is_list(String) -> - GSec = calendar:datetime_to_gregorian_seconds( - httpd_util:convert_request_date(String)), - ESec = GSec-?SEC_TO_EPOCH, - Sec = ESec rem 1000000, - MSec = ESec div 1000000, - {MSec, Sec, 0}. - -%% @spec make_tmp_dir() -> string() -%% @doc Create a unique directory in /tmp. Returns the path -%% to the new directory. -make_tmp_dir() -> - TmpId = io_lib:format("riptemp.~p", - [erlang:phash2({rand:uniform(),self()})]), - TempDir = filename:join("/tmp", TmpId), - case filelib:is_dir(TempDir) of - true -> make_tmp_dir(); - false -> - ok = file:make_dir(TempDir), - TempDir - end. - -%% @doc Atomically/safely (to some reasonable level of durablity) -%% replace file `FN' with `Data'. NOTE: since 2.0.3 semantic changed -%% slightly: If `FN' cannot be opened, will not error with a -%% `badmatch', as before, but will instead return `{error, Reason}' --spec replace_file(string(), iodata()) -> ok | {error, term()}. -replace_file(FN, Data) -> - TmpFN = FN ++ ".tmp", - case file:open(TmpFN, [write, raw]) of - {ok, FH} -> - try - ok = file:write(FH, Data), - ok = file:sync(FH), - ok = file:close(FH), - ok = file:rename(TmpFN, FN), - {ok, Contents} = read_file(FN), - true = (Contents == iolist_to_binary(Data)), - ok - catch _:Err -> - {error, Err} - end; - Err -> - Err - end. - -%% @doc Similar to {@link file:read_file/1} but uses raw file `I/O' -read_file(FName) -> - {ok, FD} = file:open(FName, [read, raw, binary]), - IOList = read_file(FD, []), - ok = file:close(FD), - {ok, iolist_to_binary(IOList)}. - -read_file(FD, Acc) -> - case file:read(FD, 4096) of - {ok, Data} -> - read_file(FD, [Data|Acc]); - eof -> - lists:reverse(Acc) - end. - -%% @spec integer_to_list(Integer :: integer(), Base :: integer()) -> -%% string() -%% @doc Convert an integer to its string representation in the given -%% base. Bases 2-62 are supported. -integer_to_list(I, 10) -> - erlang:integer_to_list(I); -integer_to_list(I, Base) - when is_integer(I), is_integer(Base),Base >= 2, Base =< 1+$Z-$A+10+1+$z-$a -> - if I < 0 -> - [$-|integer_to_list(-I, Base, [])]; - true -> - integer_to_list(I, Base, []) - end; -integer_to_list(I, Base) -> - erlang:error(badarg, [I, Base]). - -%% @spec integer_to_list(integer(), integer(), string()) -> string() -integer_to_list(I0, Base, R0) -> - D = I0 rem Base, - I1 = I0 div Base, - R1 = if D >= 36 -> - [D-36+$a|R0]; - D >= 10 -> - [D-10+$A|R0]; - true -> - [D+$0|R0] - end, - if I1 =:= 0 -> - R1; - true -> - integer_to_list(I1, Base, R1) - end. - -sha(Bin) -> - crypto:hash(sha, Bin). - -md5(Bin) -> - crypto:hash(md5, Bin). - -%% @spec unique_id_62() -> string() -%% @doc Create a random identifying integer, returning its string -%% representation in base 62. -unique_id_62() -> - Rand = sha(term_to_binary({make_ref(), os:timestamp()})), - <> = Rand, - integer_to_list(I, 62). - -%% @spec reload_all(Module :: atom()) -> -%% [{purge_response(), load_file_response()}] -%% @type purge_response() = boolean() -%% @type load_file_response() = {module, Module :: atom()}| -%% {error, term()} -%% @doc Ask each member node of the riak ring to reload the given -%% Module. Return is a list of the results of code:purge/1 -%% and code:load_file/1 on each node. -reload_all(Module) -> - {ok, Ring} = riak_core_ring_manager:get_my_ring(), - [{safe_rpc(Node, code, purge, [Module]), - safe_rpc(Node, code, load_file, [Module])} || - Node <- riak_core_ring:all_members(Ring)]. - -%% @spec mkclientid(RemoteNode :: term()) -> ClientID :: list() -%% @doc Create a unique-enough id for vclock clients. -mkclientid(RemoteNode) -> - {{Y,Mo,D},{H,Mi,S}} = erlang:universaltime(), - {_,_,NowPart} = os:timestamp(), - Id = erlang:phash2([Y,Mo,D,H,Mi,S,node(),RemoteNode,NowPart,self()]), - <>. - -%% @spec chash_key(BKey :: riak_object:bkey()) -> chash:index() -%% @doc Create a binary used for determining replica placement. -chash_key({Bucket,_Key}=BKey) -> - BucketProps = riak_core_bucket:get_bucket(Bucket), - chash_key(BKey, BucketProps). - -%% @spec chash_key(BKey :: riak_object:bkey(), [{atom(), any()}]) -> -%% chash:index() -%% @doc Create a binary used for determining replica placement. -chash_key({Bucket,Key}, BucketProps) -> - {_, {M, F}} = lists:keyfind(chash_keyfun, 1, BucketProps), - M:F({Bucket,Key}). - -%% @spec chash_std_keyfun(BKey :: riak_object:bkey()) -> chash:index() -%% @doc Default object/ring hashing fun, direct passthrough of bkey. -chash_std_keyfun({Bucket, Key}) -> chash:key_of({Bucket, Key}). - -%% @spec chash_bucketonly_keyfun(BKey :: riak_object:bkey()) -> chash:index() -%% @doc Object/ring hashing fun that ignores Key, only uses Bucket. -chash_bucketonly_keyfun({Bucket, _Key}) -> chash:key_of(Bucket). - -str_to_node(Node) when is_atom(Node) -> - str_to_node(atom_to_list(Node)); -str_to_node(NodeStr) -> - case string:tokens(NodeStr, "@") of - [NodeName] -> - %% Node name only; no host name. If the local node has a hostname, - %% append it - case node_hostname() of - [] -> - list_to_atom(NodeName); - Hostname -> - list_to_atom(NodeName ++ "@" ++ Hostname) - end; - _ -> - list_to_atom(NodeStr) - end. - -node_hostname() -> - NodeStr = atom_to_list(node()), - case string:tokens(NodeStr, "@") of - [_NodeName, Hostname] -> - Hostname; - _ -> - [] - end. - -%% @spec start_app_deps(App :: atom()) -> ok -%% @doc Start depedent applications of App. -start_app_deps(App) -> - {ok, DepApps} = application:get_key(App, applications), - _ = [ensure_started(A) || A <- DepApps], - ok. - - -%% @spec ensure_started(Application :: atom()) -> ok -%% @doc Start the named application if not already started. -ensure_started(App) -> - case application:start(App) of - ok -> - ok; - {error, {already_started, App}} -> - ok - end. - -%% @doc Returns a copy of `TupleList' where the first occurrence of a tuple whose -%% first element compares equal to `Key' is deleted, if there is such a tuple. -%% Equivalent to `lists:keydelete(Key, 1, TupleList)'. --spec keydelete(atom(), [tuple()]) -> [tuple()]. -keydelete(Key, TupleList) -> - lists:keydelete(Key, 1, TupleList). - -%% @doc Returns a copy of `TupleList' where the first occurrence of a tuple whose -%% first element compares equal to any key in `KeysToDelete' is deleted, if -%% there is such a tuple. --spec multi_keydelete([atom()], [tuple()]) -> [tuple()]. -multi_keydelete(KeysToDelete, TupleList) -> - multi_keydelete(KeysToDelete, 1, TupleList). - -%% @doc Returns a copy of `TupleList' where the Nth occurrence of a tuple whose -%% first element compares equal to any key in `KeysToDelete' is deleted, if -%% there is such a tuple. --spec multi_keydelete([atom()], non_neg_integer(), [tuple()]) -> [tuple()]. -multi_keydelete(KeysToDelete, N, TupleList) -> - lists:foldl( - fun(Key, Acc) -> lists:keydelete(Key, N, Acc) end, - TupleList, - KeysToDelete). - -%% @doc Function composition: returns a function that is the composition of -%% `F' and `G'. --spec compose(fun(), fun()) -> fun(). -compose(F, G) when is_function(F), is_function(G) -> - fun(X) -> - F(G(X)) - end. - -%% @doc Function composition: returns a function that is the composition of all -%% functions in the `Funs' list. --spec compose([fun()]) -> fun(). -compose([Fun]) -> - Fun; -compose([Fun|Funs]) -> - lists:foldl(fun compose/2, Fun, Funs). - -%% @doc Invoke function `F' over each element of list `L' in parallel, -%% returning the results in the same order as the input list. --spec pmap(F, L1) -> L2 when - F :: fun((A) -> B), - L1 :: [A], - L2 :: [B]. -pmap(F, L) -> - Parent = self(), - lists:foldl( - fun(X, N) -> - spawn_link(fun() -> - Parent ! {pmap, N, F(X)} - end), - N+1 - end, 0, L), - L2 = [receive {pmap, N, R} -> {N,R} end || _ <- L], - L3 = lists:keysort(1, L2), - [R || {_,R} <- L3]. - --record(pmap_acc,{ - mapper, - fn, - n_pending=0, - pending=sets:new(), - n_done=0, - done=[], - max_concurrent=1 - }). - -%% @doc Parallel map with a cap on the number of concurrent worker processes. -%% Note: Worker processes are linked to the parent, so a crash propagates. --spec pmap(Fun::function(), List::list(), MaxP::integer()) -> list(). -pmap(Fun, List, MaxP) when MaxP < 1 -> - pmap(Fun, List, 1); -pmap(Fun, List, MaxP) when is_function(Fun), is_list(List), is_integer(MaxP) -> - Mapper = self(), - #pmap_acc{pending=Pending, done=Done} = - lists:foldl(fun pmap_worker/2, - #pmap_acc{mapper=Mapper, - fn=Fun, - max_concurrent=MaxP}, - List), - All = pmap_collect_rest(Pending, Done), - % Restore input order - Sorted = lists:keysort(1, All), - [ R || {_, R} <- Sorted ]. - -%% @doc Fold function for {@link pmap/3} that spawns up to a max number of -%% workers to execute the mapping function over the input list. -pmap_worker(X, Acc = #pmap_acc{n_pending=NP, - pending=Pending, - n_done=ND, - max_concurrent=MaxP, - mapper=Mapper, - fn=Fn}) - when NP < MaxP -> - Worker = - spawn_link(fun() -> - R = Fn(X), - Mapper ! {pmap_result, self(), {NP+ND, R}} - end), - Acc#pmap_acc{n_pending=NP+1, pending=sets:add_element(Worker, Pending)}; -pmap_worker(X, Acc = #pmap_acc{n_pending=NP, - pending=Pending, - n_done=ND, - done=Done, - max_concurrent=MaxP}) - when NP == MaxP -> - {Result, NewPending} = pmap_collect_one(Pending), - pmap_worker(X, Acc#pmap_acc{n_pending=NP-1, pending=NewPending, - n_done=ND+1, done=[Result|Done]}). - -%% @doc Waits for one pending pmap task to finish -pmap_collect_one(Pending) -> - receive - {pmap_result, Pid, Result} -> - Size = sets:size(Pending), - NewPending = sets:del_element(Pid, Pending), - case sets:size(NewPending) of - Size -> - pmap_collect_one(Pending); - _ -> - {Result, NewPending} - end - end. - -pmap_collect_rest(Pending, Done) -> - case sets:size(Pending) of - 0 -> - Done; - _ -> - {Result, NewPending} = pmap_collect_one(Pending), - pmap_collect_rest(NewPending, [Result | Done]) - end. - - -%% @doc Wraps an rpc:call/4 in a try/catch to handle the case where the -%% 'rex' process is not running on the remote node. This is safe in -%% the sense that it won't crash the calling process if the rex -%% process is down. --spec safe_rpc(Node :: node(), Module :: atom(), Function :: atom(), - Args :: [any()]) -> {'badrpc', any()} | any(). -safe_rpc(Node, Module, Function, Args) -> - try rpc:call(Node, Module, Function, Args) of - Result -> - Result - catch - exit:{noproc, _NoProcDetails} -> - {badrpc, rpc_process_down} - end. - -%% @doc Wraps an rpc:call/5 in a try/catch to handle the case where the -%% 'rex' process is not running on the remote node. This is safe in -%% the sense that it won't crash the calling process if the rex -%% process is down. --spec safe_rpc(Node :: node(), Module :: atom(), Function :: atom(), - Args :: [any()], Timeout :: timeout()) -> {'badrpc', any()} | any(). -safe_rpc(Node, Module, Function, Args, Timeout) -> - try rpc:call(Node, Module, Function, Args, Timeout) of - Result -> - Result - catch - 'EXIT':{noproc, _NoProcDetails} -> - {badrpc, rpc_process_down} - end. - -%% @spec rpc_every_member(atom(), atom(), [term()], integer()|infinity) -%% -> {Results::[term()], BadNodes::[node()]} -%% @doc Make an RPC call to the given module and function on each -%% member of the cluster. See rpc:multicall/5 for a description -%% of the return value. -rpc_every_member(Module, Function, Args, Timeout) -> - {ok, MyRing} = riak_core_ring_manager:get_my_ring(), - Nodes = riak_core_ring:all_members(MyRing), - rpc:multicall(Nodes, Module, Function, Args, Timeout). - -%% @doc Same as rpc_every_member/4, but annotate the result set with -%% the name of the node returning the result. -rpc_every_member_ann(Module, Function, Args, Timeout) -> - {ok, MyRing} = riak_core_ring_manager:get_my_ring(), - Nodes = riak_core_ring:all_members(MyRing), - {Results, Down} = multicall_ann(Nodes, Module, Function, Args, Timeout), - {Results, Down}. - -%% @doc Perform an RPC call to a list of nodes in parallel, returning the -%% results in the same order as the input list. --spec multi_rpc([node()], module(), atom(), [any()]) -> [any()]. -multi_rpc(Nodes, Mod, Fun, Args) -> - multi_rpc(Nodes, Mod, Fun, Args, infinity). - -%% @doc Perform an RPC call to a list of nodes in parallel, returning the -%% results in the same order as the input list. --spec multi_rpc([node()], module(), atom(), [any()], timeout()) -> [any()]. -multi_rpc(Nodes, Mod, Fun, Args, Timeout) -> - pmap(fun(Node) -> - safe_rpc(Node, Mod, Fun, Args, Timeout) - end, Nodes). - -%% @doc Perform an RPC call to a list of nodes in parallel, returning the -%% results in the same order as the input list. Each result is tagged -%% with the corresponding node name. --spec multi_rpc_ann([node()], module(), atom(), [any()]) - -> [{node(), any()}]. -multi_rpc_ann(Nodes, Mod, Fun, Args) -> - multi_rpc_ann(Nodes, Mod, Fun, Args, infinity). - -%% @doc Perform an RPC call to a list of nodes in parallel, returning the -%% results in the same order as the input list. Each result is tagged -%% with the corresponding node name. --spec multi_rpc_ann([node()], module(), atom(), [any()], timeout()) - -> [{node(), any()}]. -multi_rpc_ann(Nodes, Mod, Fun, Args, Timeout) -> - Results = multi_rpc(Nodes, Mod, Fun, Args, Timeout), - lists:zip(Nodes, Results). - -%% @doc Similar to {@link rpc:multicall/4}. Performs an RPC call to a list -%% of nodes in parallel, returning a list of results as well as a list -%% of nodes that are down/unreachable. The results will be returned in -%% the same order as the input list, and each result is tagged with the -%% corresponding node name. --spec multicall_ann([node()], module(), atom(), [any()]) - -> {Results :: [{node(), any()}], Down :: [node()]}. -multicall_ann(Nodes, Mod, Fun, Args) -> - multicall_ann(Nodes, Mod, Fun, Args, infinity). - -%% @doc Similar to {@link rpc:multicall/6}. Performs an RPC call to a list -%% of nodes in parallel, returning a list of results as well as a list -%% of nodes that are down/unreachable. The results will be returned in -%% the same order as the input list, and each result is tagged with the -%% corresponding node name. --spec multicall_ann([node()], module(), atom(), [any()], timeout()) - -> {Results :: [{node(), any()}], Down :: [node()]}. -multicall_ann(Nodes, Mod, Fun, Args, Timeout) -> - L = multi_rpc_ann(Nodes, Mod, Fun, Args, Timeout), - {Results, DownAnn} = - lists:partition(fun({_, Result}) -> - Result /= {badrpc, nodedown} - end, L), - {Down, _} = lists:unzip(DownAnn), - {Results, Down}. - -%% @doc Convert a list of elements into an N-ary tree. This conversion -%% works by treating the list as an array-based tree where, for -%% example in a binary 2-ary tree, a node at index i has children -%% 2i and 2i+1. The conversion also supports a "cycles" mode where -%% the array is logically wrapped around to ensure leaf nodes also -%% have children by giving them backedges to other elements. - --spec build_tree(N :: integer(), Nodes :: [term()], Opts :: [term()]) - -> orddict:orddict(). -build_tree(N, Nodes, Opts) -> - case lists:member(cycles, Opts) of - true -> - Expand = lists:flatten(lists:duplicate(N+1, Nodes)); - false -> - Expand = Nodes - end, - {Tree, _} = - lists:foldl(fun(Elm, {Result, Worklist}) -> - Len = erlang:min(N, length(Worklist)), - {Children, Rest} = lists:split(Len, Worklist), - NewResult = [{Elm, Children} | Result], - {NewResult, Rest} - end, {[], tl(Expand)}, Nodes), - orddict:from_list(Tree). - -orddict_delta(A, B) -> - %% Pad both A and B to the same length - DummyA = [{Key, '$none'} || {Key, _} <- B], - A2 = orddict:merge(fun(_, Value, _) -> - Value - end, A, DummyA), - - DummyB = [{Key, '$none'} || {Key, _} <- A], - B2 = orddict:merge(fun(_, Value, _) -> - Value - end, B, DummyB), - - %% Merge and filter out equal values - Merged = orddict:merge(fun(_, AVal, BVal) -> - {AVal, BVal} - end, A2, B2), - Diff = orddict:filter(fun(_, {Same, Same}) -> - false; - (_, _) -> - true - end, Merged), - Diff. - -shuffle(L) -> - N = 134217727, %% Largest small integer on 32-bit Erlang - L2 = [{rand:uniform(N), E} || E <- L], - L3 = [E || {_, E} <- lists:sort(L2)], - L3. - -%% Returns a forced-lowercase architecture for this node --spec get_arch () -> string(). -get_arch () -> string:to_lower(erlang:system_info(system_architecture)). - -%% Checks if this node is of a given architecture --spec is_arch (atom()) -> boolean(). -is_arch (linux) -> string:str(get_arch(),"linux") > 0; -is_arch (darwin) -> string:str(get_arch(),"darwin") > 0; -is_arch (sunos) -> string:str(get_arch(),"sunos") > 0; -is_arch (osx) -> is_arch(darwin); -is_arch (solaris) -> is_arch(sunos); -is_arch (Arch) -> throw({unsupported_architecture,Arch}). - -format_ip_and_port(Ip, Port) when is_list(Ip) -> - lists:flatten(io_lib:format("~s:~p",[Ip,Port])); -format_ip_and_port(Ip, Port) when is_tuple(Ip) -> - lists:flatten(io_lib:format("~s:~p",[inet_parse:ntoa(Ip), - Port])). -peername(Socket, Transport) -> - case Transport:peername(Socket) of - {ok, {Ip, Port}} -> - format_ip_and_port(Ip, Port); - {error, Reason} -> - %% just return a string so JSON doesn't blow up - lists:flatten(io_lib:format("error:~p", [Reason])) - end. - -sockname(Socket, Transport) -> - case Transport:sockname(Socket) of - {ok, {Ip, Port}} -> - format_ip_and_port(Ip, Port); - {error, Reason} -> - %% just return a string so JSON doesn't blow up - lists:flatten(io_lib:format("error:~p", [Reason])) - end. - -%% @doc Convert a #riak_core_fold_req_v? record to the cluster's maximum -%% supported record version. - -make_fold_req(#riak_core_fold_req_v1{foldfun=FoldFun, acc0=Acc0}) -> - make_fold_req(FoldFun, Acc0, false, []); -make_fold_req(?FOLD_REQ{foldfun=FoldFun, acc0=Acc0, - forwardable=Forwardable, opts=Opts}) -> - make_fold_req(FoldFun, Acc0, Forwardable, Opts). - -make_fold_req(FoldFun, Acc0) -> - make_fold_req(FoldFun, Acc0, false, []). - -make_fold_req(FoldFun, Acc0, Forwardable, Opts) -> - make_fold_reqv(riak_core_capability:get({riak_core, fold_req_version}, v1), - FoldFun, Acc0, Forwardable, Opts). - -%% @doc Force a #riak_core_fold_req_v? record to the latest version, -%% regardless of cluster support - -make_newest_fold_req(#riak_core_fold_req_v1{foldfun=FoldFun, acc0=Acc0}) -> - make_fold_reqv(v2, FoldFun, Acc0, false, []); -make_newest_fold_req(?FOLD_REQ{} = F) -> - F. - -%% @doc Spawn an intermediate proxy process to handle errors during gen_xxx -%% calls. -proxy_spawn(Fun) -> - %% Note: using spawn_monitor does not trigger selective receive - %% optimization, but spawn + monitor does. Silly Erlang. - Pid = spawn(?MODULE, proxy, [self(), Fun]), - MRef = monitor(process, Pid), - Pid ! {proxy, MRef}, - receive - {proxy_reply, MRef, Result} -> - demonitor(MRef, [flush]), - Result; - {'DOWN', MRef, _, _, Reason} -> - {error, Reason} - end. - - -%% @private -make_fold_reqv(v1, FoldFun, Acc0, _Forwardable, _Opts) - when is_function(FoldFun, 3) -> - #riak_core_fold_req_v1{foldfun=FoldFun, acc0=Acc0}; -make_fold_reqv(v2, FoldFun, Acc0, Forwardable, Opts) - when is_function(FoldFun, 3) - andalso (Forwardable == true orelse Forwardable == false) - andalso is_list(Opts) -> - ?FOLD_REQ{foldfun=FoldFun, acc0=Acc0, - forwardable=Forwardable, opts=Opts}. - -%% @private - used with proxy_spawn -proxy(Parent, Fun) -> - _ = monitor(process, Parent), - receive - {proxy, MRef} -> - Result = Fun(), - Parent ! {proxy_reply, MRef, Result}; - {'DOWN', _, _, _, _} -> - ok - end. - --spec enable_job_class(atom(), atom()) -> ok | {error, term()}. -%% @doc Enables the specified Application/Operation job class. -%% This is the public API for use via RPC. -%% WARNING: This function is not suitable for parallel execution with itself -%% or its complement disable_job_class/2. -enable_job_class(Application, Operation) - when erlang:is_atom(Application) andalso erlang:is_atom(Operation) -> - enable_job_class({Application, Operation}); -enable_job_class(Application, Operation) -> - {error, {badarg, {Application, Operation}}}. - --spec disable_job_class(atom(), atom()) -> ok | {error, term()}. -%% @doc Disables the specified Application/Operation job class. -%% This is the public API for use via RPC. -%% WARNING: This function is not suitable for parallel execution with itself -%% or its complement enable_job_class/2. -disable_job_class(Application, Operation) - when erlang:is_atom(Application) andalso erlang:is_atom(Operation) -> - disable_job_class({Application, Operation}); -disable_job_class(Application, Operation) -> - {error, {badarg, {Application, Operation}}}. - --spec job_class_enabled(atom(), atom()) -> boolean() | {error, term()}. -%% @doc Reports whether the specified Application/Operation job class is enabled. -%% This is the public API for use via RPC. -job_class_enabled(Application, Operation) - when erlang:is_atom(Application) andalso erlang:is_atom(Operation) -> - job_class_enabled({Application, Operation}); -job_class_enabled(Application, Operation) -> - {error, {badarg, {Application, Operation}}}. - --spec enable_job_class(Class :: term()) -> ok | {error, term()}. -%% @doc Internal API to enable the specified job class. -%% WARNING: -%% * This function may not remain in this form once the Jobs API is live! -%% * Parameter types ARE NOT validated by the same rules as the public API! -%% You are STRONGLY advised to use enable_job_class/2. -enable_job_class(Class) -> - case app_helper:get_env(riak_core, job_accept_class) of - [_|_] = EnabledClasses -> - case lists:member(Class, EnabledClasses) of - true -> - ok; - _ -> - application:set_env( - riak_core, job_accept_class, [Class | EnabledClasses]) - end; - _ -> - application:set_env(riak_core, job_accept_class, [Class]) - end. - --spec disable_job_class(Class :: term()) -> ok | {error, term()}. -%% @doc Internal API to disable the specified job class. -%% WARNING: -%% * This function may not remain in this form once the Jobs API is live! -%% * Parameter types ARE NOT validated by the same rules as the public API! -%% You are STRONGLY advised to use disable_job_class/2. -disable_job_class(Class) -> - case app_helper:get_env(riak_core, job_accept_class) of - [_|_] = EnabledClasses -> - case lists:member(Class, EnabledClasses) of - false -> - ok; - _ -> - application:set_env(riak_core, job_accept_class, - lists:delete(Class, EnabledClasses)) - end; - _ -> - ok - end. - --spec job_class_enabled(Class :: term()) -> boolean(). -%% @doc Internal API to determine whether to accept/reject a job. -%% WARNING: -%% * This function may not remain in this form once the Jobs API is live! -%% * Parameter types ARE NOT validated by the same rules as the public API! -%% You are STRONGLY advised to use job_class_enabled/2. -job_class_enabled(Class) -> - case app_helper:get_env(riak_core, job_accept_class) of - undefined -> - true; - [] -> - false; - [_|_] = EnabledClasses -> - lists:member(Class, EnabledClasses); - Other -> - % Don't crash if it's not a list - that should never be the case, - % but since the value *can* be manipulated externally be more - % accommodating. If someone mucks it up, nothing's going to be - % allowed, but give them a chance to catch on instead of crashing. - ?LOG_ERROR( - "riak_core.job_accept_class is not a list: ~p", [Other]), - false - end. - --spec job_class_disabled_message(ReturnType :: atom(), Class :: term()) - -> binary() | string(). -%% @doc The error message to be returned to a client for a disabled job class. -%% WARNING: -%% * This function is likely to be extended to accept a Job as well as a Class -%% when the Jobs API is live. -job_class_disabled_message(binary, Class) -> - erlang:list_to_binary(job_class_disabled_message(text, Class)); -job_class_disabled_message(text, Class) -> - lists:flatten(io_lib:format("Operation '~p' is not enabled", [Class])). - --spec report_job_request_disposition(Accepted :: boolean(), Class :: term(), - Mod :: module(), Func :: atom(), Line :: pos_integer(), Client :: term()) - -> ok | {error, term()}. -%% @doc Report/record the disposition of an async job request. -%% -%% Logs an appropriate message and reports to whoever needs to know. -%% WARNING: -%% * This function is likely to be extended to accept a Job as well as a Class -%% when the Jobs API is live. -%% -%% Parameters: -%% * Accepted - Whether the specified job Class is enabled. -%% * Class - The Class of the job, by convention {Application, Operation}. -%% * Mod/Func/Line - The Module, function, and source line number, -%% respectively, that will be reported as the source of the call. -%% * Client - Any term indicating the originator of the request. -%% By convention, when meaningful client identification information is not -%% available, Client is an atom representing the protocol through which the -%% request was received. -%% -report_job_request_disposition(true, Class, _Mod, _Func, _Line, Client) -> - ?LOG_DEBUG("Request '~p' accepted from ~p", [Class, Client]); -report_job_request_disposition(false, Class, _Mod, _Func, _Line, Client) -> - ?LOG_WARNING("Request '~p' disabled from ~p", [Class, Client]). - -%% =================================================================== -%% EUnit tests -%% =================================================================== --ifdef(TEST). - -moment_test() -> - M1 = riak_core_util:moment(), - M2 = riak_core_util:moment(), - ?assert(M2 >= M1). - -clientid_uniqueness_test() -> - ClientIds = [mkclientid('somenode@somehost') || _I <- lists:seq(0, 10000)], - length(ClientIds) =:= length(sets:to_list(sets:from_list(ClientIds))). - -build_tree_test() -> - Flat = [1, - 11, 12, - 111, 112, 121, 122, - 1111, 1112, 1121, 1122, 1211, 1212, 1221, 1222], - - %% 2-ary tree decomposition - ATree = [{1, [ 11, 12]}, - {11, [ 111, 112]}, - {12, [ 121, 122]}, - {111, [1111, 1112]}, - {112, [1121, 1122]}, - {121, [1211, 1212]}, - {122, [1221, 1222]}, - {1111, []}, - {1112, []}, - {1121, []}, - {1122, []}, - {1211, []}, - {1212, []}, - {1221, []}, - {1222, []}], - - %% 2-ary tree decomposition with cyclic wrap-around - CTree = [{1, [ 11, 12]}, - {11, [ 111, 112]}, - {12, [ 121, 122]}, - {111, [1111, 1112]}, - {112, [1121, 1122]}, - {121, [1211, 1212]}, - {122, [1221, 1222]}, - {1111, [ 1, 11]}, - {1112, [ 12, 111]}, - {1121, [ 112, 121]}, - {1122, [ 122, 1111]}, - {1211, [1112, 1121]}, - {1212, [1122, 1211]}, - {1221, [1212, 1221]}, - {1222, [1222, 1]}], - - ?assertEqual(ATree, build_tree(2, Flat, [])), - ?assertEqual(CTree, build_tree(2, Flat, [cycles])), - ok. - - -counter_loop(N) -> - receive - {up, Pid} -> - N2=N+1, - Pid ! {counter_value, N2}, - counter_loop(N2); - down -> - counter_loop(N-1); - exit -> - exit(normal) - end. - -incr_counter(CounterPid) -> - CounterPid ! {up, self()}, - receive - {counter_value, N} -> N - after - 3000 -> - ?assert(false) - end. - -decr_counter(CounterPid) -> - CounterPid ! down. - -multi_keydelete_test_() -> - Languages = [{lisp, 1958}, - {ml, 1973}, - {erlang, 1986}, - {haskell, 1990}, - {ocaml, 1996}, - {clojure, 2007}, - {elixir, 2012}], - ?_assertMatch( - [{lisp, _}, {ml, _}, {erlang, _}, {haskell, _}], - multi_keydelete([ocaml, clojure, elixir], Languages)). - -compose_test_() -> - Upper = fun string:to_upper/1, - Reverse = fun lists:reverse/1, - Strip = fun(S) -> string:strip(S, both, $!) end, - Composed = compose([Upper, Reverse, Strip]), - ?_assertEqual("DLROW OLLEH", Composed("Hello world!")). - -pmap_test_() -> - Fgood = fun(X) -> 2 * X end, - Fbad = fun(3) -> throw(die_on_3); - (X) -> Fgood(X) - end, - Lin = [1,2,3,4], - Lout = [2,4,6,8], - {setup, - fun() -> error_logger:tty(false) end, - fun(_) -> error_logger:tty(true) end, - [fun() -> - % Test simple map case - ?assertEqual(Lout, pmap(Fgood, Lin)), - % Verify a crashing process will not stall pmap - Parent = self(), - Pid = spawn(fun() -> - % Caller trapping exits causes stall!! - % TODO: Consider pmapping in a spawned proc - % process_flag(trap_exit, true), - pmap(Fbad, Lin), - ?debugMsg("pmap finished just fine"), - Parent ! no_crash_yo - end), - MonRef = monitor(process, Pid), - receive - {'DOWN', MonRef, _, _, _} -> - ok; - no_crash_yo -> - ?assert(pmap_did_not_crash_as_expected) - end - end - ]}. - -bounded_pmap_test_() -> - Fun1 = fun(X) -> X+2 end, - Tests = - fun(CountPid) -> - GFun = fun(Max) -> - fun(X) -> - ?assert(incr_counter(CountPid) =< Max), - timer:sleep(1), - decr_counter(CountPid), - Fun1(X) - end - end, - [ - fun() -> - ?assertEqual(lists:seq(Fun1(1), Fun1(N)), - pmap(GFun(MaxP), - lists:seq(1, N), MaxP)) - end || - MaxP <- lists:seq(1,20), - N <- lists:seq(0,10) - ] - end, - {setup, - fun() -> - Pid = spawn_link(?MODULE, counter_loop, [0]), - monitor(process, Pid), - Pid - end, - fun(Pid) -> - Pid ! exit, - receive - {'DOWN', _Ref, process, Pid, _Info} -> ok - after - 3000 -> - ?debugMsg("pmap counter process did not go down in time"), - ?assert(false) - end, - ok - end, - Tests - }. - -make_fold_req_test_() -> - {setup, - fun() -> - meck:new(riak_core_capability, [passthrough]) - end, - fun(_) -> - meck:unload(riak_core_capability) - end, - [ - fun() -> - FoldFun = fun(_, _, _) -> ok end, - Acc0 = acc0, - Forw = true, - Opts = [opts], - F_1 = #riak_core_fold_req_v1{foldfun=FoldFun, acc0=Acc0}, - F_2 = #riak_core_fold_req_v2{foldfun=FoldFun, acc0=Acc0, - forwardable=Forw, opts=Opts}, - F_2_default = #riak_core_fold_req_v2{foldfun=FoldFun, acc0=Acc0, - forwardable=false, opts=[]}, - Newest = fun() -> F_2_default = make_newest_fold_req(F_1), - F_2 = make_newest_fold_req(F_2), - ok - end, - - meck:expect(riak_core_capability, get, - fun(_, _) -> v1 end), - F_1 = make_fold_req(F_1), - F_1 = make_fold_req(F_2), - F_1 = make_fold_req(FoldFun, Acc0), - F_1 = make_fold_req(FoldFun, Acc0, Forw, Opts), - ok = Newest(), - - meck:expect(riak_core_capability, get, - fun(_, _) -> v2 end), - F_2_default = make_fold_req(F_1), - F_2 = make_fold_req(F_2), - F_2_default = make_fold_req(FoldFun, Acc0), - F_2 = make_fold_req(FoldFun, Acc0, Forw, Opts), - ok = Newest() - end - ] - }. - -proxy_spawn_test() -> - A = proxy_spawn(fun() -> a end), - ?assertEqual(a, A), - B = proxy_spawn(fun() -> exit(killer_fun) end), - ?assertEqual({error, killer_fun}, B), - - %% Ensure no errant 'DOWN' messages - receive - {'DOWN', _, _, _, _}=Msg -> - throw({error, {badmsg, Msg}}); - _ -> - ok - after 1000 -> - ok - end. - --endif. - +%% ------------------------------------------------------------------- +%% +%% Copyright (c) 2007-2016 Basho Technologies, Inc. +%% Copyright (c) 2020-2025 Workday, Inc. +%% +%% This file is provided to you 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. +%% +%% ------------------------------------------------------------------- + +%% @doc Various functions that are useful throughout Riak. +-module(riak_core_util). + +-export([moment/0, + make_tmp_dir/0, + replace_file/2, + compare_dates/2, + reload_all/1, + integer_to_list/2, + unique_id_62/0, + str_to_node/1, + chash_key/1, chash_key/2, + chash_std_keyfun/1, + chash_bucketonly_keyfun/1, + mkclientid/1, + start_app_deps/1, + build_tree/3, + orddict_delta/2, + safe_rpc/4, + safe_rpc/5, + rpc_every_member/4, + rpc_every_member_ann/4, + keydelete/2, + multi_keydelete/2, + multi_keydelete/3, + compose/1, + compose/2, + pmap/2, + pmap/3, + multi_rpc/4, + multi_rpc/5, + multi_rpc_ann/4, + multi_rpc_ann/5, + multicall_ann/4, + multicall_ann/5, + shuffle/1, + is_arch/1, + format_ip_and_port/2, + peername/2, + sockname/2, + sha/1, + md5/1, + make_fold_req/1, + make_fold_req/2, + make_fold_req/4, + make_newest_fold_req/1, + proxy_spawn/1, + proxy/2, + enable_job_class/1, + enable_job_class/2, + disable_job_class/1, + disable_job_class/2, + job_class_enabled/1, + job_class_enabled/2, + job_class_disabled_message/2, + report_job_request_disposition/6 + ]). + +-include_lib("kernel/include/logger.hrl"). + +-include("riak_core_vnode.hrl"). + +-ifdef(TEST). +-include_lib("eunit/include/eunit.hrl"). +-export([counter_loop/1,incr_counter/1,decr_counter/1]). +-endif. + +%% R14 Compatibility +-compile({no_auto_import,[integer_to_list/2]}). + +%% =================================================================== +%% Public API +%% =================================================================== + +%% 719528 days from Jan 1, 0 to Jan 1, 1970 +%% *86400 seconds/day +-define(SEC_TO_EPOCH, 62167219200). + +%% @spec moment() -> integer() +%% @doc Get the current "moment". Current implementation is the +%% number of seconds from year 0 to now, universal time, in +%% the gregorian calendar. + +moment() -> + {Mega, Sec, _Micro} = os:timestamp(), + (Mega * 1000000) + Sec + ?SEC_TO_EPOCH. + +%% @spec compare_dates(string(), string()) -> boolean() +%% @doc Compare two RFC1123 date strings or two now() tuples (or one +%% of each). Return true if date A is later than date B. +compare_dates(A={_,_,_}, B={_,_,_}) -> + %% assume 3-tuples are now() times + A > B; +compare_dates(A, B) when is_list(A) -> + %% assume lists are rfc1123 date strings + compare_dates(rfc1123_to_now(A), B); +compare_dates(A, B) when is_list(B) -> + compare_dates(A, rfc1123_to_now(B)). + +rfc1123_to_now(String) when is_list(String) -> + GSec = calendar:datetime_to_gregorian_seconds( + httpd_util:convert_request_date(String)), + ESec = GSec-?SEC_TO_EPOCH, + Sec = ESec rem 1000000, + MSec = ESec div 1000000, + {MSec, Sec, 0}. + +%% @spec make_tmp_dir() -> string() +%% @doc Create a unique directory in /tmp. Returns the path +%% to the new directory. +make_tmp_dir() -> + TmpId = io_lib:format("riptemp.~p", + [erlang:phash2({rand:uniform(),self()})]), + TempDir = filename:join("/tmp", TmpId), + case filelib:is_dir(TempDir) of + true -> make_tmp_dir(); + false -> + ok = file:make_dir(TempDir), + TempDir + end. + +%% @doc Atomically/safely (to some reasonable level of durablity) +%% replace file `FN' with `Data'. NOTE: since 2.0.3 semantic changed +%% slightly: If `FN' cannot be opened, will not error with a +%% `badmatch', as before, but will instead return `{error, Reason}' +-spec replace_file(string(), iodata()) -> ok | {error, term()}. +replace_file(FN, Data) -> + TmpFN = FN ++ ".tmp", + case file:open(TmpFN, [write, raw]) of + {ok, FH} -> + try + ok = file:write(FH, Data), + ok = file:sync(FH), + ok = file:close(FH), + ok = file:rename(TmpFN, FN), + {ok, Contents} = read_file(FN), + true = (Contents == iolist_to_binary(Data)), + ok + catch _:Err -> + {error, Err} + end; + Err -> + Err + end. + +%% @doc Similar to {@link file:read_file/1} but uses raw file `I/O' +read_file(FName) -> + {ok, FD} = file:open(FName, [read, raw, binary]), + IOList = read_file(FD, []), + ok = file:close(FD), + {ok, iolist_to_binary(IOList)}. + +read_file(FD, Acc) -> + case file:read(FD, 4096) of + {ok, Data} -> + read_file(FD, [Data|Acc]); + eof -> + lists:reverse(Acc) + end. + +%% @spec integer_to_list(Integer :: integer(), Base :: integer()) -> +%% string() +%% @doc Convert an integer to its string representation in the given +%% base. Bases 2-62 are supported. +integer_to_list(I, 10) -> + erlang:integer_to_list(I); +integer_to_list(I, Base) + when is_integer(I), is_integer(Base),Base >= 2, Base =< 1+$Z-$A+10+1+$z-$a -> + if I < 0 -> + [$-|integer_to_list(-I, Base, [])]; + true -> + integer_to_list(I, Base, []) + end; +integer_to_list(I, Base) -> + erlang:error(badarg, [I, Base]). + +%% @spec integer_to_list(integer(), integer(), string()) -> string() +integer_to_list(I0, Base, R0) -> + D = I0 rem Base, + I1 = I0 div Base, + R1 = if D >= 36 -> + [D-36+$a|R0]; + D >= 10 -> + [D-10+$A|R0]; + true -> + [D+$0|R0] + end, + if I1 =:= 0 -> + R1; + true -> + integer_to_list(I1, Base, R1) + end. + +sha(Bin) -> + crypto:hash(sha, Bin). + +md5(Bin) -> + crypto:hash(md5, Bin). + +%% @spec unique_id_62() -> string() +%% @doc Create a random identifying integer, returning its string +%% representation in base 62. +unique_id_62() -> + Rand = sha(term_to_binary({make_ref(), os:timestamp()})), + <> = Rand, + integer_to_list(I, 62). + +%% @spec reload_all(Module :: atom()) -> +%% [{purge_response(), load_file_response()}] +%% @type purge_response() = boolean() +%% @type load_file_response() = {module, Module :: atom()}| +%% {error, term()} +%% @doc Ask each member node of the riak ring to reload the given +%% Module. Return is a list of the results of code:purge/1 +%% and code:load_file/1 on each node. +reload_all(Module) -> + {ok, Ring} = riak_core_ring_manager:get_my_ring(), + [{safe_rpc(Node, code, purge, [Module]), + safe_rpc(Node, code, load_file, [Module])} || + Node <- riak_core_ring:all_members(Ring)]. + +%% @spec mkclientid(RemoteNode :: term()) -> ClientID :: list() +%% @doc Create a unique-enough id for vclock clients. +mkclientid(RemoteNode) -> + {{Y,Mo,D},{H,Mi,S}} = erlang:universaltime(), + {_,_,NowPart} = os:timestamp(), + Id = erlang:phash2([Y,Mo,D,H,Mi,S,node(),RemoteNode,NowPart,self()]), + <>. + +%% @spec chash_key(BKey :: riak_object:bkey()) -> chash:index() +%% @doc Create a binary used for determining replica placement. +chash_key({Bucket,_Key}=BKey) -> + BucketProps = riak_core_bucket:get_bucket(Bucket), + chash_key(BKey, BucketProps). + +%% @spec chash_key(BKey :: riak_object:bkey(), [{atom(), any()}]) -> +%% chash:index() +%% @doc Create a binary used for determining replica placement. +chash_key({Bucket,Key}, BucketProps) -> + {_, {M, F}} = lists:keyfind(chash_keyfun, 1, BucketProps), + M:F({Bucket,Key}). + +%% @spec chash_std_keyfun(BKey :: riak_object:bkey()) -> chash:index() +%% @doc Default object/ring hashing fun, direct passthrough of bkey. +chash_std_keyfun({Bucket, Key}) -> chash:key_of({Bucket, Key}). + +%% @spec chash_bucketonly_keyfun(BKey :: riak_object:bkey()) -> chash:index() +%% @doc Object/ring hashing fun that ignores Key, only uses Bucket. +chash_bucketonly_keyfun({Bucket, _Key}) -> chash:key_of(Bucket). + +str_to_node(Node) when is_atom(Node) -> + str_to_node(atom_to_list(Node)); +str_to_node(NodeStr) -> + case string:tokens(NodeStr, "@") of + [NodeName] -> + %% Node name only; no host name. If the local node has a hostname, + %% append it + case node_hostname() of + [] -> + list_to_atom(NodeName); + Hostname -> + list_to_atom(NodeName ++ "@" ++ Hostname) + end; + _ -> + list_to_atom(NodeStr) + end. + +node_hostname() -> + NodeStr = atom_to_list(node()), + case string:tokens(NodeStr, "@") of + [_NodeName, Hostname] -> + Hostname; + _ -> + [] + end. + +%% @spec start_app_deps(App :: atom()) -> ok +%% @doc Start depedent applications of App. +start_app_deps(App) -> + {ok, DepApps} = application:get_key(App, applications), + _ = [ensure_started(A) || A <- DepApps], + ok. + + +%% @spec ensure_started(Application :: atom()) -> ok +%% @doc Start the named application if not already started. +ensure_started(App) -> + case application:start(App) of + ok -> + ok; + {error, {already_started, App}} -> + ok + end. + +%% @doc Returns a copy of `TupleList' where the first occurrence of a tuple whose +%% first element compares equal to `Key' is deleted, if there is such a tuple. +%% Equivalent to `lists:keydelete(Key, 1, TupleList)'. +-spec keydelete(atom(), [tuple()]) -> [tuple()]. +keydelete(Key, TupleList) -> + lists:keydelete(Key, 1, TupleList). + +%% @doc Returns a copy of `TupleList' where the first occurrence of a tuple whose +%% first element compares equal to any key in `KeysToDelete' is deleted, if +%% there is such a tuple. +-spec multi_keydelete([atom()], [tuple()]) -> [tuple()]. +multi_keydelete(KeysToDelete, TupleList) -> + multi_keydelete(KeysToDelete, 1, TupleList). + +%% @doc Returns a copy of `TupleList' where the Nth occurrence of a tuple whose +%% first element compares equal to any key in `KeysToDelete' is deleted, if +%% there is such a tuple. +-spec multi_keydelete([atom()], non_neg_integer(), [tuple()]) -> [tuple()]. +multi_keydelete(KeysToDelete, N, TupleList) -> + lists:foldl( + fun(Key, Acc) -> lists:keydelete(Key, N, Acc) end, + TupleList, + KeysToDelete). + +%% @doc Function composition: returns a function that is the composition of +%% `F' and `G'. +-spec compose(fun(), fun()) -> fun(). +compose(F, G) when is_function(F), is_function(G) -> + fun(X) -> + F(G(X)) + end. + +%% @doc Function composition: returns a function that is the composition of all +%% functions in the `Funs' list. +-spec compose([fun()]) -> fun(). +compose([Fun]) -> + Fun; +compose([Fun|Funs]) -> + lists:foldl(fun compose/2, Fun, Funs). + +%% @doc Invoke function `F' over each element of list `L' in parallel, +%% returning the results in the same order as the input list. +-spec pmap(F, L1) -> L2 when + F :: fun((A) -> B), + L1 :: [A], + L2 :: [B]. +pmap(F, L) -> + Parent = self(), + lists:foldl( + fun(X, N) -> + spawn_link(fun() -> + Parent ! {pmap, N, F(X)} + end), + N+1 + end, 0, L), + L2 = [receive {pmap, N, R} -> {N,R} end || _ <- L], + L3 = lists:keysort(1, L2), + [R || {_,R} <- L3]. + +-record(pmap_acc,{ + mapper, + fn, + n_pending=0, + pending=sets:new(), + n_done=0, + done=[], + max_concurrent=1 + }). + +%% @doc Parallel map with a cap on the number of concurrent worker processes. +%% Note: Worker processes are linked to the parent, so a crash propagates. +-spec pmap(Fun::function(), List::list(), MaxP::integer()) -> list(). +pmap(Fun, List, MaxP) when MaxP < 1 -> + pmap(Fun, List, 1); +pmap(Fun, List, MaxP) when is_function(Fun), is_list(List), is_integer(MaxP) -> + Mapper = self(), + #pmap_acc{pending=Pending, done=Done} = + lists:foldl(fun pmap_worker/2, + #pmap_acc{mapper=Mapper, + fn=Fun, + max_concurrent=MaxP}, + List), + All = pmap_collect_rest(Pending, Done), + % Restore input order + Sorted = lists:keysort(1, All), + [ R || {_, R} <- Sorted ]. + +%% @doc Fold function for {@link pmap/3} that spawns up to a max number of +%% workers to execute the mapping function over the input list. +pmap_worker(X, Acc = #pmap_acc{n_pending=NP, + pending=Pending, + n_done=ND, + max_concurrent=MaxP, + mapper=Mapper, + fn=Fn}) + when NP < MaxP -> + Worker = + spawn_link(fun() -> + R = Fn(X), + Mapper ! {pmap_result, self(), {NP+ND, R}} + end), + Acc#pmap_acc{n_pending=NP+1, pending=sets:add_element(Worker, Pending)}; +pmap_worker(X, Acc = #pmap_acc{n_pending=NP, + pending=Pending, + n_done=ND, + done=Done, + max_concurrent=MaxP}) + when NP == MaxP -> + {Result, NewPending} = pmap_collect_one(Pending), + pmap_worker(X, Acc#pmap_acc{n_pending=NP-1, pending=NewPending, + n_done=ND+1, done=[Result|Done]}). + +%% @doc Waits for one pending pmap task to finish +pmap_collect_one(Pending) -> + receive + {pmap_result, Pid, Result} -> + Size = sets:size(Pending), + NewPending = sets:del_element(Pid, Pending), + case sets:size(NewPending) of + Size -> + pmap_collect_one(Pending); + _ -> + {Result, NewPending} + end + end. + +pmap_collect_rest(Pending, Done) -> + case sets:size(Pending) of + 0 -> + Done; + _ -> + {Result, NewPending} = pmap_collect_one(Pending), + pmap_collect_rest(NewPending, [Result | Done]) + end. + + +%% @doc Wraps an rpc:call/4 in a try/catch to handle the case where the +%% 'rex' process is not running on the remote node. This is safe in +%% the sense that it won't crash the calling process if the rex +%% process is down. +-spec safe_rpc(Node :: node(), Module :: atom(), Function :: atom(), + Args :: [any()]) -> {'badrpc', any()} | any(). +safe_rpc(Node, Module, Function, Args) -> + try rpc:call(Node, Module, Function, Args) of + Result -> + Result + catch + exit:{noproc, _NoProcDetails} -> + {badrpc, rpc_process_down} + end. + +%% @doc Wraps an rpc:call/5 in a try/catch to handle the case where the +%% 'rex' process is not running on the remote node. This is safe in +%% the sense that it won't crash the calling process if the rex +%% process is down. +-spec safe_rpc(Node :: node(), Module :: atom(), Function :: atom(), + Args :: [any()], Timeout :: timeout()) -> {'badrpc', any()} | any(). +safe_rpc(Node, Module, Function, Args, Timeout) -> + try rpc:call(Node, Module, Function, Args, Timeout) of + Result -> + Result + catch + 'EXIT':{noproc, _NoProcDetails} -> + {badrpc, rpc_process_down} + end. + +%% @spec rpc_every_member(atom(), atom(), [term()], integer()|infinity) +%% -> {Results::[term()], BadNodes::[node()]} +%% @doc Make an RPC call to the given module and function on each +%% member of the cluster. See rpc:multicall/5 for a description +%% of the return value. +rpc_every_member(Module, Function, Args, Timeout) -> + {ok, MyRing} = riak_core_ring_manager:get_my_ring(), + Nodes = riak_core_ring:all_members(MyRing), + rpc:multicall(Nodes, Module, Function, Args, Timeout). + +%% @doc Same as rpc_every_member/4, but annotate the result set with +%% the name of the node returning the result. +rpc_every_member_ann(Module, Function, Args, Timeout) -> + {ok, MyRing} = riak_core_ring_manager:get_my_ring(), + Nodes = riak_core_ring:all_members(MyRing), + {Results, Down} = multicall_ann(Nodes, Module, Function, Args, Timeout), + {Results, Down}. + +%% @doc Perform an RPC call to a list of nodes in parallel, returning the +%% results in the same order as the input list. +-spec multi_rpc([node()], module(), atom(), [any()]) -> [any()]. +multi_rpc(Nodes, Mod, Fun, Args) -> + multi_rpc(Nodes, Mod, Fun, Args, infinity). + +%% @doc Perform an RPC call to a list of nodes in parallel, returning the +%% results in the same order as the input list. +-spec multi_rpc([node()], module(), atom(), [any()], timeout()) -> [any()]. +multi_rpc(Nodes, Mod, Fun, Args, Timeout) -> + pmap(fun(Node) -> + safe_rpc(Node, Mod, Fun, Args, Timeout) + end, Nodes). + +%% @doc Perform an RPC call to a list of nodes in parallel, returning the +%% results in the same order as the input list. Each result is tagged +%% with the corresponding node name. +-spec multi_rpc_ann([node()], module(), atom(), [any()]) + -> [{node(), any()}]. +multi_rpc_ann(Nodes, Mod, Fun, Args) -> + multi_rpc_ann(Nodes, Mod, Fun, Args, infinity). + +%% @doc Perform an RPC call to a list of nodes in parallel, returning the +%% results in the same order as the input list. Each result is tagged +%% with the corresponding node name. +-spec multi_rpc_ann([node()], module(), atom(), [any()], timeout()) + -> [{node(), any()}]. +multi_rpc_ann(Nodes, Mod, Fun, Args, Timeout) -> + Results = multi_rpc(Nodes, Mod, Fun, Args, Timeout), + lists:zip(Nodes, Results). + +%% @doc Similar to {@link rpc:multicall/4}. Performs an RPC call to a list +%% of nodes in parallel, returning a list of results as well as a list +%% of nodes that are down/unreachable. The results will be returned in +%% the same order as the input list, and each result is tagged with the +%% corresponding node name. +-spec multicall_ann([node()], module(), atom(), [any()]) + -> {Results :: [{node(), any()}], Down :: [node()]}. +multicall_ann(Nodes, Mod, Fun, Args) -> + multicall_ann(Nodes, Mod, Fun, Args, infinity). + +%% @doc Similar to {@link rpc:multicall/6}. Performs an RPC call to a list +%% of nodes in parallel, returning a list of results as well as a list +%% of nodes that are down/unreachable. The results will be returned in +%% the same order as the input list, and each result is tagged with the +%% corresponding node name. +-spec multicall_ann([node()], module(), atom(), [any()], timeout()) + -> {Results :: [{node(), any()}], Down :: [node()]}. +multicall_ann(Nodes, Mod, Fun, Args, Timeout) -> + L = multi_rpc_ann(Nodes, Mod, Fun, Args, Timeout), + {Results, DownAnn} = + lists:partition(fun({_, Result}) -> + Result /= {badrpc, nodedown} + end, L), + {Down, _} = lists:unzip(DownAnn), + {Results, Down}. + +%% @doc Convert a list of elements into an N-ary tree. This conversion +%% works by treating the list as an array-based tree where, for +%% example in a binary 2-ary tree, a node at index i has children +%% 2i and 2i+1. The conversion also supports a "cycles" mode where +%% the array is logically wrapped around to ensure leaf nodes also +%% have children by giving them backedges to other elements. + +-spec build_tree(N :: integer(), Nodes :: [term()], Opts :: [term()]) + -> orddict:orddict(). +build_tree(N, Nodes, Opts) -> + case lists:member(cycles, Opts) of + true -> + Expand = lists:flatten(lists:duplicate(N+1, Nodes)); + false -> + Expand = Nodes + end, + {Tree, _} = + lists:foldl(fun(Elm, {Result, Worklist}) -> + Len = erlang:min(N, length(Worklist)), + {Children, Rest} = lists:split(Len, Worklist), + NewResult = [{Elm, Children} | Result], + {NewResult, Rest} + end, {[], tl(Expand)}, Nodes), + orddict:from_list(Tree). + +orddict_delta(A, B) -> + %% Pad both A and B to the same length + DummyA = [{Key, '$none'} || {Key, _} <- B], + A2 = orddict:merge(fun(_, Value, _) -> + Value + end, A, DummyA), + + DummyB = [{Key, '$none'} || {Key, _} <- A], + B2 = orddict:merge(fun(_, Value, _) -> + Value + end, B, DummyB), + + %% Merge and filter out equal values + Merged = orddict:merge(fun(_, AVal, BVal) -> + {AVal, BVal} + end, A2, B2), + Diff = orddict:filter(fun(_, {Same, Same}) -> + false; + (_, _) -> + true + end, Merged), + Diff. + +shuffle(L) -> + N = 134217727, %% Largest small integer on 32-bit Erlang + L2 = [{rand:uniform(N), E} || E <- L], + L3 = [E || {_, E} <- lists:sort(L2)], + L3. + +%% Returns a forced-lowercase architecture for this node +-spec get_arch () -> string(). +get_arch () -> string:to_lower(erlang:system_info(system_architecture)). + +%% Checks if this node is of a given architecture +-spec is_arch (atom()) -> boolean(). +is_arch (linux) -> string:str(get_arch(),"linux") > 0; +is_arch (darwin) -> string:str(get_arch(),"darwin") > 0; +is_arch (sunos) -> string:str(get_arch(),"sunos") > 0; +is_arch (osx) -> is_arch(darwin); +is_arch (solaris) -> is_arch(sunos); +is_arch (Arch) -> throw({unsupported_architecture,Arch}). + +format_ip_and_port(Ip, Port) when is_list(Ip) -> + lists:flatten(io_lib:format("~s:~p",[Ip,Port])); +format_ip_and_port(Ip, Port) when is_tuple(Ip) -> + lists:flatten(io_lib:format("~s:~p",[inet_parse:ntoa(Ip), + Port])). +peername(Socket, Transport) -> + case Transport:peername(Socket) of + {ok, {Ip, Port}} -> + format_ip_and_port(Ip, Port); + {error, Reason} -> + %% just return a string so JSON doesn't blow up + lists:flatten(io_lib:format("error:~p", [Reason])) + end. + +sockname(Socket, Transport) -> + case Transport:sockname(Socket) of + {ok, {Ip, Port}} -> + format_ip_and_port(Ip, Port); + {error, Reason} -> + %% just return a string so JSON doesn't blow up + lists:flatten(io_lib:format("error:~p", [Reason])) + end. + +%% @doc Convert a #riak_core_fold_req_v? record to the cluster's maximum +%% supported record version. + +make_fold_req(#riak_core_fold_req_v1{foldfun=FoldFun, acc0=Acc0}) -> + make_fold_req(FoldFun, Acc0, false, []); +make_fold_req(?FOLD_REQ{foldfun=FoldFun, acc0=Acc0, + forwardable=Forwardable, opts=Opts}) -> + make_fold_req(FoldFun, Acc0, Forwardable, Opts). + +make_fold_req(FoldFun, Acc0) -> + make_fold_req(FoldFun, Acc0, false, []). + +make_fold_req(FoldFun, Acc0, Forwardable, Opts) -> + make_fold_reqv(riak_core_capability:get({riak_core, fold_req_version}, v1), + FoldFun, Acc0, Forwardable, Opts). + +%% @doc Force a #riak_core_fold_req_v? record to the latest version, +%% regardless of cluster support + +make_newest_fold_req(#riak_core_fold_req_v1{foldfun=FoldFun, acc0=Acc0}) -> + make_fold_reqv(v2, FoldFun, Acc0, false, []); +make_newest_fold_req(?FOLD_REQ{} = F) -> + F. + +%% @doc Spawn an intermediate proxy process to handle errors during gen_xxx +%% calls. +proxy_spawn(Fun) -> + %% Note: using spawn_monitor does not trigger selective receive + %% optimization, but spawn + monitor does. Silly Erlang. + Pid = spawn(?MODULE, proxy, [self(), Fun]), + MRef = monitor(process, Pid), + Pid ! {proxy, MRef}, + receive + {proxy_reply, MRef, Result} -> + demonitor(MRef, [flush]), + Result; + {'DOWN', MRef, _, _, Reason} -> + {error, Reason} + end. + + +%% @private +make_fold_reqv(v1, FoldFun, Acc0, _Forwardable, _Opts) + when is_function(FoldFun, 3) -> + #riak_core_fold_req_v1{foldfun=FoldFun, acc0=Acc0}; +make_fold_reqv(v2, FoldFun, Acc0, Forwardable, Opts) + when is_function(FoldFun, 3) + andalso (Forwardable == true orelse Forwardable == false) + andalso is_list(Opts) -> + ?FOLD_REQ{foldfun=FoldFun, acc0=Acc0, + forwardable=Forwardable, opts=Opts}. + +%% @private - used with proxy_spawn +proxy(Parent, Fun) -> + _ = monitor(process, Parent), + receive + {proxy, MRef} -> + Result = Fun(), + Parent ! {proxy_reply, MRef, Result}; + {'DOWN', _, _, _, _} -> + ok + end. + +-spec enable_job_class(atom(), atom()) -> ok | {error, term()}. +%% @doc Enables the specified Application/Operation job class. +%% This is the public API for use via RPC. +%% WARNING: This function is not suitable for parallel execution with itself +%% or its complement disable_job_class/2. +enable_job_class(Application, Operation) + when erlang:is_atom(Application) andalso erlang:is_atom(Operation) -> + enable_job_class({Application, Operation}); +enable_job_class(Application, Operation) -> + {error, {badarg, {Application, Operation}}}. + +-spec disable_job_class(atom(), atom()) -> ok | {error, term()}. +%% @doc Disables the specified Application/Operation job class. +%% This is the public API for use via RPC. +%% WARNING: This function is not suitable for parallel execution with itself +%% or its complement enable_job_class/2. +disable_job_class(Application, Operation) + when erlang:is_atom(Application) andalso erlang:is_atom(Operation) -> + disable_job_class({Application, Operation}); +disable_job_class(Application, Operation) -> + {error, {badarg, {Application, Operation}}}. + +-spec job_class_enabled(atom(), atom()) -> boolean() | {error, term()}. +%% @doc Reports whether the specified Application/Operation job class is enabled. +%% This is the public API for use via RPC. +job_class_enabled(Application, Operation) + when erlang:is_atom(Application) andalso erlang:is_atom(Operation) -> + job_class_enabled({Application, Operation}); +job_class_enabled(Application, Operation) -> + {error, {badarg, {Application, Operation}}}. + +-spec enable_job_class(Class :: term()) -> ok | {error, term()}. +%% @doc Internal API to enable the specified job class. +%% WARNING: +%% * This function may not remain in this form once the Jobs API is live! +%% * Parameter types ARE NOT validated by the same rules as the public API! +%% You are STRONGLY advised to use enable_job_class/2. +enable_job_class(Class) -> + case app_helper:get_env(riak_core, job_accept_class) of + [_|_] = EnabledClasses -> + case lists:member(Class, EnabledClasses) of + true -> + ok; + _ -> + application:set_env( + riak_core, job_accept_class, [Class | EnabledClasses]) + end; + _ -> + application:set_env(riak_core, job_accept_class, [Class]) + end. + +-spec disable_job_class(Class :: term()) -> ok | {error, term()}. +%% @doc Internal API to disable the specified job class. +%% WARNING: +%% * This function may not remain in this form once the Jobs API is live! +%% * Parameter types ARE NOT validated by the same rules as the public API! +%% You are STRONGLY advised to use disable_job_class/2. +disable_job_class(Class) -> + case app_helper:get_env(riak_core, job_accept_class) of + [_|_] = EnabledClasses -> + case lists:member(Class, EnabledClasses) of + false -> + ok; + _ -> + application:set_env(riak_core, job_accept_class, + lists:delete(Class, EnabledClasses)) + end; + _ -> + ok + end. + +-spec job_class_enabled(Class :: term()) -> boolean(). +%% @doc Internal API to determine whether to accept/reject a job. +%% WARNING: +%% * This function may not remain in this form once the Jobs API is live! +%% * Parameter types ARE NOT validated by the same rules as the public API! +%% You are STRONGLY advised to use job_class_enabled/2. +job_class_enabled(Class) -> + case app_helper:get_env(riak_core, job_accept_class) of + undefined -> + true; + [] -> + false; + [_|_] = EnabledClasses -> + lists:member(Class, EnabledClasses); + Other -> + % Don't crash if it's not a list - that should never be the case, + % but since the value *can* be manipulated externally be more + % accommodating. If someone mucks it up, nothing's going to be + % allowed, but give them a chance to catch on instead of crashing. + ?LOG_ERROR( + "riak_core.job_accept_class is not a list: ~p", [Other]), + false + end. + +-spec job_class_disabled_message(ReturnType :: atom(), Class :: term()) + -> binary() | string(). +%% @doc The error message to be returned to a client for a disabled job class. +%% WARNING: +%% * This function is likely to be extended to accept a Job as well as a Class +%% when the Jobs API is live. +job_class_disabled_message(binary, Class) -> + erlang:list_to_binary(job_class_disabled_message(text, Class)); +job_class_disabled_message(text, Class) -> + lists:flatten(io_lib:format("Operation '~p' is not enabled", [Class])). + +-spec report_job_request_disposition(Accepted :: boolean(), Class :: term(), + Mod :: module(), Func :: atom(), Line :: pos_integer(), Client :: term()) + -> ok | {error, term()}. +%% @doc Report/record the disposition of an async job request. +%% +%% Logs an appropriate message and reports to whoever needs to know. +%% WARNING: +%% * This function is likely to be extended to accept a Job as well as a Class +%% when the Jobs API is live. +%% +%% Parameters: +%% * Accepted - Whether the specified job Class is enabled. +%% * Class - The Class of the job, by convention {Application, Operation}. +%% * Mod/Func/Line - The Module, function, and source line number, +%% respectively, that will be reported as the source of the call. +%% * Client - Any term indicating the originator of the request. +%% By convention, when meaningful client identification information is not +%% available, Client is an atom representing the protocol through which the +%% request was received. +%% +report_job_request_disposition(true, Class, _Mod, _Func, _Line, Client) -> + ?LOG_DEBUG("Request '~p' accepted from ~p", [Class, Client]); +report_job_request_disposition(false, Class, _Mod, _Func, _Line, Client) -> + ?LOG_WARNING("Request '~p' disabled from ~p", [Class, Client]). + +%% =================================================================== +%% EUnit tests +%% =================================================================== +-ifdef(TEST). + +moment_test() -> + M1 = riak_core_util:moment(), + M2 = riak_core_util:moment(), + ?assert(M2 >= M1). + +clientid_uniqueness_test() -> + ClientIds = [mkclientid('somenode@somehost') || _I <- lists:seq(0, 10000)], + length(ClientIds) =:= length(sets:to_list(sets:from_list(ClientIds))). + +build_tree_test() -> + Flat = [1, + 11, 12, + 111, 112, 121, 122, + 1111, 1112, 1121, 1122, 1211, 1212, 1221, 1222], + + %% 2-ary tree decomposition + ATree = [{1, [ 11, 12]}, + {11, [ 111, 112]}, + {12, [ 121, 122]}, + {111, [1111, 1112]}, + {112, [1121, 1122]}, + {121, [1211, 1212]}, + {122, [1221, 1222]}, + {1111, []}, + {1112, []}, + {1121, []}, + {1122, []}, + {1211, []}, + {1212, []}, + {1221, []}, + {1222, []}], + + %% 2-ary tree decomposition with cyclic wrap-around + CTree = [{1, [ 11, 12]}, + {11, [ 111, 112]}, + {12, [ 121, 122]}, + {111, [1111, 1112]}, + {112, [1121, 1122]}, + {121, [1211, 1212]}, + {122, [1221, 1222]}, + {1111, [ 1, 11]}, + {1112, [ 12, 111]}, + {1121, [ 112, 121]}, + {1122, [ 122, 1111]}, + {1211, [1112, 1121]}, + {1212, [1122, 1211]}, + {1221, [1212, 1221]}, + {1222, [1222, 1]}], + + ?assertEqual(ATree, build_tree(2, Flat, [])), + ?assertEqual(CTree, build_tree(2, Flat, [cycles])), + ok. + + +counter_loop(N) -> + receive + {up, Pid} -> + N2=N+1, + Pid ! {counter_value, N2}, + counter_loop(N2); + down -> + counter_loop(N-1); + exit -> + exit(normal) + end. + +incr_counter(CounterPid) -> + CounterPid ! {up, self()}, + receive + {counter_value, N} -> N + after + 3000 -> + ?assert(false) + end. + +decr_counter(CounterPid) -> + CounterPid ! down. + +multi_keydelete_test_() -> + Languages = [{lisp, 1958}, + {ml, 1973}, + {erlang, 1986}, + {haskell, 1990}, + {ocaml, 1996}, + {clojure, 2007}, + {elixir, 2012}], + ?_assertMatch( + [{lisp, _}, {ml, _}, {erlang, _}, {haskell, _}], + multi_keydelete([ocaml, clojure, elixir], Languages)). + +compose_test_() -> + Upper = fun string:to_upper/1, + Reverse = fun lists:reverse/1, + Strip = fun(S) -> string:strip(S, both, $!) end, + Composed = compose([Upper, Reverse, Strip]), + ?_assertEqual("DLROW OLLEH", Composed("Hello world!")). + +pmap_test_() -> + Fgood = fun(X) -> 2 * X end, + Fbad = fun + (3) -> throw(die_on_3); + (X) -> Fgood(X) + end, + Lin = [1, 2, 3, 4], + Lout = [2, 4, 6, 8], + {setup, + fun() -> + logger:add_primary_filter(silence_crash, {fun + (#{meta := #{error_logger := #{emulator := true, tag := error}}}, _) -> + stop; + (_, _) -> + ignore + end, ?MODULE}) + end, + fun(_) -> logger:remove_primary_filter(silence_crash) end, + [ + fun() -> + % Test simple map case + ?assertEqual(Lout, pmap(Fgood, Lin)), + % Verify a crashing process will not stall pmap + Parent = self(), + Pid = spawn(fun() -> + % Caller trapping exits causes stall!! + % TODO: Consider pmapping in a spawned proc + % process_flag(trap_exit, true), + pmap(Fbad, Lin), + ?debugMsg("pmap finished just fine"), + Parent ! no_crash_yo + end), + MonRef = monitor(process, Pid), + receive + {'DOWN', MonRef, _, _, _} -> + ok; + no_crash_yo -> + ?assert(pmap_did_not_crash_as_expected) + end + end + ]}. + +bounded_pmap_test_() -> + Fun1 = fun(X) -> X+2 end, + Tests = + fun(CountPid) -> + GFun = fun(Max) -> + fun(X) -> + ?assert(incr_counter(CountPid) =< Max), + timer:sleep(1), + decr_counter(CountPid), + Fun1(X) + end + end, + [ + fun() -> + ?assertEqual(lists:seq(Fun1(1), Fun1(N)), + pmap(GFun(MaxP), + lists:seq(1, N), MaxP)) + end || + MaxP <- lists:seq(1,20), + N <- lists:seq(0,10) + ] + end, + {setup, + fun() -> + Pid = spawn_link(?MODULE, counter_loop, [0]), + monitor(process, Pid), + Pid + end, + fun(Pid) -> + Pid ! exit, + receive + {'DOWN', _Ref, process, Pid, _Info} -> ok + after + 3000 -> + ?debugMsg("pmap counter process did not go down in time"), + ?assert(false) + end, + ok + end, + Tests + }. + +make_fold_req_test_() -> + {setup, + fun() -> + meck:new(riak_core_capability, [passthrough]) + end, + fun(_) -> + meck:unload(riak_core_capability) + end, + [ + fun() -> + FoldFun = fun(_, _, _) -> ok end, + Acc0 = acc0, + Forw = true, + Opts = [opts], + F_1 = #riak_core_fold_req_v1{foldfun=FoldFun, acc0=Acc0}, + F_2 = #riak_core_fold_req_v2{foldfun=FoldFun, acc0=Acc0, + forwardable=Forw, opts=Opts}, + F_2_default = #riak_core_fold_req_v2{foldfun=FoldFun, acc0=Acc0, + forwardable=false, opts=[]}, + Newest = fun() -> F_2_default = make_newest_fold_req(F_1), + F_2 = make_newest_fold_req(F_2), + ok + end, + + meck:expect(riak_core_capability, get, + fun(_, _) -> v1 end), + F_1 = make_fold_req(F_1), + F_1 = make_fold_req(F_2), + F_1 = make_fold_req(FoldFun, Acc0), + F_1 = make_fold_req(FoldFun, Acc0, Forw, Opts), + ok = Newest(), + + meck:expect(riak_core_capability, get, + fun(_, _) -> v2 end), + F_2_default = make_fold_req(F_1), + F_2 = make_fold_req(F_2), + F_2_default = make_fold_req(FoldFun, Acc0), + F_2 = make_fold_req(FoldFun, Acc0, Forw, Opts), + ok = Newest() + end + ] + }. + +proxy_spawn_test() -> + A = proxy_spawn(fun() -> a end), + ?assertEqual(a, A), + B = proxy_spawn(fun() -> exit(killer_fun) end), + ?assertEqual({error, killer_fun}, B), + + %% Ensure no errant 'DOWN' messages + receive + {'DOWN', _, _, _, _}=Msg -> + throw({error, {badmsg, Msg}}); + _ -> + ok + after 1000 -> + ok + end. + +-endif. + diff --git a/src/riak_core_vm_mon.erl b/src/riak_core_vm_mon.erl new file mode 100644 index 00000000..bd87718e --- /dev/null +++ b/src/riak_core_vm_mon.erl @@ -0,0 +1,186 @@ +%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- +%% ------------------------------------------------------------------- +%% +%% Copyright (c) 2024 Workday, Inc. +%% +%% This file is provided to you 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. +%% +%% ------------------------------------------------------------------- +%% +%% @doc Reports VM statistics. +%% +%% This is a separate module to allow for future expansion into a `gen_server' +%% monitoring VM stats and reporting when they exceed threshholds, possibly +%% with the ability to shut Riak down gracefully before the VM crashes due to +%% hitting a hard limit. +%% +-module(riak_core_vm_mon). + +%% Public API +-export([ + vm_stats/0 +]). + +%% Public Types +%% Values are just aliases for integer ranges given descriptive names. +-export_type([ + vm_atom_count/0, vm_atom_limit/0, vm_atom_pct/0, + vm_ets_count/0, vm_ets_limit/0, vm_ets_pct/0, + vm_port_count/0, vm_port_limit/0, vm_port_pct/0, + vm_proc_count/0, vm_proc_limit/0, vm_proc_pct/0, + vm_stat_count/0, vm_stat_limit/0, vm_stat_pct/0, + vm_stats/0 +]). + +-type vm_stats() :: { + vm_atom_limit(), vm_atom_count(), vm_atom_pct(), + vm_ets_limit(), vm_ets_count(), vm_ets_pct(), + vm_port_limit(), vm_port_count(), vm_port_pct(), + vm_proc_limit(), vm_proc_count(), vm_proc_pct() +}. + +-type vm_stat_count() :: non_neg_integer(). +-type vm_stat_limit() :: pos_integer(). +-type vm_stat_pct() :: 0..100. + +-type vm_atom_count() :: vm_stat_count(). +-type vm_atom_limit() :: vm_stat_limit(). +-type vm_atom_pct() :: vm_stat_pct(). +-type vm_ets_count() :: vm_stat_count(). +-type vm_ets_limit() :: vm_stat_limit(). +-type vm_ets_pct() :: vm_stat_pct(). +-type vm_port_count() :: vm_stat_count(). +-type vm_port_limit() :: vm_stat_limit(). +-type vm_port_pct() :: vm_stat_pct(). +-type vm_proc_count() :: vm_stat_count(). +-type vm_proc_limit() :: vm_stat_limit(). +-type vm_proc_pct() :: vm_stat_pct(). + +-include_lib("kernel/include/logger.hrl"). +-ifdef(TEST). +-include_lib("eunit/include/eunit.hrl"). +-endif. + +%% =================================================================== +%% Public API +%% =================================================================== + +-spec vm_stats() -> vm_stats(). +%% @doc Returns the current VM stats in a flat tuple. +%% +%% The result tuple contains three values for classes of Atoms, ETS tables, +%% Ports, and Processes, in that order. +%% For each class, the three integers are Limit, Count, and PercentUsed. +%% @end +%% Several approaches have been implemented and timed, with and without +%% caching constant values in the process dictionary; this is the fastest. +vm_stats() -> + AtomLimit = erlang:system_info(atom_limit), + AtomCount = erlang:system_info(atom_count), + EtsLimit = erlang:system_info(ets_limit), + EtsCount = erlang:system_info(ets_count), + PortLimit = erlang:system_info(port_limit), + PortCount = erlang:system_info(port_count), + ProcLimit = erlang:system_info(process_limit), + ProcCount = erlang:system_info(process_count), + %% PercentUsed are calculated with a faster integer arithmetic + %% equivalent of: + %% erlang:round((Count * 100) / Limit) + %% Counter-intuitively, (Limit div 2) is consistently very slightly + %% faster than (Limit bsr 1); not digging into why but I suspect + %% parameter range checking. + AtomPct = (((AtomCount * 100) + (AtomLimit div 2)) div AtomLimit), + EtsPct = (((EtsCount * 100) + (EtsLimit div 2)) div EtsLimit ), + PortPct = (((PortCount * 100) + (PortLimit div 2)) div PortLimit), + ProcPct = (((ProcCount * 100) + (ProcLimit div 2)) div ProcLimit), + %% Calculating the percentages into discrete values above appears to be + %% slightly faster than doing it inside the tuple construction. + %% The difference is small but consistent; not digging further to figure + %% out why though I suspect JIT inlining behavior to be the root cause. + { + AtomLimit, AtomCount, AtomPct, EtsLimit, EtsCount, EtsPct, + PortLimit, PortCount, PortPct, ProcLimit, ProcCount, ProcPct + }. + +%% =================================================================== +%% Tests +%% =================================================================== + +-ifdef(TEST). + +-define(STATS_LMT_OFF, 0). +-define(STATS_CNT_OFF, 1). +-define(STATS_PCT_OFF, 2). + +-define(ATOM_REC_POS, 1). +-define(ETS_REC_POS, 4). +-define(PORT_REC_POS, 7). +-define(PROC_REC_POS, 10). + +-define(STATS_LEN, 12). +-define(STATS_REC_POS, [ + ?ATOM_REC_POS, ?ETS_REC_POS, ?PORT_REC_POS, ?PROC_REC_POS]). + +vm_stats_test() -> + Stats0 = vm_stats(), + validate_stats(Stats0), + + %% Make some new atoms to ensure the count increases. + % R will be in the range 100000001..200000000 + % random numbers in atoms will be 9 digits in the range (R+1)..(R*2) + R = (100000000 + rand:uniform(100000000)), + NewAtoms = [ + erlang:list_to_atom(lists:concat( + [?FUNCTION_NAME, "_", [$a + C], "_", (R + rand:uniform(R))])) + || C <- lists:seq(0, 7)], + + Stats1 = vm_stats(), + validate_stats(Stats1), + lists:foreach( + fun(Pos) -> + LimP = (Pos + ?STATS_LMT_OFF), + CntP = (Pos + ?STATS_CNT_OFF), + PctP = (Pos + ?STATS_PCT_OFF), + ?assertEqual( + erlang:element(LimP, Stats1), erlang:element(LimP, Stats0)), + ?assert( + erlang:element(CntP, Stats1) >= erlang:element(CntP, Stats0)), + ?assert( + erlang:element(PctP, Stats1) >= erlang:element(PctP, Stats0)) + end, ?STATS_REC_POS), + + AtomCntP = (?ATOM_REC_POS + ?STATS_CNT_OFF), + AtomCnt0 = erlang:element(AtomCntP, Stats0), + AtomCnt1 = erlang:element(AtomCntP, Stats1), + ?assertEqual(AtomCnt1, (AtomCnt0 + erlang:length(NewAtoms))). + +validate_stats(Stats) -> + ?assert(erlang:is_tuple(Stats)), + ?assertEqual(?STATS_LEN, erlang:tuple_size(Stats)), + lists:foreach( + fun(Pos) -> + ?assert(erlang:is_integer(erlang:element(Pos, Stats))) + end, lists:seq(1, ?STATS_LEN)), + lists:foreach( + fun(Pos) -> + Lmt = erlang:element((Pos + ?STATS_LMT_OFF), Stats), + Cnt = erlang:element((Pos + ?STATS_CNT_OFF), Stats), + Pct = erlang:element((Pos + ?STATS_PCT_OFF), Stats), + ?assert(Lmt > 0), + ?assert(Cnt >= 0 andalso Lmt >= Cnt), + ?assert(Pct >= 0 andalso 100 >= Pct) + end, ?STATS_REC_POS). + +-endif. % ?TEST diff --git a/test/sync_command_test.erl b/test/sync_command_test.erl index bae37eec..512f31c3 100644 --- a/test/sync_command_test.erl +++ b/test/sync_command_test.erl @@ -1,6 +1,7 @@ +%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- %% ------------------------------------------------------------------- %% -%% Copyright (c) 2007-2011 Basho Technologies, Inc. All Rights Reserved. +%% Copyright (c) 2013-2014 Basho Technologies, Inc. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file @@ -25,7 +26,7 @@ sync_test_() -> {foreach, fun setup_simple/0, - fun stop_servers/1, + fun cleanup_simple/1, [ {<<"Assert ok throw">>, fun() -> ?assertEqual(ok, mock_vnode:sync_error({0, node()}, goodthrow)) @@ -55,14 +56,14 @@ sync_test_() -> setup_simple() -> - stop_servers(self()), + LogLevel = riak_core_test_util:logger_silence(), + stop_servers(), Vars = [{ring_creation_size, 8}, {ring_state_dir, ""}, %% Don't allow rolling start of vnodes as it will cause a %% race condition with `all_nodes'. {core_vnode_eqc_pool_size, 0}, {vnode_rolling_start, 0}], - error_logger:tty(false), _ = [begin Old = app_helper:get_env(riak_core, AppKey), ok = application:set_env(riak_core, AppKey, Val), @@ -84,17 +85,22 @@ setup_simple() -> %% the vnode mgr before this call have been handled. This %% guarantees that the vnode mgr ets tab is up-to-date - riak_core:register([{vnode_module, mock_vnode}]). + riak_core:register([{vnode_module, mock_vnode}]), + LogLevel. +cleanup_simple(LogLevel) -> + stop_servers(), + logger:set_handler_config(default, level, LogLevel). -stop_servers(_Pid) -> +stop_servers() -> %% Make sure VMaster is killed before sup as start_vnode is a cast %% and there may be a pending request to start the vnode. stop_pid(whereis(mock_vnode_master)), stop_pid(whereis(riak_core_vnode_manager)), stop_pid(whereis(riak_core_vnode_events)), stop_pid(whereis(riak_core_vnode_sup)), - application:stop(exometer). + application:stop(exometer), + logger:remove_primary_filter(silence_logs). stop_pid(undefined) -> ok; diff --git a/test/worker_pool_test.erl b/test/worker_pool_test.erl index d4031f22..69feb6ed 100644 --- a/test/worker_pool_test.erl +++ b/test/worker_pool_test.erl @@ -1,6 +1,7 @@ +%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- %% ------------------------------------------------------------------- %% -%% Copyright (c) 2007-2011 Basho Technologies, Inc. All Rights Reserved. +%% Copyright (c) 2011 Basho Technologies, Inc. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file @@ -18,12 +19,12 @@ %% %% ------------------------------------------------------------------- -module(worker_pool_test). - -behaviour(riak_core_vnode_worker). --include_lib("eunit/include/eunit.hrl"). -export([init_worker/3, handle_work/3]). +-include_lib("eunit/include/eunit.hrl"). + init_worker(_VnodeIndex, Noreply, _WorkerProps) -> {ok, Noreply}. @@ -86,7 +87,7 @@ simple_node_worker_pool() -> [ ?assertEqual(true, receive_result(N)) || N <- lists:seq(1, 10)], unlink(BestEffortPool), riak_core_node_worker_pool:stop(BestEffortPool, normal), - + {ok, AssuredForwardingPool} = riak_core_node_worker_pool:start_link(?MODULE, 5, @@ -131,16 +132,14 @@ simple_noreply_worker_pool() -> pool_test_() -> {setup, - fun() -> - error_logger:tty(false) - end, - fun(_) -> - error_logger:tty(true) + fun riak_core_test_util:logger_silence/0, + fun(Level) -> + logger:set_handler_config(default, level, Level) end, [ fun simple_worker_pool/0, fun simple_noreply_worker_pool/0, - fun simple_node_worker_pool/0 + fun simple_node_worker_pool/0 ] }.