| % Licensed under the Apache License, Version 2.0 (the "License"); you may not |
| % use this file except in compliance with the License. You may obtain a copy of |
| % the License at |
| % |
| % http://www.apache.org/licenses/LICENSE-2.0 |
| % |
| % Unless required by applicable law or agreed to in writing, software |
| % distributed under the License is distributed on an "AS IS" BASIS, WITHOUT |
| % WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the |
| % License for the specific language governing permissions and limitations under |
| % the License. |
| |
| |
| %% -*- erlang-indent-level: 4;indent-tabs-mode: nil -*- |
| |
| -module(dreyfus_fabric_group1). |
| |
| -include("dreyfus.hrl"). |
| -include_lib("mem3/include/mem3.hrl"). |
| -include_lib("couch/include/couch_db.hrl"). |
| |
| -export([go/4]). |
| |
| -record(state, { |
| limit, |
| sort, |
| top_groups, |
| counters, |
| start_args, |
| replacements, |
| ring_opts |
| }). |
| |
| go(DbName, GroupId, IndexName, QueryArgs) when is_binary(GroupId) -> |
| {ok, DDoc} = fabric:open_doc(DbName, <<"_design/", GroupId/binary>>, []), |
| dreyfus_util:maybe_deny_index(DbName, GroupId, IndexName), |
| go(DbName, DDoc, IndexName, QueryArgs); |
| |
| go(DbName, DDoc, IndexName, #index_query_args{}=QueryArgs) -> |
| DesignName = dreyfus_util:get_design_docid(DDoc), |
| dreyfus_util:maybe_deny_index(DbName, DesignName, IndexName), |
| Shards = dreyfus_util:get_shards(DbName, QueryArgs), |
| RingOpts = dreyfus_util:get_ring_opts(QueryArgs, Shards), |
| Workers = fabric_util:submit_jobs(Shards, dreyfus_rpc, group1, [DDoc, |
| IndexName, dreyfus_util:export(QueryArgs)]), |
| Replacements = fabric_view:get_shard_replacements(DbName, Workers), |
| Counters = fabric_dict:init(Workers, nil), |
| RexiMon = fabric_util:create_monitors(Workers), |
| State = #state{ |
| limit = QueryArgs#index_query_args.grouping#grouping.limit, |
| sort = QueryArgs#index_query_args.grouping#grouping.sort, |
| top_groups = [], |
| counters = Counters, |
| start_args = [DDoc, IndexName, QueryArgs], |
| replacements = Replacements, |
| ring_opts = RingOpts |
| }, |
| try |
| rexi_utils:recv(Workers, #shard.ref, fun handle_message/3, |
| State, infinity, 1000 * 60 * 60) |
| after |
| rexi_monitor:stop(RexiMon), |
| fabric_util:cleanup(Workers) |
| end; |
| go(DbName, DDoc, IndexName, OldArgs) -> |
| go(DbName, DDoc, IndexName, dreyfus_util:upgrade(OldArgs)). |
| |
| handle_message({ok, NewTopGroups}, Shard, State0) -> |
| State = upgrade_state(State0), |
| #state{top_groups=TopGroups, limit=Limit, sort=Sort} = State, |
| case fabric_dict:lookup_element(Shard, State#state.counters) of |
| undefined -> |
| %% already heard from someone else in this range |
| {ok, State}; |
| nil -> |
| C1 = fabric_dict:store(Shard, ok, State#state.counters), |
| C2 = fabric_view:remove_overlapping_shards(Shard, C1), |
| MergedTopGroups = merge_top_groups(TopGroups, make_sortable(Shard, NewTopGroups), Limit, Sort), |
| State1 = State#state{ |
| counters=C2, |
| top_groups=MergedTopGroups |
| }, |
| case fabric_dict:any(nil, C2) of |
| true -> |
| {ok, State1}; |
| false -> |
| {stop, remove_sortable(MergedTopGroups)} |
| end |
| end; |
| |
| handle_message(Error, Worker, State0) -> |
| State = upgrade_state(State0), |
| case dreyfus_fabric:handle_error_message(Error, Worker, |
| State#state.counters, State#state.replacements, |
| group1, State#state.start_args, State#state.ring_opts) of |
| {ok, Counters} -> |
| {ok, State#state{counters=Counters}}; |
| {new_refs, NewRefs, NewCounters, NewReplacements} -> |
| NewState = State#state{ |
| counters = NewCounters, |
| replacements = NewReplacements |
| }, |
| {new_refs, NewRefs, NewState}; |
| Else -> |
| Else |
| end. |
| |
| merge_top_groups(TopGroupsA, TopGroupsB, Limit, Sort) -> |
| MergedGroups0 = TopGroupsA ++ TopGroupsB, |
| GNs = lists:usort([N || #sortable{item={N,_}} <- MergedGroups0]), |
| MergedGroups = [merge_top_group(Sort, [S || #sortable{item={N,_}}=S <- MergedGroups0, N =:= GN]) || GN <- GNs], |
| lists:sublist(dreyfus_util:sort(Sort, MergedGroups), Limit). |
| |
| merge_top_group(_Sort, [Group]) -> |
| Group; |
| merge_top_group(Sort, [_, _] = Groups) -> |
| hd(dreyfus_util:sort(Sort, Groups)). |
| |
| make_sortable(Shard, TopGroups) -> |
| [#sortable{item=G, order=Order, shard=Shard} || {_Name, Order}=G <- TopGroups]. |
| |
| remove_sortable(Sortables) -> |
| [Item || #sortable{item=Item} <- Sortables]. |
| |
| upgrade_state({state, Limit, Sort, TopGroups, Counters}) -> |
| #state{limit=Limit, sort=Sort, top_groups=TopGroups, counters=Counters, |
| replacements=[]}; |
| upgrade_state(#state{}=State) -> |
| State. |