diff options
| -rw-r--r-- | src/rabbit.erl | 4 | ||||
| -rw-r--r-- | src/rabbit_mnesia.erl | 292 | ||||
| -rw-r--r-- | src/rabbit_table.erl | 311 | ||||
| -rw-r--r-- | src/rabbit_upgrade.erl | 5 | ||||
| -rw-r--r-- | src/rabbit_upgrade_functions.erl | 4 |
5 files changed, 324 insertions, 292 deletions
diff --git a/src/rabbit.erl b/src/rabbit.erl index 353a5bb96f..afa97ddc86 100644 --- a/src/rabbit.erl +++ b/src/rabbit.erl @@ -419,7 +419,7 @@ stop(_State) -> ok = rabbit_alarm:stop(), ok = case rabbit_mnesia:is_clustered() of true -> rabbit_amqqueue:on_node_down(node()); - false -> rabbit_mnesia:empty_ram_only_tables() + false -> rabbit_table:clear_ram_only_tables() end, ok. @@ -546,7 +546,7 @@ recover() -> rabbit_binding:recover(rabbit_exchange:recover(), rabbit_amqqueue:start()). maybe_insert_default_data() -> - case rabbit_mnesia:is_db_empty() of + case rabbit_table:is_empty() of true -> insert_default_data(); false -> ok end. diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl index c4920dc481..f7a355be6b 100644 --- a/src/rabbit_mnesia.erl +++ b/src/rabbit_mnesia.erl @@ -14,7 +14,6 @@ %% Copyright (c) 2007-2012 VMware, Inc. All rights reserved. %% - -module(rabbit_mnesia). -export([init/0, @@ -26,18 +25,14 @@ forget_cluster_node/2, status/0, - is_db_empty/0, is_clustered/0, cluster_nodes/1, node_type/0, dir/0, - table_names/0, cluster_status_from_mnesia/0, init_db_unchecked/2, - empty_ram_only_tables/0, copy_db/1, - wait_for_tables/1, check_cluster_consistency/0, ensure_mnesia_dir/0, @@ -51,10 +46,6 @@ is_running_remote/0 ]). -%% create_tables/0 exported for helping embed RabbitMQ in or alongside -%% other mnesia-using Erlang applications, such as ejabberd --export([create_tables/0]). - -include("rabbit.hrl"). %%---------------------------------------------------------------------------- @@ -78,21 +69,16 @@ %% Various queries to get the status of the db -spec(status/0 :: () -> [{'nodes', [{node_type(), [node()]}]} | {'running_nodes', [node()]}]). --spec(is_db_empty/0 :: () -> boolean()). -spec(is_clustered/0 :: () -> boolean()). -spec(cluster_nodes/1 :: ('all' | 'disc' | 'ram' | 'running') -> [node()]). -spec(node_type/0 :: () -> node_type()). -spec(dir/0 :: () -> file:filename()). --spec(table_names/0 :: () -> [atom()]). -spec(cluster_status_from_mnesia/0 :: () -> rabbit_types:ok_or_error2( cluster_status(), any())). %% Operations on the db and utils, mainly used in `rabbit_upgrade' and `rabbit' -spec(init_db_unchecked/2 :: ([node()], node_type()) -> 'ok'). --spec(empty_ram_only_tables/0 :: () -> 'ok'). --spec(create_tables/0 :: () -> 'ok'). -spec(copy_db/1 :: (file:filename()) -> rabbit_types:ok_or_error(any())). --spec(wait_for_tables/1 :: ([atom()]) -> 'ok'). -spec(check_cluster_consistency/0 :: () -> 'ok'). -spec(ensure_mnesia_dir/0 :: () -> 'ok'). @@ -307,7 +293,7 @@ remove_node_offline_node(Node) -> case cluster_nodes(running) -- [node(), Node] of [] -> start_mnesia(), try - [mnesia:force_load_table(T) || T <- table_names()], + rabbit_table:force_load(), forget_cluster_node(Node, false), ensure_mnesia_running() after @@ -335,10 +321,6 @@ status() -> no -> [] end. -is_db_empty() -> - lists:all(fun (Tab) -> mnesia:dirty_first(Tab) == '$end_of_table' end, - table_names()). - is_clustered() -> AllNodes = cluster_nodes(all), AllNodes =/= [] andalso AllNodes =/= [node()]. @@ -358,9 +340,7 @@ mnesia_nodes() -> true -> disc; false -> ram end, - Tables = mnesia:system_info(tables), - [{Table, _} | _] = table_definitions(NodeType), - case lists:member(Table, Tables) of + case rabbit_table:is_present() of true -> AllNodes = mnesia:system_info(db_nodes), DiscCopies = mnesia:table_info(schema, disc_copies), DiscNodes = case NodeType of @@ -415,8 +395,6 @@ node_type() -> dir() -> mnesia:system_info(directory). -table_names() -> [Tab || {Tab, _} <- table_definitions()]. - %%---------------------------------------------------------------------------- %% Operations on the db %%---------------------------------------------------------------------------- @@ -446,18 +424,8 @@ init_db(ClusterNodes, NodeType, CheckOtherNodes) -> %% Subsequent node in cluster, catch up ensure_version_ok( rpc:call(AnotherNode, rabbit_version, recorded, [])), - ok = wait_for_replicated_tables(), - %% The sequence in which we delete the schema and then the - %% other tables is important: if we delete the schema - %% first when moving to RAM mnesia will loudly complain - %% since it doesn't make much sense to do that. But when - %% moving to disc, we need to move the schema first. - case NodeType of - disc -> create_local_table_copy(schema, disc_copies), - create_local_table_copies(disc); - ram -> create_local_table_copies(ram), - create_local_table_copy(schema, ram_copies) - end + ok = rabbit_table:wait_for_replicated(), + ok = rabbit_table:create_local_copy(NodeType) end, ensure_schema_integrity(), rabbit_node_monitor:update_cluster_status(), @@ -478,7 +446,7 @@ init_db_and_upgrade(ClusterNodes, NodeType, CheckOtherNodes) -> case NodeType of ram -> start_mnesia(), change_extra_db_nodes(ClusterNodes, false), - wait_for_replicated_tables(); + rabbit_table:wait_for_replicated(); disc -> ok end, ok. @@ -524,70 +492,17 @@ ensure_mnesia_not_running() -> end. ensure_schema_integrity() -> - case check_schema_integrity() of + case rabbit_table:check_schema_integrity() of ok -> ok; {error, Reason} -> throw({error, {schema_integrity_check_failed, Reason}}) end. -check_schema_integrity() -> - Tables = mnesia:system_info(tables), - case check_tables(fun (Tab, TabDef) -> - case lists:member(Tab, Tables) of - false -> {error, {table_missing, Tab}}; - true -> check_table_attributes(Tab, TabDef) - end - end) of - ok -> ok = wait_for_tables(table_names()), - check_tables(fun check_table_content/2); - Other -> Other - end. - -empty_ram_only_tables() -> - Node = node(), - lists:foreach( - fun (TabName) -> - case lists:member(Node, mnesia:table_info(TabName, ram_copies)) of - true -> {atomic, ok} = mnesia:clear_table(TabName); - false -> ok - end - end, table_names()), - ok. - -create_tables() -> create_tables(disc). - -create_tables(Type) -> - lists:foreach(fun ({Tab, TabDef}) -> - TabDef1 = proplists:delete(match, TabDef), - case mnesia:create_table(Tab, TabDef1) of - {atomic, ok} -> ok; - {aborted, Reason} -> - throw({error, {table_creation_failed, - Tab, TabDef1, Reason}}) - end - end, - table_definitions(Type)), - ok. - copy_db(Destination) -> ok = ensure_mnesia_not_running(), rabbit_file:recursive_copy(dir(), Destination). -wait_for_replicated_tables() -> - wait_for_tables([Tab || {Tab, TabDef} <- table_definitions(), - not lists:member({local_content, true}, TabDef)]). - -wait_for_tables(TableNames) -> - case mnesia:wait_for_tables(TableNames, 30000) of - ok -> - ok; - {timeout, BadTabs} -> - throw({error, {timeout_waiting_for_tables, BadTabs}}); - {error, Reason} -> - throw({error, {failed_waiting_for_tables, Reason}}) - end. - %% This does not guarantee us much, but it avoids some situations that %% will definitely end up badly check_cluster_consistency() -> @@ -682,158 +597,8 @@ discover_cluster(Node) -> end end. -%% The tables aren't supposed to be on disk on a ram node -table_definitions(disc) -> - table_definitions(); -table_definitions(ram) -> - [{Tab, copy_type_to_ram(TabDef)} || {Tab, TabDef} <- table_definitions()]. - -table_definitions() -> - [{rabbit_user, - [{record_name, internal_user}, - {attributes, record_info(fields, internal_user)}, - {disc_copies, [node()]}, - {match, #internal_user{_='_'}}]}, - {rabbit_user_permission, - [{record_name, user_permission}, - {attributes, record_info(fields, user_permission)}, - {disc_copies, [node()]}, - {match, #user_permission{user_vhost = #user_vhost{_='_'}, - permission = #permission{_='_'}, - _='_'}}]}, - {rabbit_vhost, - [{record_name, vhost}, - {attributes, record_info(fields, vhost)}, - {disc_copies, [node()]}, - {match, #vhost{_='_'}}]}, - {rabbit_listener, - [{record_name, listener}, - {attributes, record_info(fields, listener)}, - {type, bag}, - {match, #listener{_='_'}}]}, - {rabbit_durable_route, - [{record_name, route}, - {attributes, record_info(fields, route)}, - {disc_copies, [node()]}, - {match, #route{binding = binding_match(), _='_'}}]}, - {rabbit_semi_durable_route, - [{record_name, route}, - {attributes, record_info(fields, route)}, - {type, ordered_set}, - {match, #route{binding = binding_match(), _='_'}}]}, - {rabbit_route, - [{record_name, route}, - {attributes, record_info(fields, route)}, - {type, ordered_set}, - {match, #route{binding = binding_match(), _='_'}}]}, - {rabbit_reverse_route, - [{record_name, reverse_route}, - {attributes, record_info(fields, reverse_route)}, - {type, ordered_set}, - {match, #reverse_route{reverse_binding = reverse_binding_match(), - _='_'}}]}, - {rabbit_topic_trie_node, - [{record_name, topic_trie_node}, - {attributes, record_info(fields, topic_trie_node)}, - {type, ordered_set}, - {match, #topic_trie_node{trie_node = trie_node_match(), _='_'}}]}, - {rabbit_topic_trie_edge, - [{record_name, topic_trie_edge}, - {attributes, record_info(fields, topic_trie_edge)}, - {type, ordered_set}, - {match, #topic_trie_edge{trie_edge = trie_edge_match(), _='_'}}]}, - {rabbit_topic_trie_binding, - [{record_name, topic_trie_binding}, - {attributes, record_info(fields, topic_trie_binding)}, - {type, ordered_set}, - {match, #topic_trie_binding{trie_binding = trie_binding_match(), - _='_'}}]}, - {rabbit_durable_exchange, - [{record_name, exchange}, - {attributes, record_info(fields, exchange)}, - {disc_copies, [node()]}, - {match, #exchange{name = exchange_name_match(), _='_'}}]}, - {rabbit_exchange, - [{record_name, exchange}, - {attributes, record_info(fields, exchange)}, - {match, #exchange{name = exchange_name_match(), _='_'}}]}, - {rabbit_exchange_serial, - [{record_name, exchange_serial}, - {attributes, record_info(fields, exchange_serial)}, - {match, #exchange_serial{name = exchange_name_match(), _='_'}}]}, - {rabbit_runtime_parameters, - [{record_name, runtime_parameters}, - {attributes, record_info(fields, runtime_parameters)}, - {disc_copies, [node()]}, - {match, #runtime_parameters{_='_'}}]}, - {rabbit_durable_queue, - [{record_name, amqqueue}, - {attributes, record_info(fields, amqqueue)}, - {disc_copies, [node()]}, - {match, #amqqueue{name = queue_name_match(), _='_'}}]}, - {rabbit_queue, - [{record_name, amqqueue}, - {attributes, record_info(fields, amqqueue)}, - {match, #amqqueue{name = queue_name_match(), _='_'}}]}] - ++ gm:table_definitions() - ++ mirrored_supervisor:table_definitions(). - -binding_match() -> - #binding{source = exchange_name_match(), - destination = binding_destination_match(), - _='_'}. -reverse_binding_match() -> - #reverse_binding{destination = binding_destination_match(), - source = exchange_name_match(), - _='_'}. -binding_destination_match() -> - resource_match('_'). -trie_node_match() -> - #trie_node{ exchange_name = exchange_name_match(), _='_'}. -trie_edge_match() -> - #trie_edge{ exchange_name = exchange_name_match(), _='_'}. -trie_binding_match() -> - #trie_binding{exchange_name = exchange_name_match(), _='_'}. -exchange_name_match() -> - resource_match(exchange). -queue_name_match() -> - resource_match(queue). -resource_match(Kind) -> - #resource{kind = Kind, _='_'}. - -check_table_attributes(Tab, TabDef) -> - {_, ExpAttrs} = proplists:lookup(attributes, TabDef), - case mnesia:table_info(Tab, attributes) of - ExpAttrs -> ok; - Attrs -> {error, {table_attributes_mismatch, Tab, ExpAttrs, Attrs}} - end. - -check_table_content(Tab, TabDef) -> - {_, Match} = proplists:lookup(match, TabDef), - case mnesia:dirty_first(Tab) of - '$end_of_table' -> - ok; - Key -> - ObjList = mnesia:dirty_read(Tab, Key), - MatchComp = ets:match_spec_compile([{Match, [], ['$_']}]), - case ets:match_spec_run(ObjList, MatchComp) of - ObjList -> ok; - _ -> {error, {table_content_invalid, Tab, Match, ObjList}} - end - end. - -check_tables(Fun) -> - case [Error || {Tab, TabDef} <- table_definitions(node_type()), - case Fun(Tab, TabDef) of - ok -> Error = none, false; - {error, Error} -> true - end] of - [] -> ok; - Errors -> {error, Errors} - end. - schema_ok_or_move() -> - case check_schema_integrity() of + case rabbit_table:check_schema_integrity() of ok -> ok; {error, Reason} -> @@ -862,7 +627,7 @@ create_schema() -> stop_mnesia(), rabbit_misc:ensure_ok(mnesia:create_schema([node()]), cannot_create_schema), start_mnesia(), - ok = create_tables(disc), + ok = rabbit_table:create(), ensure_schema_integrity(), ok = rabbit_version:record_desired(). @@ -887,47 +652,6 @@ move_db() -> start_mnesia(), ok. -copy_type_to_ram(TabDef) -> - [{disc_copies, []}, {ram_copies, [node()]} - | proplists:delete(ram_copies, proplists:delete(disc_copies, TabDef))]. - -table_has_copy_type(TabDef, DiscType) -> - lists:member(node(), proplists:get_value(DiscType, TabDef, [])). - -create_local_table_copies(Type) -> - lists:foreach( - fun ({Tab, TabDef}) -> - HasDiscCopies = table_has_copy_type(TabDef, disc_copies), - HasDiscOnlyCopies = table_has_copy_type(TabDef, disc_only_copies), - LocalTab = proplists:get_bool(local_content, TabDef), - StorageType = - if - Type =:= disc orelse LocalTab -> - if - HasDiscCopies -> disc_copies; - HasDiscOnlyCopies -> disc_only_copies; - true -> ram_copies - end; - Type =:= ram -> - ram_copies - end, - ok = create_local_table_copy(Tab, StorageType) - end, - table_definitions(Type)), - ok. - -create_local_table_copy(Tab, Type) -> - StorageType = mnesia:table_info(Tab, storage_type), - {atomic, ok} = - if - StorageType == unknown -> - mnesia:add_table_copy(Tab, node(), Type); - StorageType /= Type -> - mnesia:change_table_copy_type(Tab, node(), Type); - true -> {atomic, ok} - end, - ok. - remove_node_if_mnesia_running(Node) -> case mnesia:system_info(is_running) of yes -> diff --git a/src/rabbit_table.erl b/src/rabbit_table.erl new file mode 100644 index 0000000000..fa1c5bbd01 --- /dev/null +++ b/src/rabbit_table.erl @@ -0,0 +1,311 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (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.mozilla.org/MPL/ +%% +%% Software distributed under the License is distributed on an "AS IS" +%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See +%% the License for the specific language governing rights and +%% limitations under the License. +%% +%% The Original Code is RabbitMQ. +%% +%% The Initial Developer of the Original Code is VMware, Inc. +%% Copyright (c) 2007-2012 VMware, Inc. All rights reserved. +%% + +-module(rabbit_table). + +-export([create/0, create_local_copy/1, wait_for_replicated/0, wait/1, + force_load/0, is_present/0, is_empty/0, + check_schema_integrity/0, clear_ram_only_tables/0]). + +-include("rabbit.hrl"). + +%%---------------------------------------------------------------------------- + +-ifdef(use_specs). + +-spec(create/0 :: () -> 'ok'). +-spec(create_local_copy/1 :: ('disc' | 'ram') -> 'ok'). +-spec(wait_for_replicated/0 :: () -> 'ok'). +-spec(wait/1 :: ([atom()]) -> 'ok'). +-spec(force_load/0 :: () -> 'ok'). +-spec(is_present/0 :: () -> boolean()). +-spec(is_empty/0 :: () -> boolean()). +-spec(check_schema_integrity/0 :: () -> rabbit_types:ok_or_error(any())). +-spec(clear_ram_only_tables/0 :: () -> 'ok'). + +-endif. + +%%---------------------------------------------------------------------------- +%% Main interface +%%---------------------------------------------------------------------------- + +create() -> + lists:foreach(fun ({Tab, TabDef}) -> + TabDef1 = proplists:delete(match, TabDef), + case mnesia:create_table(Tab, TabDef1) of + {atomic, ok} -> ok; + {aborted, Reason} -> + throw({error, {table_creation_failed, + Tab, TabDef1, Reason}}) + end + end, definitions()), + ok. + +%% The sequence in which we delete the schema and then the other +%% tables is important: if we delete the schema first when moving to +%% RAM mnesia will loudly complain since it doesn't make much sense to +%% do that. But when moving to disc, we need to move the schema first. +create_local_copy(disc) -> + create_local_copy(schema, disc_copies), + create_local_copies(disc); +create_local_copy(ram) -> + create_local_copies(ram), + create_local_copy(schema, ram_copies). + +wait_for_replicated() -> + wait([Tab || {Tab, TabDef} <- definitions(), + not lists:member({local_content, true}, TabDef)]). + +wait(TableNames) -> + case mnesia:wait_for_tables(TableNames, 30000) of + ok -> + ok; + {timeout, BadTabs} -> + throw({error, {timeout_waiting_for_tables, BadTabs}}); + {error, Reason} -> + throw({error, {failed_waiting_for_tables, Reason}}) + end. + +force_load() -> [mnesia:force_load_table(T) || T <- names()], ok. + +is_present() -> names() -- mnesia:system_info(tables) =:= []. + +is_empty() -> + lists:all(fun (Tab) -> mnesia:dirty_first(Tab) == '$end_of_table' end, + names()). + +check_schema_integrity() -> + Tables = mnesia:system_info(tables), + case check(fun (Tab, TabDef) -> + case lists:member(Tab, Tables) of + false -> {error, {table_missing, Tab}}; + true -> check_attributes(Tab, TabDef) + end + end) of + ok -> ok = wait(names()), + check(fun check_content/2); + Other -> Other + end. + +clear_ram_only_tables() -> + Node = node(), + lists:foreach( + fun (TabName) -> + case lists:member(Node, mnesia:table_info(TabName, ram_copies)) of + true -> {atomic, ok} = mnesia:clear_table(TabName); + false -> ok + end + end, names()), + ok. + +%%-------------------------------------------------------------------- +%% Internal helpers +%%-------------------------------------------------------------------- + +create_local_copies(Type) -> + lists:foreach( + fun ({Tab, TabDef}) -> + HasDiscCopies = has_copy_type(TabDef, disc_copies), + HasDiscOnlyCopies = has_copy_type(TabDef, disc_only_copies), + LocalTab = proplists:get_bool(local_content, TabDef), + StorageType = + if + Type =:= disc orelse LocalTab -> + if + HasDiscCopies -> disc_copies; + HasDiscOnlyCopies -> disc_only_copies; + true -> ram_copies + end; + Type =:= ram -> + ram_copies + end, + ok = create_local_copy(Tab, StorageType) + end, definitions(Type)), + ok. + +create_local_copy(Tab, Type) -> + StorageType = mnesia:table_info(Tab, storage_type), + {atomic, ok} = + if + StorageType == unknown -> + mnesia:add_table_copy(Tab, node(), Type); + StorageType /= Type -> + mnesia:change_table_copy_type(Tab, node(), Type); + true -> {atomic, ok} + end, + ok. + +has_copy_type(TabDef, DiscType) -> + lists:member(node(), proplists:get_value(DiscType, TabDef, [])). + +check_attributes(Tab, TabDef) -> + {_, ExpAttrs} = proplists:lookup(attributes, TabDef), + case mnesia:table_info(Tab, attributes) of + ExpAttrs -> ok; + Attrs -> {error, {table_attributes_mismatch, Tab, ExpAttrs, Attrs}} + end. + +check_content(Tab, TabDef) -> + {_, Match} = proplists:lookup(match, TabDef), + case mnesia:dirty_first(Tab) of + '$end_of_table' -> + ok; + Key -> + ObjList = mnesia:dirty_read(Tab, Key), + MatchComp = ets:match_spec_compile([{Match, [], ['$_']}]), + case ets:match_spec_run(ObjList, MatchComp) of + ObjList -> ok; + _ -> {error, {table_content_invalid, Tab, Match, ObjList}} + end + end. + +check(Fun) -> + case [Error || {Tab, TabDef} <- definitions(), + case Fun(Tab, TabDef) of + ok -> Error = none, false; + {error, Error} -> true + end] of + [] -> ok; + Errors -> {error, Errors} + end. + +%%-------------------------------------------------------------------- +%% Table definitions +%%-------------------------------------------------------------------- + +names() -> [Tab || {Tab, _} <- definitions()]. + +%% The tables aren't supposed to be on disk on a ram node +definitions(disc) -> + definitions(); +definitions(ram) -> + [{Tab, [{disc_copies, []}, {ram_copies, [node()]} | + proplists:delete( + ram_copies, proplists:delete(disc_copies, TabDef))]} || + {Tab, TabDef} <- definitions()]. + +definitions() -> + [{rabbit_user, + [{record_name, internal_user}, + {attributes, record_info(fields, internal_user)}, + {disc_copies, [node()]}, + {match, #internal_user{_='_'}}]}, + {rabbit_user_permission, + [{record_name, user_permission}, + {attributes, record_info(fields, user_permission)}, + {disc_copies, [node()]}, + {match, #user_permission{user_vhost = #user_vhost{_='_'}, + permission = #permission{_='_'}, + _='_'}}]}, + {rabbit_vhost, + [{record_name, vhost}, + {attributes, record_info(fields, vhost)}, + {disc_copies, [node()]}, + {match, #vhost{_='_'}}]}, + {rabbit_listener, + [{record_name, listener}, + {attributes, record_info(fields, listener)}, + {type, bag}, + {match, #listener{_='_'}}]}, + {rabbit_durable_route, + [{record_name, route}, + {attributes, record_info(fields, route)}, + {disc_copies, [node()]}, + {match, #route{binding = binding_match(), _='_'}}]}, + {rabbit_semi_durable_route, + [{record_name, route}, + {attributes, record_info(fields, route)}, + {type, ordered_set}, + {match, #route{binding = binding_match(), _='_'}}]}, + {rabbit_route, + [{record_name, route}, + {attributes, record_info(fields, route)}, + {type, ordered_set}, + {match, #route{binding = binding_match(), _='_'}}]}, + {rabbit_reverse_route, + [{record_name, reverse_route}, + {attributes, record_info(fields, reverse_route)}, + {type, ordered_set}, + {match, #reverse_route{reverse_binding = reverse_binding_match(), + _='_'}}]}, + {rabbit_topic_trie_node, + [{record_name, topic_trie_node}, + {attributes, record_info(fields, topic_trie_node)}, + {type, ordered_set}, + {match, #topic_trie_node{trie_node = trie_node_match(), _='_'}}]}, + {rabbit_topic_trie_edge, + [{record_name, topic_trie_edge}, + {attributes, record_info(fields, topic_trie_edge)}, + {type, ordered_set}, + {match, #topic_trie_edge{trie_edge = trie_edge_match(), _='_'}}]}, + {rabbit_topic_trie_binding, + [{record_name, topic_trie_binding}, + {attributes, record_info(fields, topic_trie_binding)}, + {type, ordered_set}, + {match, #topic_trie_binding{trie_binding = trie_binding_match(), + _='_'}}]}, + {rabbit_durable_exchange, + [{record_name, exchange}, + {attributes, record_info(fields, exchange)}, + {disc_copies, [node()]}, + {match, #exchange{name = exchange_name_match(), _='_'}}]}, + {rabbit_exchange, + [{record_name, exchange}, + {attributes, record_info(fields, exchange)}, + {match, #exchange{name = exchange_name_match(), _='_'}}]}, + {rabbit_exchange_serial, + [{record_name, exchange_serial}, + {attributes, record_info(fields, exchange_serial)}, + {match, #exchange_serial{name = exchange_name_match(), _='_'}}]}, + {rabbit_runtime_parameters, + [{record_name, runtime_parameters}, + {attributes, record_info(fields, runtime_parameters)}, + {disc_copies, [node()]}, + {match, #runtime_parameters{_='_'}}]}, + {rabbit_durable_queue, + [{record_name, amqqueue}, + {attributes, record_info(fields, amqqueue)}, + {disc_copies, [node()]}, + {match, #amqqueue{name = queue_name_match(), _='_'}}]}, + {rabbit_queue, + [{record_name, amqqueue}, + {attributes, record_info(fields, amqqueue)}, + {match, #amqqueue{name = queue_name_match(), _='_'}}]}] + ++ gm:table_definitions() + ++ mirrored_supervisor:table_definitions(). + +binding_match() -> + #binding{source = exchange_name_match(), + destination = binding_destination_match(), + _='_'}. +reverse_binding_match() -> + #reverse_binding{destination = binding_destination_match(), + source = exchange_name_match(), + _='_'}. +binding_destination_match() -> + resource_match('_'). +trie_node_match() -> + #trie_node{ exchange_name = exchange_name_match(), _='_'}. +trie_edge_match() -> + #trie_edge{ exchange_name = exchange_name_match(), _='_'}. +trie_binding_match() -> + #trie_binding{exchange_name = exchange_name_match(), _='_'}. +exchange_name_match() -> + resource_match(exchange). +queue_name_match() -> + resource_match(queue). +resource_match(Kind) -> + #resource{kind = Kind, _='_'}. diff --git a/src/rabbit_upgrade.erl b/src/rabbit_upgrade.erl index d037f954b4..455134da3f 100644 --- a/src/rabbit_upgrade.erl +++ b/src/rabbit_upgrade.erl @@ -201,7 +201,7 @@ primary_upgrade(Upgrades, Nodes) -> mnesia, Upgrades, fun () -> - force_tables(), + rabbit_table:force_load(), case Others of [] -> ok; _ -> info("mnesia upgrades: Breaking cluster~n", []), @@ -211,9 +211,6 @@ primary_upgrade(Upgrades, Nodes) -> end), ok. -force_tables() -> - [mnesia:force_load_table(T) || T <- rabbit_mnesia:table_names()]. - secondary_upgrade(AllNodes) -> %% must do this before we wipe out schema NodeType = node_type_legacy(), diff --git a/src/rabbit_upgrade_functions.erl b/src/rabbit_upgrade_functions.erl index 47b22b98f6..7c05499389 100644 --- a/src/rabbit_upgrade_functions.erl +++ b/src/rabbit_upgrade_functions.erl @@ -258,12 +258,12 @@ sync_slave_pids() -> %%-------------------------------------------------------------------- transform(TableName, Fun, FieldList) -> - rabbit_mnesia:wait_for_tables([TableName]), + rabbit_table:wait([TableName]), {atomic, ok} = mnesia:transform_table(TableName, Fun, FieldList), ok. transform(TableName, Fun, FieldList, NewRecordName) -> - rabbit_mnesia:wait_for_tables([TableName]), + rabbit_table:wait([TableName]), {atomic, ok} = mnesia:transform_table(TableName, Fun, FieldList, NewRecordName), ok. |