...
...
...
After partition is received the historical rebalance must be initiated to load other cache updates.
...
The swapped temporary storage will log all the cache updates to the temporary WAL storage (per each partition) for further applying them to the corresponding partition file.
The demander node will use a preloaded patition file as a new source of cache data entries to load.
Disadvantages:
In the process of balancing data:
The whole process is described in terms of rebalancing a single partition file of a cache group. All the other partitions would be rebalanced one-by-one.
In terms of a high-level abstraction, Apache Ignite must support the features described below.
The node partition preloader machinery download cache partition files from cluster nodes which owns desired partitions (the zero copy algorithm [1] assume to be used by default). To achieve this, the file transmission process must be implemented at Apache Ignite over Communication SPI.
IThe Comminication SPI must support to:
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
public interface CommunicationListenerEx<T extends Serializable> extends EventListener {
/**
* @param nodeId Remote node id.
* @param initMsg Init channel message.
* @param channel Locally created channel endpoint.
*/
public void onChannelOpened(UUID nodeId, Message initMsg, Channel channel);
} |
IO manager must support to:
While the Demander is being receive partition files it must save sequentially all cache entries corresponding to the MOVING partition into a new temporary storage. These entries will be applied later one by one on the newly received cache partition file. All asynchronous operations will be enrolled to the end of temporary storage during storage reads until it becomes fully read. The file-based FIFO approach assumes to be used by this process.
The temporary storage is chosen to be WAL-based. The storage must support to:
Expected problems to be solved
The demander node will use a preloaded patition file as a new source of cache data entries to load.
Disadvantages:
In the process of balancing data:
The whole process is described in terms of rebalancing a single partition file of a cache group. All the other partitions would be rebalanced one-by-one.
In terms of a high-level abstraction, Apache Ignite must support the features described below.
The node partition preloader machinery download cache partition files from cluster nodes which owns desired partitions (the zero copy algorithm [1] assume to be used by default). To achieve this, the file transmission process must be implemented at Apache Ignite over Communication SPI.
IThe Comminication SPI must support to:
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
public interface TransmissionHandlerCommunicationListenerEx<T extends Serializable> extends EventListener { /** * @param errnodeId TheRemote err of fail handling processnode id. */ @param initMsg Init public void onException(UUID nodeId, Throwable err); channel message. /** * @param nodeIdchannel RemoteLocally nodecreated idchannel fromendpoint. which request has been received.*/ public *void onChannelOpened(UUID nodeId, Message initMsg, Channel channel); } |
IO manager must support to:
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
public interface TransmissionHandler { /*@param fileMeta File meta info. * @return Absolute pathname denoting a file. */ public String filePath(UUID nodeId, TransmissionMeta fileMeta); /** * <em>Chunk handler</em> represents by itself the way of input data stream processing. * It accepts within each chunk a {@link ByteBuffer} with data from input for further processing. * * @param nodeIderr RemoteThe nodeerr idof fromfail whichhandling requestprocess. has been received. */ * @parampublic initMetavoid Initial handler meta info.onException(UUID nodeId, Throwable err); /** @return Instance of chunk handler* to@param processnodeId incomingRemote datanode by chunksid from which request has been received. * @param fileMeta File meta info. * @return Absolute pathname denoting a file. */ public Consumer<ByteBuffer>String chunkHandlerfilePath(UUID nodeId, TransmissionMeta initMetafileMeta); /** * <em>File<em>Chunk handler</em> represents by itself the way of input data stream processing. All the data will * It accepts within *each bechunk processeda under{@link theByteBuffer} hoodwith usingdata zero-copyfrom transferringinput algorithm and only start file processing andfor further processing. * the end of processing will* be@param provided. nodeId Remote node id * * @param nodeId Remote node id from from which request has been received. * @param initMeta Initial handler meta info. * @return IntanceInstance of readchunk handler to process incoming data like the {@link FileChannel} mannerby chunks. */ public Consumer<File>Consumer<ByteBuffer> fileHandlerchunkHandler(UUID nodeId, TransmissionMeta initMeta); } | ||||||||
Code Block | ||||||||
| ||||||||
public class TransmissionSender implements Closeable { /** * @param file Source file to send to remote. /** * <em>File handler</em> represents by itself the way of input data stream processing. All the data will * @parambe paramsprocessed Additionalunder transferthe filehood description keys. * @param plc The policy of handling data on remote. * @throws IgniteCheckedException If failsusing zero-copy transferring algorithm and only start file processing and * the end of processing will be provided. */ public void* send( @param nodeId Remote node id from which request has Filebeen file,received. * @param initMeta Initial Map<String,handler Serializable>meta params,info. * @return Intance TransmissionPolicyof plc read handler to process )incoming throwsdata IgniteCheckedException,like InterruptedException,the IOException { @link FileChannel} manner. */ public Consumer<File> fileHandler(UUID nodeId, TransmissionMeta initMeta); } |
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
public class TransmissionSender implements Closeable { send(file, 0, file.length(), params, plc); } /** * @param file Source file to send to remote. * @param plc params Additional transfer file description keys. * @param plc The policy of handling data on remote. * @throws IgniteCheckedException If fails. */ public void send( File file, Map<String, Serializable> params, TransmissionPolicy plc ) throws IgniteCheckedException, InterruptedException, IOException { send(file, 0, file.length(), new HashMap<>()params, plc); } /** * @param file Source file to send to remote. * @param offset Position to start trasfer atplc The policy of handling data on remote. * @param@throws cntIgniteCheckedException Number of bytes to transferIf fails. */ @param params Additional transferpublic file description keys.void send( * @param plc The policy of handling data on remote.File file, * @throws IgniteCheckedException IfTransmissionPolicy fails.plc */ public void send() throws IgniteCheckedException, InterruptedException, IOException { File send(file, 0, file.length(), new HashMap<>(), plc); long offset,} /** long cnt, * @param file Source file to send Map<String, Serializable> params, TransmissionPolicy plc to remote. * @param offset Position to start trasfer at. )* @param throwscnt IgniteCheckedException,Number InterruptedException,of IOException { // Implbytes to transfer. } } |
When the supplier node receives the cache partition file demand request it will send the file over the CommunicationSpi. The cache partition file can be concurrently updated by checkpoint thread during its transmission. To guarantee the file consistency Сheckpointer must use Copy-on-Write [3] tehnique and save a copy of updated chunk into the temporary file.
While the demander node is in the partition file transmission state it must save sequentially all cache entries corresponding to the MOVING partition into a new temporary storage. These entries will be applied later one by one on the newly received cache partition file. All asynchronous operations will be enrolled to the end of temporary storage during storage reads until it becomes fully read. The file-based FIFO approach assumes to be used by this process.
The temporary storage is chosen to be WAL-based. The storage must support to:
Expected problems to be solved
...
* @param params Additional transfer file description keys.
* @param plc The policy of handling data on remote.
* @throws IgniteCheckedException If fails.
*/
public void send(
File file,
long offset,
long cnt,
Map<String, Serializable> params,
TransmissionPolicy plc
) throws IgniteCheckedException, InterruptedException, IOException {
// Impl.
}
}
|
When the supplier node receives the cache partition file demand request it will send the file over the CommunicationSpi. The cache partition file can be concurrently updated by checkpoint thread during its transmission. To guarantee the file consistency Сheckpointer must use Copy-on-Write [3] tehnique and save a copy of updated chunk into the temporary file.
...
The node is ready to become partition owner when partition data is rebalanced and cache indexes are ready. For the message-based cluster rebalancing approach indexes are rebuilding simultaneously with cache data loading. For the file-based rebalancing approach, the index rebuild procedure must be finished before the partition state is set to the OWNING state.
be finished before the partition state is set to the OWNING state.
Ignite doesn't provide any recovery guarantees for the partitions with the MOVING state. The cache partitions will be fully loaded when the next rebalance procedure occurs.
The node which is beeing rebalancing left the cluster. For such nodes WAL is always disabled (all partitions have MOVING state due to this node is new for the cluster and has no cache data).
Since WAL is disabled we can guarantee that all operations with loaded partition files are safe to be done (renaming partition files, applying async updates) due to a cache directory will be fully dropped on recoveryApache Ignite doesn't provide any recovery guarantees for the partitions with the MOVING state. The cache partitions will be fully loaded when the next rebalance procedure occurs.
Each topology change event JOIN/LEFT/FAILED may or may not change cache affinity assignments of currently rebalacning caches. If assignments is not changed and the node is still needs partitions being rebalanced we can continue the current rebalance process (see for details IGNITE-7165).
...
To provide basic recovery guarantees we must to:
Recovery from different stages:
The SSL must be disabled to take an advantage of Java NIO zero-copy file transmission using FileChannel#transferTo method. If we need to use SSL the file must be splitted on chunks the same way to send them over the socket channel with ByteBuffer. As the SSL engine generally needs a direct ByteBuffer to do encryption we can't avoid copying buffer payload from the kernel level to the application level.
...