public interface TransmissionHandler {
/**
* @param err The err of fail handling process.
*/
public void onException(UUID nodeId, Throwable err);
/**
* @param nodeId Remote node id from which request has been received.
* @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 nodeId Remote node id from which request has been received.
* @param initMeta Initial handler meta info.
* @return Instance of chunk handler to process incoming data by chunks.
*/
public Consumer<ByteBuffer> chunkHandler(UUID nodeId, TransmissionMeta initMeta);
/**
* <em>File handler</em> represents by itself the way of input data stream processing. All the data will
* be processed under the hood using zero-copy transferring algorithm and only start file processing and
* the end of processing will be provided.
*
* @param nodeId Remote node id from which request has been received.
* @param initMeta Initial handler meta info.
* @return Intance of read handler to process incoming data like the {@link FileChannel} manner.
*/
public Consumer<File> fileHandler(UUID nodeId, TransmissionMeta initMeta);
}
ID | IEP-28 | ||||||
Author | |||||||
Sponsor | |||||||
Created | 31-Oct-2018 | ||||||
Status |
|
Table of Contents |
---|
The Apache Ignite cluster balance procedure with enabled persitence currently doesn't utilize network and storage device throughout to its full extent. The balance procedure processes cache data entries one by one which is not efficient enough for the cluster with enabled persistence.
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 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:
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:
...
...
...
31-Oct-2018
...
Status | ||||
---|---|---|---|---|
|
Table of Contents |
---|
The Apache Ignite cluster balance procedure with enabled persitence currently doesn't utilize network and storage device throughout to its full extent. The balance procedure processes cache data entries one by one which is not efficient enough for the cluster with enabled persistence.
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 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:
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:
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);
} |
The file transmission manager must support to:
...
language | java |
---|---|
theme | Confluence |
title | FileWriter.java |
collapse | true |
...
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
/** * */ public interface TransmitSessionHandler CommunicationListenerEx<T extends Serializable> extends EventListener { /** * @param nodeId The remoteRemote node id connected from. * @param sessionIdinitMsg TheInit uniquechannel session idmessage. */ @param channel Locally publiccreated void begin(UUID nodeId, String sessionId); /**channel endpoint. */ @return Thepublic instancevoid of read handler to process incoming data by chunks. */ public ChunkHandler chunkHandler(); onChannelOpened(UUID nodeId, Message initMsg, Channel channel); } |
The file transmission manager must support to:
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
public interface TransmissionHandler { /** * @param @returnerr The intanceerr of readfail handler tohandling process incoming data like the {@link FileChannel} manner. */ public FileHandler fileHandler( public void onException(UUID nodeId, Throwable err); /** * The@param endnodeId ofRemote sessionnode transmissionid process. from which request has been */received. public void end(); /*** @param fileMeta File meta info. * @param@return causeAbsolute Thepathname cause of fail handling processdenoting a file. */ public voidString onException(Throwable cause); } | ||||||||
Code Block | ||||||||
| ||||||||
filePath(UUID nodeId, TransmissionMeta fileMeta); /** * <em>Chunk */ public interface FileHandler { /**handler</em> represents by itself the way of input data stream processing. * @paramIt nameaccepts Thewithin fileeach namechunk transfer from. * @param position The start position pointer of download file in original source.a {@link ByteBuffer} with data from input for further processing. * * @param countnodeId TotalRemote countnode ofid bytesfrom readedwhich fromrequest thehas originalbeen sourcereceived. * @param paramsinitMeta TheInitial additionalhandler transfer file description paramsmeta info. * @return The absolute pathname string denoting the file or {@code null} if there is no sense. * @throws IOException If fails. Instance of chunk handler to process incoming data by chunks. */ public Consumer<ByteBuffer> chunkHandler(UUID nodeId, TransmissionMeta initMeta); /** */ <em>File handler</em> represents publicby Stringitself begin(String name, long position, long count, Map<String, Serializable> params) throws IOException; the way of input data stream processing. All the data will /** be processed under the * @param file The file with fully downloaded data into.hood using zero-copy transferring algorithm and only start file processing and * @param params The additional transfer file description params. the end of processing will be provided. * */ @param nodeId Remote publicnode id voidfrom end(File file, Map<String, Serializable> params); }which request has been received. * @param initMeta Initial handler meta info. * @return Intance of read handler to process incoming data like the {@link FileChannel} manner. */ public Consumer<File> fileHandler(UUID nodeId, TransmissionMeta initMeta); } |
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
public class TransmissionSender implements Closeable {
/**
* @param file Source file to send to remote.
* @param params Additional | ||||||||
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 paramskeys. * @param plc * @return The sizepolicy of ofhandling {@linkdata ByteBuffer} to read the input channel into. on remote. * @throws IOExceptionIgniteCheckedException If fails. */ public void send( File 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); }file, Map<String, Serializable> params, TransmissionPolicy plc ) throws IgniteCheckedException, InterruptedException, IOException { send(file, 0, file.length(), params, plc); } /** * @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 { send(file, 0, file.length(), new HashMap<>(), plc); } /** * @param file Source file to send to remote. * @param offset Position to start trasfer at. * @param cnt Number of bytes to transfer. * @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 { ... } } |
...