gnodet-bot commented on code in PR #29664: URL: https://github.com/apache/camel/pull/29664#discussion_r4544756595
########## components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/Commits.java: ########## @@ -0,0 +1,588 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.sanitizer; + +import java.lang.ref.WeakReference; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.WeakHashMap; + +import org.apache.camel.spi.Synchronization; +import org.apache.camel.spi.UnitOfWork; + +/** + * The commits of consumers that rule R2 follows: where each one is (the unit of work that holds it, or the exchange + * that holds it while it has no unit of work, such as a copy in a seda queue), and how often it ran. It keeps no + * exchange. + * <p/> + * A commit is an on completion that an exchange holds when it gets a unit of work created for it (it moves into the + * unit of work then). The unit of work of an exchange is the one it is given first (the routing engine creates a unit + * of work for an exchange and gives it to that exchange at once; an exchange that shares it, such as a part of a split, + * is given it later). There is one record per add, as one {@link Synchronization} can be added to many exchanges, or + * twice to one; a record holds its on completion weakly, by identity. + * <p/> + * The places are a weak map of the units of work (an entry goes when its unit of work is gone, so nothing is removed by + * hand and no exchange is kept: the place holds the id of the exchange it was created for), and a map of the exchanges + * that hold commits without a unit of work, by exchange id. That map and the commits taken out to be run by someone + * else are bounded ({@link #MAX_FOLLOWED}): past it the oldest is no longer followed, and counted. A unit of work that + * is done keeps its records for as long as it lives, so a commit that it runs again is seen. + * <p/> + * All methods take the lock of this table, which is held briefly and never while calling out of the sanitizer (the + * suppliers of stacks are the sanitizer's). The events of one thread come in the order of the code; events of different + * threads in any order, but Camel hands over to a copy before it publishes the copy (such as before it queues it), so + * the events of a copy come after the handover to it. + */ +final class Commits { + + /** + * The most commits followed in the exchanges that hold them without a unit of work, and taken out; a safety net, + * not a sampling limit. + */ + static final int MAX_FOLLOWED = 10_000; + + /** + * The most units of work that are done kept by the id of their exchange, for a second run of a commit that comes + * after the exchange let go of its unit of work (such as a second done of the unit of work). + */ + static final int MAX_DONE = 1000; + + /** + * The most commits taken out that ran kept for a second run (not part of {@link #MAX_FOLLOWED}). + */ + static final int MAX_RAN = 1000; + + // guarded by this + private final Map<UnitOfWork, Place> unitsOfWork = new WeakHashMap<>(); + // the exchanges that hold commits without a unit of work, by exchange id, the oldest first + private final Map<String, Place> exchanges = new LinkedHashMap<>(); + // the commits taken out to be run by someone else (handoverCompletions without a target), the oldest first + private final ArrayDeque<Commit> taken = new ArrayDeque<>(); + // the last commits taken out that ran, for a second run (not counted in the bound), the oldest first + private final ArrayDeque<Commit> ran = new ArrayDeque<>(); + // how many commits in taken and ran have an on completion of this identity hash, so a run of an on completion that + // is not one of them does not look through them + private final Map<Integer, Integer> takenIndex = new HashMap<>(); + // the last units of work that are done, by the id of the exchange they were created for, the oldest first + private final Map<String, Place> done = new LinkedHashMap<>(64, 0.75f, false) { + @Override + protected boolean removeEldestEntry(Map.Entry<String, Place> eldest) { + return size() > MAX_DONE; + } + }; + private int followed; + // read without the lock to log once that the bound was reached + private volatile long dropped; + private long commits; + + /** + * An exchange is given a unit of work, and the on completions it held move into it. + * + * @param exchangeId the exchange + * @param routeId the route the exchange came from (for a report), or null + * @param endpoint the URI of the endpoint the exchange came from (for a report), or null + */ + synchronized void unitOfWork( + String exchangeId, String routeId, String endpoint, UnitOfWork unitOfWork, List<Synchronization> moved) { + Place place = unitOfWorkPlace(unitOfWork); + if (place.owner == null) { + // the first exchange given it is the one it was created for + place.owner = exchangeId; + } + Place held = exchanges.remove(exchangeId); + if (held != null) { + followed -= held.commits.size(); + } + boolean owner = exchangeId.equals(place.owner); + for (Synchronization s : moved) { + Commit c = held != null ? take(held.commits, s) : null; + if (c != null) { + // a commit handed over to this exchange moves into its unit of work + place.add(c); + } else if (owner) { + commits++; + place.add(new Commit(s, exchangeId, routeId, endpoint)); + } + } + if (held != null) { + // the list of an exchange moves whole, so nothing should be left + for (Commit c : held.commits) { + c.place = null; + } + } + } + + /** + * An on completion is added to a unit of work: it is one of Camel, which reports its adds and its runs. + */ + synchronized void added(UnitOfWork unitOfWork) { + Place place = unitsOfWork.get(unitOfWork); + if (place != null) { + place.reportsRuns = true; + } + } + + /** + * An on completion is removed from a unit of work, so it does not run. + */ + synchronized void removed(UnitOfWork unitOfWork, Synchronization s) { + Place place = unitsOfWork.get(unitOfWork); + if (place != null) { + Commit c = take(place.commits, s); + if (c != null) { + c.place = null; + } + } + } + + /** + * On completions are handed over from one exchange (from its unit of work, or from itself when it has none) to + * another (into its unit of work, or itself when it has none). + * + * @param step the step that runs on this thread (for a report), or null + * @return what moved and the violations found, for the stack of this thread to be added outside the lock (null + * if no commit moved and nothing was found) + */ + synchronized Handover handover( + String fromId, UnitOfWork fromUnitOfWork, String toId, UnitOfWork toUnitOfWork, List<Synchronization> moved, + SanitizerStepListener.StepCall step) { + Place from = fromUnitOfWork != null ? unitsOfWork.get(fromUnitOfWork) : exchanges.get(fromId); + if (from == null || moved == null) { + // no commit there; or a unit of work of a user that does not tell what moved (its commits are no longer + // known to be there, but it reports no run either, so none is reported as never run) + return null; + } + Handover result = null; + Place to = null; + for (Synchronization s : moved) { + Commit c = take(from.commits, s); + if (c == null) { + Commit gone = from.handedOver != null ? find(from.handedOver, s) : null; + if (gone != null && gone.place != null) { + // it left this unit of work by an earlier handover: it is handed over twice + result = result != null ? result : new Handover(fromId); + result.found.add(new Finding(Violation.COMMIT_HANDED_OVER_TWICE, gone, fromId, toId, step)); + } + continue; + } + result = result != null ? result : new Handover(fromId); + if (from.unitOfWork && from.owner != null && !from.owner.equals(fromId)) { + result.found.add(new Finding(Violation.COMMIT_HANDED_OVER_BY_OTHER_EXCHANGE, c, fromId, toId, step)); + } + if (to == null) { + to = toUnitOfWork != null ? unitOfWorkPlace(toUnitOfWork) : exchangePlace(toId); + } + if (from.unitOfWork) { + c.handedOverBy = fromId; + // the stack is added outside the lock (handedOver) + c.handover = null; + result.fromUnitOfWork.add(c); + if (from.handedOver == null) { + from.handedOver = new ArrayList<>(2); + } + from.handedOver.add(c); + } else { + followed--; + } + if (to.handedOver != null) { + // back where it was handed over from (such as from a copy that seda did not queue) + to.handedOver.remove(c); + } + to.add(c); + if (!to.unitOfWork) { + follow(); + } + } + if (!from.unitOfWork && from.commits.isEmpty()) { + exchanges.remove(fromId, from); + } + return result; + } + + /** + * Adds the stack of the thread that handed over to the commits that left a unit of work, for a report of a second + * handover (a second handover that comes before this shows no stack of the first). + */ + synchronized void handedOver(Handover handover, Sanitizer.Trace trace) { + for (Commit c : handover.fromUnitOfWork) { + // not when another handover of it came in between (it shows its own exchange) + if (c.handover == null && handover.fromId.equals(c.handedOverBy)) { + c.handover = trace; + } + } + } + + /** + * On completions are taken out of an exchange (out of the exchange itself: what it holds without a unit of work), + * to be run by someone else: they are followed for running once only. + */ + synchronized void taken(String fromId, List<Synchronization> list) { + Place from = exchanges.get(fromId); + if (from == null) { + return; + } + for (Synchronization s : list) { + Commit c = take(from.commits, s); + if (c != null) { + // it left the exchange (followed) for the commits taken out (followed too) + c.place = Place.TAKEN; + taken.add(c); + takenIndex.merge(c.hash, 1, Integer::sum); + } + } + if (from.commits.isEmpty()) { + exchanges.remove(fromId, from); + } + } + + /** + * An on completion is about to run, with an exchange and its unit of work. + * + * @return the run of a commit, for the stack of this thread to be added outside the lock; null if it is not a + * commit + */ + synchronized Run run(String exchangeId, UnitOfWork unitOfWork, Synchronization s) { + Place place = unitOfWork != null ? unitsOfWork.get(unitOfWork) : null; + if (place == null) { + // the exchange let go of its unit of work when it was done: a unit of work run again after that + place = done.get(exchangeId); + } + Commit c = place != null ? find(place.commits, s) : null; + if (c == null && place != null && place.handedOver != null) { + // it was handed over from this unit of work, but runs here: it runs again where it was handed over to + c = find(place.handedOver, s); + } + if (c == null && takenIndex.containsKey(System.identityHashCode(s))) { + // taken out (or ran after it was taken out): matched by the on completion only, as nothing tells which + // exchange runs it; an on completion shared by many exchanges can match the record of another exchange + c = find(taken, s); + if (c != null) { + // taken out and run once: no longer followed, but kept a while for a second run + taken.remove(c); + followed--; + ran.add(c); + if (ran.size() > MAX_RAN) { + Commit gone = ran.removeFirst(); + gone.place = null; + unindex(gone); + } + } else { + c = find(ran, s); Review Comment: **[low]** `find(ran, s)` scans the entire `ran` deque linearly (up to `MAX_RAN`=1000 entries) on a second run of a taken-out commit. `taken.remove(c)` in the happy-path branch above is similarly O(n). Acceptable for a dev/test tool, but worth a comment so a future maintainer doesn't raise `MAX_RAN` to 100k without reconsidering the data structure. ########## components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/SanitizerContextServicePlugin.java: ########## @@ -188,6 +203,8 @@ public void onRoutesRemove(Collection<Route> routes) { public void onContextStopped(CamelContext context) { Sanitizer s = sanitizer; if (s != null) { + // the commits still in a copy that was never routed never run + s.contextStopped(); // the calls still pending when the context stops are listed s.stopPendingCheck(reportOnShutdown); Review Comment: **[low]** If `contextStopped()` throws (e.g. a `recordCommit` → session assertion failure), `stopPendingCheck` is skipped — pending R1 violations are silently lost. Not a production concern (the context is stopping and this is a dev tool), but a `try/finally` is cleaner. ```suggestion try { // the commits still in a copy that was never routed never run s.contextStopped(); } finally { // the calls still pending when the context stops are listed s.stopPendingCheck(reportOnShutdown); } ``` -- 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]
