From 45399c038c6d2d97cb619b17a18c73aefddd633c Mon Sep 17 00:00:00 2001 From: Hellblazer Date: Sun, 14 Apr 2024 17:40:17 -0700 Subject: [PATCH] Further simplification of View Reassembly. Remove ASSEMBLE block type. Further clean up of view assembly --- choam/pom.xml | 4 + .../com/salesforce/apollo/choam/CHOAM.java | 59 ++++---- .../salesforce/apollo/choam/Committee.java | 3 - .../com/salesforce/apollo/choam/Producer.java | 130 +++++------------- .../salesforce/apollo/choam/ViewAssembly.java | 51 +++---- .../salesforce/apollo/choam/ViewContext.java | 56 +++++++- .../salesforce/apollo/choam/fsm/Driven.java | 28 +--- .../apollo/choam/support/ChoamMetrics.java | 2 +- .../choam/support/ChoamMetricsImpl.java | 8 +- .../apollo/choam/support/TxDataSource.java | 44 +++--- .../apollo/choam/GenesisAssemblyTest.java | 5 - grpc/src/main/proto/choam.proto | 14 +- .../salesforce/apollo/model/DomainTest.java | 2 +- .../apollo/model/FireFliesTest.java | 2 +- model/src/test/resources/logback-test.xml | 2 +- pom.xml | 5 + 16 files changed, 175 insertions(+), 240 deletions(-) diff --git a/choam/pom.xml b/choam/pom.xml index d4c07c80e..1a91f65b4 100644 --- a/choam/pom.xml +++ b/choam/pom.xml @@ -30,6 +30,10 @@ org.jooq joou + + com.googlecode.concurrentlinkedhashmap + concurrentlinkedhashmap-lru + diff --git a/choam/src/main/java/com/salesforce/apollo/choam/CHOAM.java b/choam/src/main/java/com/salesforce/apollo/choam/CHOAM.java index 436bedeb3..d05256ed5 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/CHOAM.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/CHOAM.java @@ -11,6 +11,7 @@ import com.google.protobuf.ByteString; import com.google.protobuf.InvalidProtocolBufferException; import com.google.protobuf.Message; +import com.googlecode.concurrentlinkedhashmap.ConcurrentLinkedHashMap; import com.salesforce.apollo.archipelago.RouterImpl.CommonCommunications; import com.salesforce.apollo.bloomFilters.BloomFilter; import com.salesforce.apollo.choam.comm.*; @@ -95,11 +96,15 @@ public class CHOAM { private final TransSubmission txnSubmission = new TransSubmission(); private final AtomicReference view = new AtomicReference<>(); private final AtomicReference> pendingView = new AtomicReference<>(); + private final ConcurrentLinkedHashMap> pendingViews; public CHOAM(Parameters params) { this.store = new Store(params.digestAlgorithm(), params.mvBuilder().clone().build()); this.params = params; executions = Executors.newVirtualThreadPerTaskExecutor(); + pendingViews = new ConcurrentLinkedHashMap.Builder>().maximumWeightedCapacity(100) + .initialCapacity(10) + .build(); nextView(); var bContext = new DelegatedContext<>(params.context()); @@ -321,10 +326,15 @@ public void nextView(Context context, Digest diadem) { if (c != null) { c.nextView(context); } else { - log.info("Acquiring new view, diadem: {} size: {} on: {}", diadem, context.size(), params.member().getId()); + log.info("Acquiring new view of: {}, diadem: {} size: {} on: {}", context.getId(), diadem, context.size(), + params.member().getId()); params.context().setContext(context); pendingView.set(null); } + + log.info("Pushing pending view of: {}, diadem: {} size: {} on: {}", context.getId(), diadem, context.size(), + params.member().getId()); + pendingViews.putIfAbsent(diadem, context); } public void start() { @@ -512,20 +522,6 @@ public void onFailure() { transitions.fail(); } - @Override - public Block produce(ULong height, Digest prev, Assemble assemble, HashedBlock checkpoint) { - final HashedCertifiedBlock v = view.get(); - var block = Block.newBuilder() - .setHeader( - buildHeader(params.digestAlgorithm(), assemble, prev, height, checkpoint.height(), - checkpoint.hash, v.height(), v.hash)) - .setAssemble(assemble) - .build(); - log.trace("Produced block: {} height: {} on: {}", block.getBodyCase(), block.getHeader().getHeight(), - params.member().getId()); - return block; - } - @Override public Block produce(ULong height, Digest prev, Executions executions, HashedBlock checkpoint) { final HashedCertifiedBlock v = view.get(); @@ -674,22 +670,19 @@ private Supplier> pendingView() { }; } + private Supplier>> pendingViews() { + return () -> pendingViews; + } + private void process() { final var c = current.get(); final HashedCertifiedBlock h = head.get(); log.info("Begin block: {} hash: {} height: {} committee: {} on: {}", h.block.getBodyCase(), h.hash, h.height(), c.getClass().getSimpleName(), params.member().getId()); switch (h.block.getBodyCase()) { - case ASSEMBLE: { - params.processor().beginBlock(h.height(), h.hash); - nextViewId.set(Digest.from(h.block.getAssemble().getNextView())); - log.info("Assembled next view id: {} on: {}", nextViewId.get(), params.member().getId()); - c.assembled(); - break; - } case RECONFIGURE: { params.processor().beginBlock(h.height(), h.hash); - reconfigure(h.block.getReconfigure()); + reconfigure(h.hash, h.block.getReconfigure()); break; } case GENESIS: { @@ -697,7 +690,7 @@ private void process() { cancelBootstrap(); transitions.regenerated(); genesisInitialization(h, h.block.getGenesis().getInitializeList()); - reconfigure(h.block.getGenesis().getInitialView()); + reconfigure(h.hash, h.block.getGenesis().getInitialView()); break; } case EXECUTIONS: { @@ -719,9 +712,9 @@ private void process() { params.member().getId()); } - private void reconfigure(Reconfigure reconfigure) { - log.info("Clearing next view id on: {}", params.member().getId()); - nextViewId.set(null); + private void reconfigure(Digest hash, Reconfigure reconfigure) { + log.info("Setting next view id: {} on: {}", hash, params.member().getId()); + nextViewId.set(hash); var pv = pendingView.getAndSet(null); if (pv != null) { // always advance view. @@ -741,7 +734,7 @@ private void reconfigure(Reconfigure reconfigure) { } else { log.warn("Reconfiguration to associate failed: {} in view: {} on:{}", validators.size(), new Digest(reconfigure.getId()), params.member().getId()); - current.set(new Client(validators, getViewId())); + transitions.fail(); } } else { current.set(new Client(validators, getViewId())); @@ -1013,8 +1006,6 @@ public interface BlockProducer { void onFailure(); - Block produce(ULong height, Digest prev, Assemble assemble, HashedBlock checkpoint); - Block produce(ULong height, Digest prev, Executions executions, HashedBlock checkpoint); void publish(Digest hash, CertifiedBlock cb); @@ -1306,16 +1297,12 @@ private class Associate extends Administration { params.member().getId()); Signer signer = new SignerImpl(nextView.consensusKeyPair.getPrivate(), ULong.MIN); Supplier> pv = pendingView(); - producer = new Producer(new ViewContext(context, params, pv, signer, validators, constructBlock()), + producer = new Producer(nextViewId.get(), + new ViewContext(context, params, pv, signer, validators, constructBlock()), head.get(), checkpoint.get(), comm, getLabel()); producer.start(); } - @Override - public void assembled() { - producer.assembled(); - } - @Override public void complete() { producer.stop(); diff --git a/choam/src/main/java/com/salesforce/apollo/choam/Committee.java b/choam/src/main/java/com/salesforce/apollo/choam/Committee.java index c7ad7fcf4..a9c86256c 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/Committee.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/Committee.java @@ -64,9 +64,6 @@ static Set viewMembersOf(Digest hash, Context baseContex void accept(HashedCertifiedBlock next); - default void assembled() { - } - void complete(); boolean isMember(); diff --git a/choam/src/main/java/com/salesforce/apollo/choam/Producer.java b/choam/src/main/java/com/salesforce/apollo/choam/Producer.java index dd6954b33..31219e253 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/Producer.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/Producer.java @@ -21,7 +21,6 @@ import com.salesforce.apollo.choam.support.TxDataSource; import com.salesforce.apollo.cryptography.Digest; import com.salesforce.apollo.cryptography.DigestAlgorithm; -import com.salesforce.apollo.cryptography.JohnHancock; import com.salesforce.apollo.ethereal.Config; import com.salesforce.apollo.ethereal.Config.Builder; import com.salesforce.apollo.ethereal.Ethereal; @@ -45,7 +44,6 @@ public class Producer { private static final Logger log = LoggerFactory.getLogger(Producer.class); private final AtomicBoolean assembled = new AtomicBoolean(); - private final AtomicReference assembly = new AtomicReference<>(); private final AtomicReference checkpoint = new AtomicReference<>(); private final CommonCommunications comms; private final Ethereal controller; @@ -61,15 +59,17 @@ public class Producer { private final AtomicBoolean started = new AtomicBoolean(false); private final Transitions transitions; private final ViewContext view; - private volatile Digest nextViewId; + private final Digest nextViewId; + private ViewAssembly assembly; - public Producer(ViewContext view, HashedBlock lastBlock, HashedBlock checkpoint, + public Producer(Digest nextViewId, ViewContext view, HashedBlock lastBlock, HashedBlock checkpoint, CommonCommunications comms, String label) { assert view != null; this.view = view; this.previousBlock.set(lastBlock); this.comms = comms; this.checkpoint.set(checkpoint); + this.nextViewId = nextViewId; final Parameters params = view.params(); final var producerParams = params.producer(); @@ -112,22 +112,14 @@ public Producer(ViewContext view, HashedBlock lastBlock, HashedBlock checkpoint, log.debug("Roster for: {} is: {} on: {}", getViewId(), view.roster(), params().member().getId()); } - public void assembled() { - transitions.assembled(); - } - - public Digest getNextViewId() { - final Digest current = nextViewId; - return current; - } - public void start() { if (!started.compareAndSet(false, true)) { return; } + reconfigure(); final Block prev = previousBlock.get().block; - if (prev.hasReconfigure() && prev.getReconfigure().getCheckpointTarget() == 0) { // genesis block won't ever be - // 0 + // genesis block won't ever be 0 + if (prev.hasReconfigure() && prev.getReconfigure().getCheckpointTarget() == 0) { transitions.checkpoint(); } else { transitions.start(); @@ -141,7 +133,7 @@ public void stop() { log.trace("Closing producer for: {} on: {}", getViewId(), params().member().getId()); controller.stop(); coordinator.stop(); - final var c = assembly.get(); + final var c = assembly; if (c != null) { c.stop(); } @@ -159,11 +151,11 @@ public SubmitResult submit(Transaction transaction) { } } - private void addReassemble(Reassemble r) { - if (ds.offer(r)) { - log.trace("Adding joins: {} on: {}", r.getMembersList(), params().member().getId()); + private void addJoin(SignedJoin signedJoin) { + if (ds.offer(signedJoin)) { + log.trace("Adding on: {}", params().member().getId()); } else { - log.trace("Cannot add joins: {} on: {}", r.getMembersCount(), params().member().getId()); + log.trace("Cannot add join on: {}", params().member().getId()); } } @@ -189,13 +181,13 @@ private void create(List preblock, boolean last) { .filter(p -> p.witnesses.size() >= params().majority()) .forEach(this::publish); - var joins = aggregate.stream().flatMap(e -> e.getJoinsList().stream()).filter(j -> validate(j)).toList(); - final var ass = assembly.get(); + var joins = aggregate.stream().flatMap(e -> e.getJoinsList().stream()).filter(view::validate).toList(); + final var ass = assembly; if (ass != null) { - log.trace("Consuming joins: {} on: {}", aggregate.size(), joins.size(), params().member().getId()); + log.trace("Consuming {} units, {} joins on: {}", aggregate.size(), joins.size(), params().member().getId()); ass.inbound().accept(joins); } else { - log.trace("Pending joins: {} on: {}", aggregate.size(), joins.size(), params().member().getId()); + log.trace("Pending {} units, {} joins on: {}", aggregate.size(), joins.size(), params().member().getId()); pendingJoins.addAll(joins); } @@ -254,30 +246,6 @@ private void processPendingValidations(HashedBlock block, PendingBlock p) { } } - private void produceAssemble() { - final var vlb = previousBlock.get(); - nextViewId = vlb.hash; - for (var m : Committee.viewMembersOf(nextViewId, view.pendingView())) { - nextAssembly.put(m.getId(), m); - } - log.debug("Assembling: {} on: {}", nextViewId, params().member().getId()); - final var assemble = new HashedBlock(params().digestAlgorithm(), view.produce(vlb.height().add(1), vlb.hash, - Assemble.newBuilder() - .setNextView( - vlb.hash.toDigeste()) - .build(), - checkpoint.get())); - previousBlock.set(assemble); - final var validation = view.generateValidation(assemble); - final var p = new PendingBlock(assemble, new HashMap<>(), new AtomicBoolean()); - pending.put(assemble.hash, p); - p.witnesses.put(params().member(), validation); - ds.offer(validation); - log.debug("Produced block: {} hash: {} height: {} from: {} on: {}", assemble.block.getBodyCase(), assemble.hash, - assemble.height(), getViewId(), params().member().getId()); - processPendingValidations(assemble, p); - } - private void publish(PendingBlock p) { assert p.witnesses.size() >= params().majority() : "Publishing non majority block"; log.debug("Published pending: {} hash: {} height: {} witnesses: {} on: {}", p.block.block.getBodyCase(), @@ -291,26 +259,23 @@ private void publish(PendingBlock p) { view.publish(new HashedCertifiedBlock(params().digestAlgorithm(), cb)); } - private boolean validate(SignedJoin join) { - var mid = Digest.from(join.getMember()); - var m = nextAssembly.get(mid); - if (m == null) { - log.trace("Cannot validate join view: {} of: {} signed by: {} on: {}", - Digest.from(join.getJoin().getVm().getView()), Digest.from(join.getJoin().getVm().getId()), mid, - params().member().getId()); - return false; - } - var validated = m.verify(JohnHancock.from(join.getSignature()), join.getJoin().toByteString()); - if (!validated) { - log.trace("Cannot validate view join: {} of: {} signed by: {} on: {}", - Digest.from(join.getJoin().getVm().getView()), Digest.from(join.getJoin().getVm().getId()), mid, - params().member().getId()); - } else { - log.trace("Validated view join: {} of: {} signed by: {} on: {}", - Digest.from(join.getJoin().getVm().getView()), Digest.from(join.getJoin().getVm().getId()), mid, - params().member().getId()); - } - return validated; + private void reconfigure() { + log.debug("Starting view reconfiguration: {} on: {}", nextViewId, params().member().getId()); + assembly = new ViewAssembly(nextViewId, view, Producer.this::addJoin, comms) { + @Override + public void complete() { + super.complete(); + log.debug("View reconfiguration: {} gathered: {} complete on: {}", nextViewId, getSlate().size(), + params().member().getId()); + assembled.set(true); + Producer.this.transitions.viewComplete(); + } + }; + assembly.start(); + assembly.assembled(); + var joins = new ArrayList<>(pendingJoins); + pendingJoins.clear(); + assembly.inbound().accept(joins); } private PendingBlock validate(Validate v) { @@ -345,7 +310,7 @@ public void assembled() { log.debug("assembly already complete on: {}", params().member().getId()); return; } - final var slate = assembly.get().getSlate(); + final var slate = assembly.getSlate(); var reconfiguration = new HashedBlock(params().digestAlgorithm(), view.reconfigure(slate, nextViewId, previousBlock.get(), checkpoint.get())); @@ -368,7 +333,7 @@ public void checkAssembly() { if (dropped != 0) { log.warn("Dropped txns: {} on: {}", dropped, params().member().getId()); } - final var viewAssembly = assembly.get(); + final var viewAssembly = assembly; if (viewAssembly == null) { log.error("Assemble block never processed on: {}", params().member().getId()); transitions.failed(); @@ -420,31 +385,6 @@ public void fail() { view.onFailure(); } - @Override - public void produceAssemble() { - Producer.this.produceAssemble(); - } - - @Override - public void reconfigure() { - log.debug("Starting view reconfiguration: {} on: {}", nextViewId, params().member().getId()); - assembly.set(new ViewAssembly(nextViewId, view, Producer.this::addReassemble, comms) { - @Override - public void complete() { - super.complete(); - log.debug("View reconfiguration: {} gathered: {} complete on: {}", nextViewId, getSlate().size(), - params().member().getId()); - assembled.set(true); - Producer.this.transitions.viewComplete(); - } - }); - assembly.get().start(); - assembly.get().assembled(); - var joins = new ArrayList<>(pendingJoins); - pendingJoins.clear(); - assembly.get().inbound().accept(joins); - } - @Override public void startProduction() { log.debug("Starting production for: {} on: {}", getViewId(), params().member().getId()); diff --git a/choam/src/main/java/com/salesforce/apollo/choam/ViewAssembly.java b/choam/src/main/java/com/salesforce/apollo/choam/ViewAssembly.java index 1cfc03a25..847aac3c7 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/ViewAssembly.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/ViewAssembly.java @@ -13,7 +13,6 @@ import com.salesforce.apollo.choam.fsm.Reconfiguration.Reconfigure; import com.salesforce.apollo.choam.fsm.Reconfiguration.Transitions; import com.salesforce.apollo.choam.proto.Join; -import com.salesforce.apollo.choam.proto.Reassemble; import com.salesforce.apollo.choam.proto.SignedJoin; import com.salesforce.apollo.choam.proto.SignedViewMember; import com.salesforce.apollo.context.Context; @@ -55,7 +54,7 @@ public class ViewAssembly { private final AtomicBoolean cancelSlice = new AtomicBoolean(); private final Digest nextViewId; private final Map proposals = new ConcurrentHashMap<>(); - private final Consumer publisher; + private final Consumer publisher; private final Map slate = new ConcurrentSkipListMap<>(); private final ViewContext view; private final CommonCommunications comms; @@ -63,7 +62,7 @@ public class ViewAssembly { new ConcurrentSkipListMap<>()); private volatile Map nextAssembly; - public ViewAssembly(Digest nextViewId, ViewContext vc, Consumer publisher, + public ViewAssembly(Digest nextViewId, ViewContext vc, Consumer publisher, CommonCommunications comms) { view = vc; this.nextViewId = nextViewId; @@ -118,23 +117,17 @@ Consumer> inbound() { }; } - private void completeSlice(AtomicReference retryDelay, AtomicReference reiterate) { + private void completeSlice(Duration retryDelay, AtomicReference reiterate) { if (gathered()) { return; } - final var delay = retryDelay.get(); - if (delay.compareTo(params().producer().maxGossipDelay()) < 0) { - retryDelay.accumulateAndGet(Duration.ofMillis(100), Duration::plus); - } - - log.trace("Proposal incomplete of: {} polled: {}, total: {} majority: {}, retrying: {} on: {}", nextViewId, - polled.stream().sorted().toList(), nextAssembly.size(), params().majority(), delay, - params().member().getId()); + log.trace("Proposal incomplete of: {} polled: {}, total: {} retrying: {} on: {}", nextViewId, + polled.stream().sorted().toList(), nextAssembly.size(), retryDelay, params().member().getId()); if (!cancelSlice.get()) { Executors.newScheduledThreadPool(1, Thread.ofVirtual().factory()) - .schedule(() -> Thread.ofVirtual().start(Utils.wrapped(reiterate.get(), log)), delay.toMillis(), - TimeUnit.MILLISECONDS); + .schedule(() -> Thread.ofVirtual().start(Utils.wrapped(reiterate.get(), log)), + retryDelay.toNanos(), TimeUnit.NANOSECONDS); } } @@ -217,16 +210,25 @@ private void join(SignedViewMember svm, boolean direct) { } return; } + polled.add(mid); if (proposals.putIfAbsent(mid, svm) == null) { - if (log.isTraceEnabled()) { - log.trace("Adding view member: {} on: {}", ViewContext.print(svm, params().digestAlgorithm()), - params().member().getId()); - } if (direct) { - publisher.accept(Reassemble.newBuilder().addMembers(svm).build()); + var sig = params().member().sign(svm.toByteString()).toSig(); + publisher.accept(SignedJoin.newBuilder() + .setJoin(svm) + .setMember(params().member().getId().toDigeste()) + .setSignature(sig) + .build()); + if (log.isTraceEnabled()) { + log.trace("Publishing view member: {} sig: {} on: {}", + ViewContext.print(svm, params().digestAlgorithm()), + params().digestAlgorithm().digest(sig.toByteString()), params().member().getId()); + } + } else if (log.isTraceEnabled()) { + log.trace("Adding discovered view member: {} on: {}", + ViewContext.print(svm, params().digestAlgorithm()), params().member().getId()); } } - polled.add(mid); } private Parameters params() { @@ -262,13 +264,12 @@ public void elect() { proposals.entrySet().stream().forEach(e -> slate.put(e.getKey(), joinOf(e.getValue()))); if (slate.size() == view.pendingView().getRingCount()) { cancelSlice.set(true); - log.debug("Electing: {} of: {} slate: {} proposals: {} on: {}", slate.size(), nextViewId, - slate.keySet().stream().sorted().toList(), proposals.keySet().stream().sorted().toList(), - params().member().getId()); + log.debug("Electing: {} of: {} slate: {} on: {}", slate.size(), nextViewId, + slate.keySet().stream().sorted().toList(), params().member().getId()); transitions.complete(); } else { Context memberContext = view.pendingView(); - log.error("Failed election, required: {} slate: {} of: {} on: {}", memberContext.majority(), + log.error("Failed election, required: {} slate: {} of: {} on: {}", view.pendingView().getRingCount(), proposals.keySet().stream().sorted().toList(), nextViewId, params().member().getId()); } } @@ -284,7 +285,7 @@ public void failed() { public void gather() { log.trace("Gathering assembly for: {} on: {}", nextViewId, params().member().getId()); AtomicReference reiterate = new AtomicReference<>(); - AtomicReference retryDelay = new AtomicReference<>(Duration.ofMillis(10)); + var retryDelay = Duration.ofMillis(10); reiterate.set(() -> { nextAssembly = Committee.viewMembersOf(nextViewId, view.pendingView()) .stream() diff --git a/choam/src/main/java/com/salesforce/apollo/choam/ViewContext.java b/choam/src/main/java/com/salesforce/apollo/choam/ViewContext.java index 1a64d79ae..d7aa35adc 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/ViewContext.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/ViewContext.java @@ -144,10 +144,6 @@ public Context pendingView() { return pendingView.get(); } - public Block produce(ULong l, Digest hash, Assemble assemble, HashedBlock checkpoint) { - return blockProducer.produce(l, hash, assemble, checkpoint); - } - public Block produce(ULong l, Digest hash, Executions executions, HashedBlock checkpoint) { return blockProducer.produce(l, hash, executions, checkpoint); } @@ -191,12 +187,35 @@ public boolean validate(SignedViewMember svm, Validate validate) { return valid; } + public boolean validate(SignedJoin join) { + if (true) { + return true; + } + Verifier v = verifierOf(join); + if (v == null) { + log.debug("no verifier: {} for join: {} on: {}", Digest.from(join.getMember()), + Digest.from(join.getJoin().getVm().getId()), params.member().getId()); + return false; + } + var validated = v.verify(JohnHancock.from(join.getSignature()), join.getJoin().toByteString()); + if (!validated) { + log.trace("Cannot validate view join: [{}] sig: {} signed by: {} on: {}", + print(join.getJoin(), params.digestAlgorithm()), + params.digestAlgorithm().digest(join.getSignature().toByteString()), + Digest.from(join.getMember()), params().member().getId()); + } else { + log.trace("Validated view join: [{}] signed by: {} on: {}", print(join.getJoin(), params.digestAlgorithm()), + Digest.from(join.getMember()), params().member().getId()); + } + return validated; + } + protected Verifier verifierOf(Validate validate) { final var mid = Digest.from(validate.getWitness().getId()); var m = context.getMember(mid); if (m == null) { if (log.isDebugEnabled()) { - log.debug("Unable to validate key by non existent validator: [{}] on: {}", + log.debug("Unable to get verifier by non existent member: [{}] on: {}", print(validate, params.digestAlgorithm()), params.member().getId()); } return null; @@ -204,11 +223,36 @@ protected Verifier verifierOf(Validate validate) { Verifier v = validators.get(m); if (v == null) { if (log.isDebugEnabled()) { - log.debug("Unable to validate key by non existent validator: [{}] on: {}", + log.debug("Unable to get verifier by non existent validator: [{}] on: {}", print(validate, params.digestAlgorithm()), params.member().getId()); } return null; } return v; } + + protected Verifier verifierOf(SignedJoin sj) { + final var mid = Digest.from(sj.getMember()); + var m = context.getMember(mid); + if (m == null) { + if (log.isDebugEnabled()) { + log.debug("Unable to get verifier by non existent member: [{}] on: {}", + String.format("id: %s sig: %s", Digest.from(sj.getMember()), + params.digestAlgorithm().digest(sj.getSignature().toByteString())), + params.member().getId()); + } + return null; + } + Verifier v = validators.get(m); + if (v == null) { + if (log.isDebugEnabled()) { + log.debug("Unable to validate key by non existent validator: [{}] on: {}", + String.format("id: %s sig: %s", Digest.from(sj.getMember()), + params.digestAlgorithm().digest(sj.getSignature().toByteString())), + params.member().getId()); + } + return null; + } + return v; + } } diff --git a/choam/src/main/java/com/salesforce/apollo/choam/fsm/Driven.java b/choam/src/main/java/com/salesforce/apollo/choam/fsm/Driven.java index d0f8c43e8..88ae2bada 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/fsm/Driven.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/fsm/Driven.java @@ -33,20 +33,10 @@ public interface Driven { void fail(); - void produceAssemble(); - - void reconfigure(); - void startProduction(); enum Earner implements Driven.Transitions { AWAIT_VIEW { - @Override - public Transitions assembled() { - context().assembled(); - return null; - } - @Entry public void checkAssembly() { context().checkAssembly(); @@ -125,21 +115,9 @@ public void terminate() { context().fail(); } }, SPICE { - @Override - public Transitions assembled() { - context().reconfigure(); - return null; - } - @Override public Transitions newEpoch(int epoch, int lastEpoch) { - if (lastEpoch == epoch) { - return AWAIT_VIEW; - } - if (epoch == 0) { - context().produceAssemble(); - } - return null; + return (lastEpoch == epoch) ? AWAIT_VIEW : null; } @Entry @@ -158,10 +136,6 @@ public Transitions viewComplete() { interface Transitions extends FsmExecutor { Logger log = LoggerFactory.getLogger(Transitions.class); - default Transitions assembled() { - return null; - } - default Transitions checkpoint() { throw fsm().invalidTransitionOn(); } diff --git a/choam/src/main/java/com/salesforce/apollo/choam/support/ChoamMetrics.java b/choam/src/main/java/com/salesforce/apollo/choam/support/ChoamMetrics.java index 77b62f177..f1d6129f9 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/support/ChoamMetrics.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/support/ChoamMetrics.java @@ -27,7 +27,7 @@ public interface ChoamMetrics extends EndpointMetrics { EtherealMetrics getProducerMetrics(); - void publishedBatch(int batchSize, int byteSize, int validations, int reassemblies); + void publishedBatch(int batchSize, int byteSize, int validations, int joins); void transactionCancelled(); diff --git a/choam/src/main/java/com/salesforce/apollo/choam/support/ChoamMetricsImpl.java b/choam/src/main/java/com/salesforce/apollo/choam/support/ChoamMetricsImpl.java index 6ff2e0e84..71b732242 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/support/ChoamMetricsImpl.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/support/ChoamMetricsImpl.java @@ -35,7 +35,7 @@ public class ChoamMetricsImpl extends EndpointMetricsImpl implements ChoamMetric private final EtherealMetrics genesisMetrics; private final EtherealMetrics producerMetrics; private final Histogram publishedBytes; - private final Meter publishedReassemblies; + private final Meter publishedJoins; private final Meter publishedTransactions; private final Meter publishedValidations; private final MetricRegistry registry; @@ -67,7 +67,7 @@ public ChoamMetricsImpl(Digest context, MetricRegistry registry) { cancelledTransactions = registry.meter(name(context.shortString(), "transactions.cancelled")); publishedTransactions = registry.meter(name(context.shortString(), "transactions.published")); publishedBytes = registry.histogram(name(context.shortString(), "unit.bytes")); - publishedReassemblies = registry.meter(name(context.shortString(), "reassemblies.published")); + publishedJoins = registry.meter(name(context.shortString(), "joins.published")); publishedValidations = registry.meter(name(context.shortString(), "validations.published")); transactionLatency = registry.timer(name(context.shortString(), "transaction.latency")); transactionSubmitRetry = registry.meter(name(context.shortString(), "transaction.submit.retry")); @@ -117,11 +117,11 @@ public EtherealMetrics getProducerMetrics() { } @Override - public void publishedBatch(int transactions, int byteSize, int validations, int reassemblies) { + public void publishedBatch(int transactions, int byteSize, int validations, int joins) { publishedTransactions.mark(transactions); publishedBytes.update(byteSize); publishedValidations.mark(validations); - publishedReassemblies.mark(reassemblies); + publishedJoins.mark(joins); } @Override diff --git a/choam/src/main/java/com/salesforce/apollo/choam/support/TxDataSource.java b/choam/src/main/java/com/salesforce/apollo/choam/support/TxDataSource.java index 5acb7c4cb..4a34fc326 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/support/TxDataSource.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/support/TxDataSource.java @@ -7,7 +7,7 @@ package com.salesforce.apollo.choam.support; import com.google.protobuf.ByteString; -import com.salesforce.apollo.choam.proto.Reassemble; +import com.salesforce.apollo.choam.proto.SignedJoin; import com.salesforce.apollo.choam.proto.Transaction; import com.salesforce.apollo.choam.proto.UnitData; import com.salesforce.apollo.choam.proto.Validate; @@ -37,13 +37,13 @@ public class TxDataSource implements DataSource { private final static Logger log = LoggerFactory.getLogger(TxDataSource.class); private final Duration batchInterval; - private final AtomicBoolean draining = new AtomicBoolean(); + private final AtomicBoolean draining = new AtomicBoolean(); private final ExponentialBackoffPolicy drainPolicy; private final Member member; private final ChoamMetrics metrics; private final BatchingQueue processing; - private final BlockingQueue reassemblies = new LinkedBlockingQueue<>(); - private final BlockingQueue validations = new LinkedBlockingQueue<>(); + private final BlockingQueue joins = new LinkedBlockingQueue<>(); + private final BlockingQueue validations = new LinkedBlockingQueue<>(); private volatile Thread blockingThread; public TxDataSource(Member member, int maxElements, ChoamMetrics metrics, int maxBatchByteSize, @@ -63,10 +63,10 @@ public void close() { } blockingThread = null; if (metrics != null) { - metrics.dropped(processing.size(), validations.size(), reassemblies.size()); + metrics.dropped(processing.size(), validations.size(), joins.size()); } log.trace("Closing with remaining txns: {}({}:{}) validations: {} reassemblies: {} on: {}", processing.size(), - processing.added(), processing.taken(), validations.size(), reassemblies.size(), member.getId()); + processing.added(), processing.taken(), validations.size(), joins.size(), member.getId()); } public void drain() { @@ -81,23 +81,23 @@ public ByteString getData() { log.trace("Requesting unit data on: {}", member.getId()); blockingThread = Thread.currentThread(); try { - var r = new ArrayList(); + var r = new ArrayList(); var v = new ArrayList(); if (draining.get()) { var target = Instant.now().plus(drainPolicy.nextBackoff()); - while (target.isAfter(Instant.now()) && builder.getReassembliesCount() == 0 + while (target.isAfter(Instant.now()) && builder.getJoinsCount() == 0 && builder.getValidationsCount() == 0) { // rinse and repeat - r = new ArrayList(); - reassemblies.drainTo(r); - builder.addAllReassemblies(r); + r = new ArrayList<>(); + joins.drainTo(r); + builder.addAllJoins(r); v = new ArrayList(); validations.drainTo(v); builder.addAllValidations(v); - if (builder.getReassembliesCount() != 0 || builder.getValidationsCount() != 0) { + if (builder.getJoinsCount() != 0 || builder.getValidationsCount() != 0) { break; } @@ -122,9 +122,9 @@ public ByteString getData() { } // One more time into ye breech - r = new ArrayList(); - reassemblies.drainTo(r); - builder.addAllReassemblies(r); + r = new ArrayList<>(); + joins.drainTo(r); + builder.addAllJoins(r); v = new ArrayList(); validations.drainTo(v); @@ -133,11 +133,11 @@ public ByteString getData() { ByteString bs = builder.build().toByteString(); if (metrics != null) { metrics.publishedBatch(builder.getTransactionsCount(), bs.size(), builder.getValidationsCount(), - builder.getReassembliesCount()); + builder.getJoinsCount()); } - log.trace("Unit data: {} txns, {} validations, {} reassemblies totalling: {} bytes on: {}", - builder.getTransactionsCount(), builder.getValidationsCount(), builder.getReassembliesCount(), - bs.size(), member.getId()); + log.trace("Unit data: {} txns, {} validations, {} joins totalling: {} bytes on: {}", + builder.getTransactionsCount(), builder.getValidationsCount(), builder.getJoinsCount(), bs.size(), + member.getId()); return bs; } finally { blockingThread = null; @@ -145,7 +145,7 @@ public ByteString getData() { } public int getRemainingReassemblies() { - return reassemblies.size(); + return joins.size(); } public int getRemainingTransactions() { @@ -156,8 +156,8 @@ public int getRemainingValidations() { return validations.size(); } - public boolean offer(Reassemble reassembly) { - return reassemblies.offer(reassembly); + public boolean offer(SignedJoin signedJoin) { + return joins.offer(signedJoin); } public boolean offer(Transaction txn) { diff --git a/choam/src/test/java/com/salesforce/apollo/choam/GenesisAssemblyTest.java b/choam/src/test/java/com/salesforce/apollo/choam/GenesisAssemblyTest.java index 83d441916..136b7e37a 100644 --- a/choam/src/test/java/com/salesforce/apollo/choam/GenesisAssemblyTest.java +++ b/choam/src/test/java/com/salesforce/apollo/choam/GenesisAssemblyTest.java @@ -147,11 +147,6 @@ public void onFailure() { // do nothing } - @Override - public Block produce(ULong height, Digest prev, Assemble assemble, HashedBlock checkpoint) { - return null; - } - @Override public Block produce(ULong height, Digest prev, Executions executions, HashedBlock checkpoint) { return null; diff --git a/grpc/src/main/proto/choam.proto b/grpc/src/main/proto/choam.proto index 5c5c7416c..21d653cb8 100644 --- a/grpc/src/main/proto/choam.proto +++ b/grpc/src/main/proto/choam.proto @@ -47,7 +47,6 @@ message Block { Reconfigure reconfigure = 3; Checkpoint checkpoint = 4; Executions executions = 5; - Assemble assemble = 6; } } @@ -84,10 +83,6 @@ message Executions { repeated Transaction executions = 1; } -message Assemble { - crypto.Digeste nextView = 1; -} - message FoundationSeal { stereotomy.KeyEvent_ foundation = 1; crypto.Sig signature = 2; @@ -103,8 +98,7 @@ message Transaction { message UnitData { repeated Validate validations = 1; repeated Transaction transactions = 2; - repeated Reassemble reassemblies = 3; - repeated SignedJoin joins = 4; + repeated SignedJoin joins = 3; } message Join { @@ -157,12 +151,6 @@ message Validate { Certification witness = 2; } -message Reassemble { - repeated SignedViewMember members = 1; - repeated Validate validations = 2; - repeated crypto.Digeste slate = 3; -} - message Validations { repeated Validate validations = 1; } diff --git a/model/src/test/java/com/salesforce/apollo/model/DomainTest.java b/model/src/test/java/com/salesforce/apollo/model/DomainTest.java index 4c72ce176..45a6e9e47 100644 --- a/model/src/test/java/com/salesforce/apollo/model/DomainTest.java +++ b/model/src/test/java/com/salesforce/apollo/model/DomainTest.java @@ -258,7 +258,7 @@ private Builder params() { .setMaxBatchCount(3000) .build()) .setCheckpointBlockDelta(200); - params.getProducer().ethereal().setNumberOfEpochs(4); + params.getProducer().ethereal().setEpochLength(7).setNumberOfEpochs(3); return params; } } diff --git a/model/src/test/java/com/salesforce/apollo/model/FireFliesTest.java b/model/src/test/java/com/salesforce/apollo/model/FireFliesTest.java index 9140f7793..509577801 100644 --- a/model/src/test/java/com/salesforce/apollo/model/FireFliesTest.java +++ b/model/src/test/java/com/salesforce/apollo/model/FireFliesTest.java @@ -181,7 +181,7 @@ private Builder params() { .build()) .setCheckpointBlockDelta(200); - params.getProducer().ethereal().setNumberOfEpochs(5); + params.getProducer().ethereal().setEpochLength(7).setNumberOfEpochs(3); return params; } } diff --git a/model/src/test/resources/logback-test.xml b/model/src/test/resources/logback-test.xml index 6c5f75706..87a97a0aa 100644 --- a/model/src/test/resources/logback-test.xml +++ b/model/src/test/resources/logback-test.xml @@ -62,7 +62,7 @@ - + diff --git a/pom.xml b/pom.xml index 977d4005d..c1a3a9a7e 100644 --- a/pom.xml +++ b/pom.xml @@ -558,6 +558,11 @@ org-netbeans-modules-keyring RELEASE200 + + com.googlecode.concurrentlinkedhashmap + concurrentlinkedhashmap-lru + 1.4.2 +