Loading repository data…
Loading repository data…
kanyun-inc / repository
Ytk-mp4j is a fast, user-friendly, cross-platform, multi-process, multi-thread collective message passing java library which includes gather, scatter, allgather, reduce-scatter, broadcast, reduce, allreduce communications for distributed machine learning.
A transparent discovery signal based on current public GitHub metadata.
This score does not audit code, security, maintainers, documentation quality, or suitability. Verify the repository and its current documentation before adoption.
Ytk-mp4j is a fast, user-friendly, cross-platform, multi-process, multi-thread collective message passing java library for distributed machine learning. It's similar to MPI but it has several important features:
MPI with OpenMP to use multi-process, multi-thread. Serial codes can easily be paralleled and distributed.double, float, long, int, short, byte, String, Object).Serializer interface of Kryo.array data container, but also map container(used for sparse communication).Ytk-mp4j uses master-slave mode, only one master, one or more slaves(workers). Responsibilities of master are:
Communicative slaves are embedded in real workers, which can lodge with any kinds of workers(e.g. reduce of hadoop, executors of spark, …).
Ytk-mp4j implements state-of-art collective algorithms descrited in [1,2].
You can use shell command below to start master directly.
nohup java -server -Xmx600m -classpath .:lib/*:config -Dlog4j.configuration=file:config/log4j_master.properties com.fenbi.mp4j.comm.CommMaster ${slave_num} ${master_port} > log/master_startup.log 2>&1 & echo $! > master_${master_port}.pid
You can also assemble codes below into your project.
CommMaster master = null;
try {
int nWorker = Integer.parseInt(args[0]);
int rpcPort = Integer.parseInt(args[1]);
LOG.info("worker:" + nWorker);
LOG.info("port:" + rpcPort);
master = new CommMaster(nWorker, rpcPort);
master.start();
LOG.info("server hostname:" + master.getHostName());
LOG.info("server hostport:" + master.getHostPort());
} catch (IOException e) {
LOG.error("IO exception", e);
} finally {
int code;
if (master == null) {
code = 1;
LOG.error("master start error, failed!");
} else {
code = master.stop();
}
LOG.info("exit code:" + code);
System.exit(code);
}
int errorCode = 0;
ProcessCommSlave comm = null;
try {
comm = new ProcessCommSlave(loginName, hostName, hostPort);
// ...
// array = comm.allreduceArray(array, ......)
// ...
} catch (Exception e) {
errorCode = 1;
LOG.error("existed exception!", e);
if (comm != null) {
try {
comm.exception(e);
} catch (Mp4jException e1) {
LOG.error("comm send exception message error!", e);
}
}
} finally {
try {
comm.close(errorCode);
} catch (Mp4jException e) {
errorCode = 1;
LOG.error("comm close exception!", e);
}
}
return errorCode == 0;
int errorCode = 0;
ThreadCommSlave comm = null;
try {
comm = new ThreadCommSlave(loginName, threadNum, hostName, hostPort);
final ThreadCommSlave finalComm = comm;
Thread []threads = new Thread[threadNum];
for (int t = 0; t < threadNum; t++) {
final int tidx = t;
threads[t] = new Thread(t + "") {
@Override
public void run() {
finalComm.setThreadId(tidx);
try {
// ...
// array = comm.allreduceArray(array, ......)
// ...
} catch (Exception e) {
try {
finalComm.exception(e);
finalComm.close(1);
} catch (Mp4jException e1) {
LOG.error("comm send exception message error!", e);
}
System.exit(1);
}
}
};
threads[t].start();
}
for (int t = 0; t < threadNum; t++) {
threads[t].join();
}
} catch (Exception e) {
errorCode = 1;
LOG.error("existed exception!", e);
if (comm != null) {
try {
comm.exception(e);
} catch (Mp4jException e1) {
LOG.error("comm send exception message error!", e);
}
}
} finally {
try {
comm.close(errorCode);
} catch (Mp4jException e) {
errorCode = 1;
LOG.error("comm close exception!", e);
}
}
return errorCode == 0;
Collective communication is a method of communication which involves participation of all processes(all threads). Ytk-mp4j doesn't contain point-to-point communication(using collective comunication can realize it, but it is not recommended). A communication operation in ytk-mp4j usually contains an Operand. In reduction operation, it also contains a Operator.
Each process will be assigned a special number—rank which is encoded with lexicographical order of hostname from 0 to slaveNum - 1. Use getRank() to get rank. Use getSlaveNum() to get the total number of processes.
Ytk-mp4j provides 8 predefined Operands, which almost meets all your needs. If you want to define a new Operand, you can extend Operand abstract class. compress is to decide whether the communication uses compression(default is false). To create a ObjectOperand, you must provide a serializer of Kryo and class type for your object. Kryo is a fast and efficient object graph serialization framework for java. ytk-learn uses Kryo to serialize and compress objects. More details about predefined Operands see Operands.
Using KryoUtils.getDefaultSerializer(Class type) function, you can get serializer easily for most simple objects.
| Operand | Support Java Types | How to Create? |
|---|---|---|
| DoubleOperand | single double, double array, Double object | Operands.DOUBLE_OPERAND(boolean compress) |
| FloatOperand | single float, double array, Float object | Operands.FLOAT_OPERAND(boolean compress) |
| LongOperand | single long, long array, Long object | Operands.LONG_OPERAND(boolean compress) |
| IntOperand | single int, int array, Integer object | Operands.INT_OPERAND(boolean compress) |
| ShortOperand | single short, short array, Short object | Operands.SHORT_OPERAND(boolean compress) |
| ByteOperand | single byte, type array, Byte object | Operands.BYTE_OPERAND(boolean compress) |
| StringOperand | String object | Operands.STRING_OPERAND(boolean compress) |
| ObjectOperand | Object | Operands.OBJECT_OPERAND(Serializer serializer, Class type, boolean compress) |
Different Operands support different Operators. Ytk-mp4j provides some predefined reduction operations. The reduction operations should be commutative and associative. You can implement interface of different operands to get custom reduction operation.More details about predefined Operands see Operators.
| Operand | Predefined Reduction IOperators | Interface |
|---|---|---|
| DoubleOperand | Operators.Double.SUM/MAX/MIN/PROD/FLOAT_MAX_LOC/FLOAT_MIN_LOC | IDoubleOperator |
| FloatOperand | Operators.Float.SUM/MAX/MIN/PROD | IFloatOperator |
| LongOperand | Operators.Long.SUM/MAX/MIN/PROD/BITS_AND/BITS_OR/BITS_XOR/INT_MAX_LOC/INT_MIN_LOC | ILongOperator |
| IntOperand | Operators.Int.SUM/MAX/MIN/PROD/BITS_AND/BITS_OR/BITS_XOR | IIntOperator |
| ShortOperand | Operators.Short.SUM/MAX/MIN/PROD/BITS_AND/BITS_OR/BITS_XOR | IShortOperator |
| ByteOperand | Operators.Byte.SUM/MAX/MIN/PROD/BITS_AND/BITS_OR/BITS_XOR | IByteOperator |
| StringOperand | null | IStringOperator |
| ObjectOperand | null | IObjectOperator |
Set union, intersection and List concat(partial) are commutative and associative. Those operations are often used. More details see ThreadCommSlave, ProcessCommSlave.
gather: Before the gather, each process node owns a piece of the data. After the gather, root process owns the entire data.
allgather: Before the allgather, each process node owns a piece of the data. After the allgather, all processes own all of the data.
broadcast: Before the broadcast, only root process owns entire data. After the broadcast, all processes own all of the data.
scatter: Before the scatter, only root process owns entire data. After scatter, each process owns a piece of the data.
reduce: Before the reduce, each process owns the data x(i). After the reduce, only root process owns the data of (x(0) + x(1) + … + x(p-1)). "+" is a general reduction operation.
allreduce: Identical to the reduce, except all the processes own the data of reduced.