...
...
(Partition Map) Exchange
PME - is a process of exchange partition information between nodes. Process A process goal is to setup set up an actual state of partitions for all cluster nodes.
...
Events which causes exchange
Topology events:
Custom events:
Partition Map Exchage Exchange process is following the following:
...
Put to this queue (GridCachePartitionExchangeManager.ExchangeWorker#futQ) is done from the discovery thread.
'Exhange Exchange worker' thread manages this queue. Using only one exhange exchange worker thread per node provides a strict processing order for futures.
Step 1) First of all exhange worker thread runs init() on this exchange future.
There are important actions happened in this phase:
Exchange type determination
Basically there are 3 types of exchange - ALL, CLIENT and NONE.
We briefly describe CLIENT and NONE types here as they are much simpler and shorter than ALL type.
CLIENT exchange happened on client nodes. It doesn’t have any distributed phase. The client node just sends SingleMessage to the coordinator and receives FullMessage when the coordinator finished exchange.
NONE exchange happened on server and client nodes when Node join / left event is triggered by the client node. NONE exchange just updates the topology version. Client node which joins to cluster has CLIENT exchange type.
All other things following are applicable for ALL exchange type.
Cache actions
Dynamic cache start / Activation case:
All caches needed to start (located at org.apache.ignite.internal.processors.cache.ExchangeActions#cacheStartRequests) start here (org.apache.ignite.internal.processors.cache.CacheAffinitySharedManager#onCacheChangeRequest)
Dynamic cache stop case:
All caches needed to stop (located at org.apache.ignite.internal.processors.cache.ExchangeActions#cacheStopRequests) close public API to prevent new updates here (org.apache.ignite.internal.processors.cache.CacheAffinitySharedManager#processCacheStopRequests)
Node join case:
If joining node has new configured caches (that did not exist before this exchange) such caches additionally started on all other nodes (org.apache.ignite.internal.processors.cache.GridCacheProcessor#startReceivedCaches)
On joining node all cluster and node caches start (org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture#initCachesOnLocalJoin)
Other cases:
No additional actions with caches required on this phase.
Affinity actions
For all started caches coordinator node calculates affinity for ‘initialVersion’, non-coordinator nodes also try to calculate affinity for ‘initialVersion’ (@see org.apache.ignite.internal.processors.cache.CacheAffinitySharedManager#canCalculateAffinity), if it’s not possible nodes fetch affinity from coordinator.
After all necessary caches started and initial affinity initialized we set current exchange future as topology future (@see org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopologyImpl#topReadyFut) to our Topology (@see org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture#updateTopologies).
The topology future is used for mapping atomic updates, transactions, and explicit locks. Each atomic operation or transaction should be associated with some topology version on what it should be performed or remapped to the next topology version. Each operation obtains topology future using topology read lock (e.g. @see org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockFuture#mapOnTopology). If the future is already completed operation proceed using the topology version from this future. If the future is not completed operation asynchronously waits for its completion and remaps on this future version.
Before the distributed exchange start, we need to wait for finishing all ongoing updates that were mapped on the previous topology version. After the topology future is set we have operations that had time to obtain previous topology version (we called them as “ongoing”), other operations asynchronously wait for exchange future finish so we do nothing with them. They will proceed after the exchange future is finished.
Finishing ongoing updates have happened on the Partitions Release phase.
Partitions Release / Partitions Release Latch
Each update operation (atomic update, transaction) associated with some top-level future indicates the operation completion. We collect all of these futures to one CompoundFuture (or Multi-future) called “Partitions release future” and wait for its completion on all server nodes and then proceed to the next phase of exchange.
However, such simple waiting for completion maybe not enough.
There are 3 cases:
For cases 1 and 2 a 2-phase waiting of transactions is introduced. In the first phase, we wait for finishing all transaction futures that we see locally on each of the server nodes. After waiting is finished all nodes waits for distributed countdown latch completion. After latch is released we have invariant that only transaction futures that represent commit from primary to backup can appear. So in the second phase, we do the same thing as in the first phase but without latch. [IGNITE-7871]
For case 3 a transaction timeout on PME is introduced. If the transaction can’t be completed within the specified timeout transaction will be forcibly rolled back during partitions release. [IGNITE-6827]
WAL History reservation
WAL History reservation is needed for historical rebalance. On this step each node lookups onto their current CheckpointHistory that is held in-memory and finds the earliest available checkpoint that is presented in WAL as CheckpointRecord for all cache groups. CheckpointRecord contains information about all groups and partitions - which update counter for each partition has been seen during this checkpoint creation. Earliest CheckpointRecord is reserved after successful found. Reservation means that WAL segment which contains this CheckpointRecord and more fresh segments will not be truncated (deleted) before segment release. After earliest CheckpointRecord is found and reserved, for all cache groups and partitions earliest available update is calculated. These update counters are used as starting points for historical rebalance. If joining node has the latest update counter for some partition greater than the earliest reserved counter on such node - it means that joining node can use this node as a supplier for historical rebalance and replay all updates since the latest update counter till WAL end.
Database callback
After partitions release phase is finished and we have no ongoing updates we can start with partitions manipulations. On this step, we need to restore partition states from page memory. Despite restoring partition states happen before node joins to cluster [IGNITE-9420] and it’s idempotent operation there is a case when this step is still needed. This is needed when Snapshot restore happens. During snapshot restore, we stop some caches and start, it again. After the cache start, we need to re-create partitions and read their state from disk (IgniteCacheDatabaseSharedManager#beforeExchange).
After existing partitions are attached to topology we need to create remaining partitions according to calculated affinity assignment on init phase and initialize partitions full map (GridDhtPartitionTopology#beforeExchange).
Single message sending
If a node is not a coordinator, as a result, this method sends a single map ( GridDhtPartitionsSingleMessage ). Signle A single map is sent using communication SPI (in peer2peer manner).
This signle single message contains partitions map for each cache ( GridDhtPartitionMap ). It contains owners of each paritition partition for each cache group (cache group id is mapped to Partitions Map). Following is an example from the log for 2 caches:
msg=GridDhtPartitionsSingleMessage [parts= {1544803905=GridDhtPartitionMap [moving=0, top=AffinityTopologyVersion [topVer=9, minorTopVer=3], updateSeq=2, size=406], -2100569601=GridDhtPartitionMap [moving=0, top=AffinityTopologyVersion [topVer=9, minorTopVer=3], updateSeq=161, size=100]}
Step 2) Coordinator maintains The coordinator maintains a list of nodes to wait for a response from (N1, N2, N3 for example at pic.1).
Single messages collecting
Each node sends the state of it’s local partitions to coordinator in message called “GridDhtPartitionsSingleMessage”. This message contains several maps representing the state of partitions of all registered caches. The state contains partition state, update counter and size of partition. This information is needed for validation, affinity calculation, assigning primary-backup and rebalance. Each exchange coordinator knows a set of server nodes that should send these single messages. When the coordinator receives all of them from all server nodes it starts to analyze the cluster state of partitions and perform affinity and partitions manipulations.
Exchanges merge
Exchanges merge is optimization when multiple topology events (node join / left) are happening at the same time. Instead of sequential execution of such exchanges they can be merged into one [IGNITE-6124]. How it works:
When all single messages of the current exchanges are received, the coordinator sequentially scans the exchange worker queue and finds all topology events that can be merged together. If the topology event requires a single message (node join) coordinator will wait for it. If next topology event in the queue is not supposed to be merged (e.g. it’s custom discovery event, or node joins with new caches) queue scan is stopped. All additional single messages from other joining nodes are placed here (GridDhtPartitionsExchangeFuture#mergedJoinExchMsgs), if other joining node hasn’t sent a single message yet, await counter is incremented (GridDhtPartitionsExchangeFuture#awaitMergedMsgs) when a single message is received this counter is decremented. If this counter becomes zero coordinator finishes exchange as in regular way but with another result topology version equal to the freshest version that it sees in merged exchange futures. Discovery events from such merged exchanges are placed to the current exchange future and they are used for further calculations. Merged exchanges are removed from the exchange queue on all nodes when the current exchange is finished.
Affinity calculation
On this step, we need to calculate ideal and “real” affinity partitions distribution on coordinator considering the state of partitions received from all other nodes. Ideal affinity partitions distribution is controlled by affinity function (Rendezvous by default). This function assigns primary and backup (configurable backup factor) nodes for each of partitions in all cache groups. Primary node always reflects the most actual state of data in each of the partitions. Primary node usually is a leader for any operations on partition and it replicates state to backup nodes.
Partition on a primary node should always have OWNING state. In the case of full and stable topology, ideal distribution is equal to real. In case of unstable topology, join-leave nodes, baseline change ideal affinity distribution not always following invariant that primary node has partition in OWNING state, because ideal affinity distribution doesn’t take into account the state of all partitions, it just shows desired state affinity.
In this case, we need to place some backups that have OWNING state onto primary place before rebalance is finished. After rebalance such temporary distribution is changed to ideal and all primary nodes are now on the place, while extra backups become rent. This process is called as LateAffinityAssignment.
There are 2 main strategies to calculate “real” affinity:
This strategy is used for almost all types of exchange - activation, baseline change and node left case.
Validate partition states
On this step, we compare partition update counters and partition sizes between nodes that reported partition OWNING status. If there is a difference it may signal about data inconsistency problems in a cluster. The appropriate warning will be printed.
Assign partition states
On this step, we need to find outdated OWNING partitions (having update counter less than others) move them into MOVING state and prepare to rebalance after exchange.
If partition can’t be rebalanced using WAL (@see WAL History reservation) asynchronous clearing will be started after changing such partition to MOVING state.
Usually OWNING partitions that have outdated update counter are come from joining nodes that were offline some time.
This step is evaluated during cluster activation, baseline change or node join.
Detect lost partitions
On this step, we need to find partitions that don’t have at least 1 alive owner (at least 1 partition on some node in OWNING state). Such partitions are marked as LOST, reading-writing from such partitions are controlled by partition loss policy. If partition loss policy is IGNORE such partitions will be immediately moved to OWNING state without any restrictions regarding read-write operations on them.
This step is evaluated only when the node failed or left from cluster instead of previous (@see Assign partition states) step execution.
Full message preparing and sending
According to single maps received and to its own state coordinator build the full map.
When the state of partitions and affinity are determined we need to notify about this all other server nodes that participated in the exchange. Participated nodes set are formed
To send all necessary information messages called “GridDhtPartitionsFullMessage” are used. It contains the following important information:
When all expected responses were received coordinator sends full map to other nodes.
...
Partition state in full and signle single maps are encoded using bit sets for compression.
On receiving full map for current exchange, nodes calls onDone() for this future. Listeners for full map are called (order is no not guaranteed).
...
Info |
---|
SingleMessage with exchId=null is sent when a node updates local partitions' state and schedules a background cluster notification. In contrast, when a partition map exchange happendshappens, it is completed with exchId != null. |
...
Using ReadWrite lock gives semantic N transactions, or 1 exchange.
Late Affinity Assignment is optimization which assumes delayed switch of primary partitions after topology change.
This is feature is enabled by default since Optimisation, Centralized affinity, flag - true by default 2.0+ , and is the only possible option since 2.1+.
Affinity assigns primary partition (to be migrated) to new node. But instead we create temporary backup at this new node.
When temporary backup loads all actual data it becomes primary.
On each topology change, for each started cache partition-to-node mapping is calculated using {@link AffinityFunction} configured for cache. When late affinity assignment mode is disabled then new affinity mapping is applied immediately.
In late affinity assignment mode if primary node was changed for some partition then current primary is not changed and new primary is temporarily assigned as backup. Later, when it's ensured that all new "ideal affinity" primaries are ready to become true primaries, cluster performs late affinity switch procedure: on separate PME (triggered by CacheAffinityChangeMessage) primary assignments are recalculated to match with {@link AffinityFunction}. There are three cases which may require late affinity assignment:
Moment of the late affinity switch is controlled by the coordinator node. It stores information about all ongoing rebalancing processes in CacheAffinitySharedManager#waitInfo. When it's empty, coordinator node sends affinity change message (see {@link CacheAffinitySharedManager#checkRebalanceState}), which triggers late affinity switch PME. In case all partitions on joined node are up-to-date, such PME is triggered right away after topology change PMEAdditional synthetic Exchange will be issued.