...
The Apache Ignite can support cache rebalancing as transferring partition files using zero copy algorithm [1] based on an extension of communication SPI and Java NIO API. When the partition file has been transferred to the demander node there are two a few possible approaches can be implemented to preload entries from particular partition file.
The demander node will initialize a preloaded partition file as a new PageStore, make this storage up-to-date by applying previously saved async cache operations and then hot swap it under the checkpoint write lock.
Disadvantages:
Demander node first under checkpoint write lock must swap cache data storage with the temporary one to perform recovery operations under original cache data storage. After partition file has been received from the Supplier node there are to possible cases to make this partition file up-to-date.
Disadvantages:
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 rebalance single cache group and partition files would be rebalanced one-by-one:
In terms of a high-level abstraction, Apache Ignite must support the features described below.
Partition preloader must support cache partition file relocation from one cluster node to another (the zero copy algorithm [1] assume to be used by default). To achieve this, the file transfer machinery must be implemented at Apache Ignite over Communication SPI.
Ignite's Comminication SPI must support to:
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. 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 CommunicationListenerEx<T extends Serializable> extends EventListener { | ||||||||
Code Block | ||||||||
| ||||||||
/** * */ public interface Channel extends Closeable { /** * @return Connection id to remote node. */ public ChannelId id(); /** * @return The channel's configuration. */ public ChannelConfig config(); /** * @return@param AnodeId mapRemote of all channel attributesnode id. */ @param initMsg Init public Map<String, Serializable> attrs(); /**channel message. * @param namechannel TheLocally namecreated tochannel get attributeendpoint. */ @param <T> The attributepublic type. * @return The corresponding channel attribute instance. */ public <T extends Serializable> T attr(String name); /** * @param name The name to get attribute. * @param obj The chanel attribute instance. * @param <T> The attribute typevoid onChannelOpened(UUID nodeId, Message initMsg, Channel channel); } |
IO manager must support to:
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
public interface TransmissionHandler { /** * @param err The err of fail handling process. */ public <T extends Serializable> void attronException(StringUUID namenodeId, TThrowable objerr); /** * @return@param <tt>true</tt>nodeId ifRemote thenode channelid isfrom configuredwhich andrequest readyhas tobeen usereceived. */ @param fileMeta File public boolean active(); /**meta info. * Make@return theAbsolute channelpathname readydenoting fora usefile. */ public voidString activate(); } | ||||||||
Code Block | ||||||||
| ||||||||
/** * */ public interface ChannelConfig { /** * Gets the channel's modefilePath(UUID nodeId, TransmissionMeta fileMeta); /** * <em>Chunk handler</em> represents by itself the way of input data stream processing. */ It accepts within publiceach boolean blocking(); chunk a {@link ByteBuffer} with data from input for further processing. /** * @param blocking The channel's mode to set nodeId Remote node id from which request has been received. * @param initMeta @returnInitial Thehandler configmeta instanceinfo. */ @return Instance of publicchunk ChannelConfighandler blocking(boolean blocking); /** * Gets the timeout option optionto process incoming data by chunks. */ public intConsumer<ByteBuffer> timeout(chunkHandler(UUID nodeId, TransmissionMeta initMeta); /** * @param<em>File millishandler</em> Therepresents timeout in milliseconds.by itself the way of input data stream processing. All the data will * @return The config instance. */ be processed under the hood using zero-copy transferring algorithm and only start file processing and public ChannelConfig* timeout(int millis); } | ||||||||
Code Block | ||||||||
| ||||||||
public interface CommunicationSpi<T extends Serializable> extends IgniteSpi { the end of processing will be provided. /** * @param remote Destination cluster node to communicate with nodeId Remote node id from which request has been received. * @param attrsinitMeta Initial Configurationhandler channelmeta attributesinfo. * @throws IgniteSpiException If fails @return Intance of read handler to process incoming data like the {@link FileChannel} manner. */ public defaultConsumer<File> Channel channelfileHandler(ClusterNodeUUID remotenodeId, Map<String, Serializable> attrs) throws IgniteSpiException { throw new UnsupportedOperationException(); } TransmissionMeta initMeta); } |
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
public interfaceclass CommunicationListener<TTransmissionSender extends Serializable> extends EventListenerimplements Closeable { /** * @param chfile ASource channel instancefile to processsend configureto requestremote. * @param @returnparams Additional attributestransfer tofile send to remote nodedescription keys. */ @param plc The publicpolicy defaultof Map<String,handling Serializable>data onChannelConfigure(Channel ch) {on remote. * @throws IgniteCheckedException returnIf null;fails. } */** public *void @paramsend( nodeId Remote node id. * @param ch Locally created channel endpoint. File file, Map<String, Serializable> params, */ TransmissionPolicy plc public default void) onChannelCreated(UUIDthrows nodeIdIgniteCheckedException, ChannelInterruptedException, ch)IOException { // No-op. send(file, 0, file.length(), params, plc); } } |
The file transmission manager must support to:
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
/** * */ public interface FileWriter extends Closeable { /** * @param file The source file to send at. * @param offset The position to start at. /** * @param file Source file to send to remote. * @param plc The policy of handling data on remote. * @throws IgniteCheckedException If fails. */ public void send( File file, TransmissionPolicy plc ) throws IgniteCheckedException, InterruptedException, IOException { * @param count The number of bytes to transfer.send(file, 0, file.length(), new HashMap<>(), plc); } * @param params The additional transfer file description keys. /** * @param plcfile TheSource policyfile ofto handlingsend data onto remote. * @throws IgniteCheckedException If fails @param offset Position to start trasfer at. */ @param cnt Number publicof void write( bytes to transfer. * @param params Additional Filetransfer file, description keys. * @param plc longThe offset, policy of handling data on remote. long count, * @throws IgniteCheckedException If fails. Map<String, Serializable> params,*/ public void send( ReadPolicy plc ) throws IgniteCheckedException; } | ||||||||
Code Block | ||||||||
| ||||||||
/** * */ public interface TransmitSession { /**File file, long offset, * @param nodeId Thelong remotecnt, node id connected from. Map<String, *Serializable> @paramparams, sessionId The unique session id. TransmissionPolicy */plc public) void begin(UUID nodeIdthrows IgniteCheckedException, InterruptedException, StringIOException sessionId); /** * @return The instance of read handler to process incoming data by chunks. */ public ChunkHandler chunkHandler(); /** * @return The intance of read handler to process incoming data like the {@link FileChannel} manner. */ public FileHandler fileHandler(); /** * The end of session transmission process. */ public void end(); /** * @param cause The cause of fail handling process. */ public void onException(Throwable cause); } |
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
/**
*
*/
public interface FileHandler {
/**
* @param name The file name transfer from.
* @param position The start position pointer of download file in original source.
* @param count Total count of bytes readed from the original source.
* @param params The additional transfer file description params.
* @return The absolute pathname string denoting the file or {@code null} if there is no sense.
* @throws IOException If fails.
*/
public String begin(String name, long position, long count, Map<String, Serializable> params) throws IOException;
/**
* @param file The file with fully downloaded data into.
* @param params The additional transfer file description params.
*/
public void end(File file, Map<String, Serializable> params);
} |
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
/**
*
*/
public interface ChunkHandler {
/**
* @param name The file name on remote.
* @param position The start position pointer of downloading file in original source.
* @param count Total count of bytes to read from the original source.
* @param params The additional transfer file description params.
* @return The size of of {@link ByteBuffer} to read the input channel into.
* @throws IOException If fails.
*/
public int begin(String name, long position, long count, Map<String, Serializable> params) throws IOException;
/**
* @param buff The data filled buffer.
* @return {@code true} if the chunk of data have been successfully accepted.
* @throws IOException If fails.
*/
public boolean chunk(ByteBuffer buff) throws IOException;
/**
* @param params The additional handling channel description params.
*/
public void end(Map<String, Serializable> params);
} |
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
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.
{
// 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.
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.
...