flo-l commented on a change in pull request #1525:
URL: https://github.com/apache/systemds/pull/1525#discussion_r793780489
##########
File path:
src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/HEParamServer.java
##########
@@ -0,0 +1,136 @@
+package org.apache.sysds.runtime.controlprogram.paramserv;
+
+import org.apache.sysds.parser.Statement;
+import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
+import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
+import
org.apache.sysds.runtime.controlprogram.paramserv.homomorphicEncryption.PublicKey;
+import
org.apache.sysds.runtime.controlprogram.paramserv.homomorphicEncryption.SEALServer;
+import org.apache.sysds.runtime.instructions.cp.CiphertextMatrix;
+import org.apache.sysds.runtime.instructions.cp.ListObject;
+import org.apache.sysds.runtime.instructions.cp.PlaintextMatrix;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/// This class implements Homomorphic Encryption (HE) for LocalParamServer. It
only supports modelAvg=true.
+public class HEParamServer extends LocalParamServer {
+ private int _thread_counter = 0;
+ private final List<FederatedPSControlThread> _threads;
+ private final List<Object> _result_buffer; // one per thread
+ private Object _result;
+ private final SEALServer _seal_server;
+
+ public static HEParamServer create(ListObject model, String aggFunc,
Statement.PSUpdateType updateType,
+ Statement.PSFrequency freq,
ExecutionContext ec, int workerNum, String valFunc, int numBatchesPerEpoch,
+ MatrixObject valFeatures,
MatrixObject valLabels, int nbatches)
+ {
+ return new HEParamServer(model, aggFunc, updateType, freq, ec,
+ workerNum, valFunc, numBatchesPerEpoch, valFeatures,
valLabels, nbatches);
+ }
+
+ private HEParamServer(ListObject model, String aggFunc,
Statement.PSUpdateType updateType,
+ Statement.PSFrequency freq, ExecutionContext ec,
int workerNum, String valFunc, int numBatchesPerEpoch,
+ MatrixObject valFeatures, MatrixObject valLabels,
int nbatches)
+ {
+ super(model, aggFunc, updateType, freq, ec, workerNum, valFunc,
numBatchesPerEpoch, valFeatures, valLabels, nbatches, true);
+
+ _seal_server = new SEALServer();
+
+ _threads = Collections.synchronizedList(new ArrayList<>(workerNum));
+ for (int i = 0; i < getNumWorkers(); i++) {
+ _threads.add(null);
+ }
+
+ _result_buffer = new ArrayList<>(workerNum);
+ resetResultBuffer();
+ }
+
+ public void registerThread(int thread_id, FederatedPSControlThread thread)
{
+ _threads.set(thread_id, thread);
+ }
+
+ private synchronized void resetResultBuffer() {
+ _result_buffer.clear();
+ for (int i = 0; i < getNumWorkers(); i++) {
+ _result_buffer.add(null);
+ }
+ }
+
+ public byte[] generateA() {
+ return _seal_server.generateA();
+ }
+
+ public PublicKey aggregatePartialPublicKeys(PublicKey[]
partial_public_keys) {
+ return _seal_server.aggregatePartialPublicKeys(partial_public_keys);
+ }
+
+ // this method collects all T Objects from each worker into a list and
then calls f once on this list to produce
+ // another T, which it returns.
+ private synchronized <T,U> U collectAndDo(int workerId, T obj,
Function<List<T>, U> f) {
+ _result_buffer.set(workerId, obj);
+ _thread_counter++;
+
+ if (_thread_counter == getNumWorkers()) {
+ List<T> buf = _result_buffer.stream().map(x ->
(T)x).collect(Collectors.toList());
+ _result = f.apply(buf);
+ resetResultBuffer();
+ _thread_counter = 0;
+ notifyAll();
+ } else {
+ try {
+ wait();
+ } catch (InterruptedException i) {
+ throw new RuntimeException("thread interrupted");
+ }
+ }
+
+ return (U) _result;
+ }
+
+ private CiphertextMatrix[] homomorphicAggregation(List<ListObject>
encrypted_models) {
+ CiphertextMatrix[] result = new
CiphertextMatrix[encrypted_models.get(0).getLength()];
+ for (int matrix_idx = 0; matrix_idx <
encrypted_models.get(0).getLength(); matrix_idx++) {
+ CiphertextMatrix[] summands = new
CiphertextMatrix[encrypted_models.size()];
+ for (int i = 0; i < encrypted_models.size(); i++) {
+ summands[i] = (CiphertextMatrix)
encrypted_models.get(i).getData(matrix_idx);
+ }
+ result[matrix_idx] = _seal_server.accumulateCiphertexts(summands);;
+ }
+ return result;
+ }
+
+ private Void homomorphicAverage(CiphertextMatrix[] encrypted_sums,
List<PlaintextMatrix[]> partial_decryptions) {
Review comment:
it is necessary, because this function is used in a generic context (in
collectAndDo).
##########
File path:
src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/FederatedPSControlThread.java
##########
@@ -324,6 +401,7 @@ protected ListObject pullModel() {
}
protected void weightAndPushGradients(ListObject gradients) {
+ assert (!(_weighting && _use_homomorphic_encryption)) :
"weights and homomorphic encryption are not supported together";
Review comment:
good catch! I moved this check into
ParamservBuiltinCPInstruction.java#runFederated. There is a similar check
already there which makes sure we're using modelAvg. I'll leave the assert as a
sanity check. fixed in 3f9bbdb64e771343576c6ab50588e605a49acf5e.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
