From 6e886e53920a148e5d16f90aa0dd87a477b16d39 Mon Sep 17 00:00:00 2001 From: Duncan Tebbs Date: Mon, 17 May 2021 11:31:26 +0100 Subject: [PATCH 1/4] fix for group element serialization --- .../barreto_lynn_scott/BLSBinaryReader.java | 16 +++++++++++++--- .../curves/barreto_naehrig/BNBinaryReader.java | 18 ++++++++++++++---- .../curves/GenericBinaryWriterTest.java | 6 ++++++ 3 files changed, 33 insertions(+), 7 deletions(-) diff --git a/src/main/java/algebra/curves/barreto_lynn_scott/BLSBinaryReader.java b/src/main/java/algebra/curves/barreto_lynn_scott/BLSBinaryReader.java index 15d22af..5866b46 100644 --- a/src/main/java/algebra/curves/barreto_lynn_scott/BLSBinaryReader.java +++ b/src/main/java/algebra/curves/barreto_lynn_scott/BLSBinaryReader.java @@ -78,14 +78,24 @@ public BLSFrT readFr() throws IOException { @Override public BLSG1T readG1() throws IOException { + final var x = readFq(); + final var y = readFq(); + if (x.isZero() && y.isOne()) { + return G1One.zero(); + } + return G1One.construct(readFq(), readFq(), FqOne); } @Override public BLSG2T readG2() throws IOException { - final BLSFq2T X = readFq2(); - final BLSFq2T Y = readFq2(); - return G2One.construct(X, Y, Y.one()); + final BLSFq2T x = readFq2(); + final BLSFq2T y = readFq2(); + if (x.isZero() && y.isOne()) { + return G2One.zero(); + } + + return G2One.construct(x, y, y.one()); } protected BLSFqT readFq() throws IOException { diff --git a/src/main/java/algebra/curves/barreto_naehrig/BNBinaryReader.java b/src/main/java/algebra/curves/barreto_naehrig/BNBinaryReader.java index 9793202..0a60fa6 100644 --- a/src/main/java/algebra/curves/barreto_naehrig/BNBinaryReader.java +++ b/src/main/java/algebra/curves/barreto_naehrig/BNBinaryReader.java @@ -76,14 +76,24 @@ public BNFrT readFr() throws IOException { @Override public BNG1T readG1() throws IOException { - return G1One.construct(readFq(), readFq(), FqOne); + final var x = readFq(); + final var y = readFq(); + if (x.isZero() && y.isOne()) { + return G1One.zero(); + } + + return G1One.construct(x, y, FqOne); } @Override public BNG2T readG2() throws IOException { - final BNFq2T X = readFq2(); - final BNFq2T Y = readFq2(); - return G2One.construct(X, Y, Y.one()); + final var x = readFq2(); + final var y = readFq2(); + if (x.isZero() && y.isOne()) { + return G2One.zero(); + } + + return G2One.construct(x, y, y.one()); } protected BNFqT readFq() throws IOException { diff --git a/src/test/java/algebra/curves/GenericBinaryWriterTest.java b/src/test/java/algebra/curves/GenericBinaryWriterTest.java index a3854c2..edd6422 100644 --- a/src/test/java/algebra/curves/GenericBinaryWriterTest.java +++ b/src/test/java/algebra/curves/GenericBinaryWriterTest.java @@ -30,14 +30,17 @@ public void testBinaryWriter( { var writer = mkWriter.apply(os); + writer.writeFr(frOne.zero()); writer.writeFr(frOne); writer.writeFr(frOne.construct(-1)); writer.writeFr(frOne.construct(2)); writer.writeFr(frOne.construct(-2)); + writer.writeG1(g1One.zero()); writer.writeG1(g1One); writer.writeG1(g1One.mul(frOne.construct(-1))); writer.writeG1(g1One.mul(frOne.construct(2))); writer.writeG1(g1One.mul(frOne.construct(-2))); + writer.writeG2(g2One.zero()); writer.writeG2(g2One); writer.writeG2(g2One.mul(frOne.construct(-1))); writer.writeG2(g2One.mul(frOne.construct(2))); @@ -51,14 +54,17 @@ public void testBinaryWriter( final var is = new ByteArrayInputStream(buffer); final var reader = mkReader.apply(is); + assertEquals(frOne.zero(), reader.readFr()); assertEquals(frOne, reader.readFr()); assertEquals(frOne.construct(-1), reader.readFr()); assertEquals(frOne.construct(2), reader.readFr()); assertEquals(frOne.construct(-2), reader.readFr()); + assertEquals(g1One.zero(), reader.readG1()); assertEquals(g1One, reader.readG1()); assertEquals(g1One.mul(frOne.construct(-1)), reader.readG1()); assertEquals(g1One.mul(frOne.construct(2)), reader.readG1()); assertEquals(g1One.mul(frOne.construct(-2)), reader.readG1()); + assertEquals(g2One.zero(), reader.readG2()); assertEquals(g2One, reader.readG2()); assertEquals(g2One.mul(frOne.construct(-1)), reader.readG2()); assertEquals(g2One.mul(frOne.construct(2)), reader.readG2()); From d8fb4891cf700001a7a7126625f2b956c29ec080 Mon Sep 17 00:00:00 2001 From: Duncan Tebbs Date: Thu, 6 May 2021 16:56:07 +0100 Subject: [PATCH 2/4] WIP: RDD checkpointing --- src/main/java/common/PairRDDAggregator.java | 11 ++++++++++- src/main/java/prover/Prover.java | 6 ++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/src/main/java/common/PairRDDAggregator.java b/src/main/java/common/PairRDDAggregator.java index 4790586..ce394f8 100644 --- a/src/main/java/common/PairRDDAggregator.java +++ b/src/main/java/common/PairRDDAggregator.java @@ -53,7 +53,16 @@ public JavaPairRDD aggregate() { } void processBatch() { - batches.add(sc.parallelizePairs(currentBatch, numPartitions)); + System.out.println("processBatch: " + String.valueOf(batches.size())); + final var newBatchRDD = sc.parallelizePairs(currentBatch, numPartitions); + + // To avoid running out of memory, 'checkpoint' the RDD. (The goal is to + // force it to be fully evaluated and then stored on disk, removing any need + // to recompute it, since receomputing requires that the original array of + // batch data must be present in memory somewhere). + newBatchRDD.checkpoint(); + + batches.add(newBatchRDD); currentBatch = null; } diff --git a/src/main/java/prover/Prover.java b/src/main/java/prover/Prover.java index 3b17e57..8ac4bb0 100644 --- a/src/main/java/prover/Prover.java +++ b/src/main/java/prover/Prover.java @@ -125,6 +125,12 @@ static JavaSparkContext createSparkContext(boolean local) { final SparkSession spark = sessionBuilder.getOrCreate(); spark.sparkContext().conf().set("spark.files.overwrite", "true"); + + // checkpoint directory + spark.sparkContext().setCheckpointDir("checkpoint"); + // clean checkpoint files if the reference is out of scope + spark.sparkContext().conf().set("spark.cleaner.referenceTracking.cleanCheckpoints", "true"); + // TODO: reinstate this when it can be made to work // spark.sparkContext().conf().set( // "spark.serializer", From 7aedb51436d3afc6fe4397e918865a791dbf3b58 Mon Sep 17 00:00:00 2001 From: Duncan Tebbs Date: Mon, 10 May 2021 10:30:56 +0100 Subject: [PATCH 3/4] WIP: more tmp changes to avoid out-of-memory --- src/main/java/common/PairRDDAggregator.java | 10 +++++++--- src/main/java/prover/Prover.java | 4 ++-- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/src/main/java/common/PairRDDAggregator.java b/src/main/java/common/PairRDDAggregator.java index ce394f8..4027013 100644 --- a/src/main/java/common/PairRDDAggregator.java +++ b/src/main/java/common/PairRDDAggregator.java @@ -3,6 +3,8 @@ import java.util.ArrayList; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.storage.StorageLevel; + import scala.Tuple2; import scala.collection.JavaConverters; @@ -57,10 +59,12 @@ void processBatch() { final var newBatchRDD = sc.parallelizePairs(currentBatch, numPartitions); // To avoid running out of memory, 'checkpoint' the RDD. (The goal is to - // force it to be fully evaluated and then stored on disk, removing any need - // to recompute it, since receomputing requires that the original array of - // batch data must be present in memory somewhere). + // force it to be fully evaluated (and potentially evicted to disk), + // removing any need to recompute it, since receomputing requires that the + // original array of batch data must be present in memory somewhere). + newBatchRDD.cache(); newBatchRDD.checkpoint(); + // newBatchRDD.persist(StorageLevel.MEMORY_AND_DISK()); batches.add(newBatchRDD); currentBatch = null; diff --git a/src/main/java/prover/Prover.java b/src/main/java/prover/Prover.java index 8ac4bb0..9c6b68b 100644 --- a/src/main/java/prover/Prover.java +++ b/src/main/java/prover/Prover.java @@ -127,9 +127,9 @@ static JavaSparkContext createSparkContext(boolean local) { spark.sparkContext().conf().set("spark.files.overwrite", "true"); // checkpoint directory - spark.sparkContext().setCheckpointDir("checkpoint"); + spark.sparkContext().setCheckpointDir("hdfs://ip-172-31-42-216:9000/checkpoints/"); // clean checkpoint files if the reference is out of scope - spark.sparkContext().conf().set("spark.cleaner.referenceTracking.cleanCheckpoints", "true"); + // spark.sparkContext().conf().set("spark.cleaner.referenceTracking.cleanCheckpoints", "true"); // TODO: reinstate this when it can be made to work // spark.sparkContext().conf().set( From 66b3c12758f307f8e2679659cc044c45c48a0fdf Mon Sep 17 00:00:00 2001 From: Duncan Tebbs Date: Mon, 17 May 2021 12:09:34 +0100 Subject: [PATCH 4/4] FIX: 6e886e5 fix for group element serialization --- .../java/algebra/curves/barreto_lynn_scott/BLSBinaryReader.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/algebra/curves/barreto_lynn_scott/BLSBinaryReader.java b/src/main/java/algebra/curves/barreto_lynn_scott/BLSBinaryReader.java index 5866b46..e0bf1a0 100644 --- a/src/main/java/algebra/curves/barreto_lynn_scott/BLSBinaryReader.java +++ b/src/main/java/algebra/curves/barreto_lynn_scott/BLSBinaryReader.java @@ -84,7 +84,7 @@ public BLSG1T readG1() throws IOException { return G1One.zero(); } - return G1One.construct(readFq(), readFq(), FqOne); + return G1One.construct(x, y, FqOne); } @Override