1010import java .util .concurrent .CompletableFuture ;
1111import java .util .concurrent .atomic .AtomicReference ;
1212import java .util .function .BooleanSupplier ;
13+ import java .util .function .UnaryOperator ;
1314
1415import ai .timefold .solver .core .api .solver .Solver ;
1516import ai .timefold .solver .core .api .solver .change .ProblemChange ;
2021
2122/**
2223 * The goal of this class is to register problem changes and best solutions in a thread-safe way.
23- * Problem changes are {@link #addProblemChange(Solver, ProblemChange ) put in a queue}
24+ * Problem changes are {@link #addProblemChange(Solver, List ) put in a queue}
2425 * and later associated with the best solution which contains them.
2526 * The best solution is associated with a version number
2627 * that is incremented each time a {@link #set new best solution is set}.
2930 *
3031 * <p>
3132 * This class needs to be thread-safe.
32- * Due to complicated interactions between the solver, solver manager and problem changes,
33- * it is best if we avoid explicit locking here,
34- * reducing cognitive complexity of the whole system.
35- * The core idea being to never modify the same data structure from multiple threads;
36- * instead, we replace the data structure with a new one atomically.
37- * The code contains comments throughout the class that explain the reasoning behind the design.
3833 *
3934 * @param <Solution_>
4035 */
4136@ NullMarked
4237final class BestSolutionHolder <Solution_ > {
4338
44- private final AtomicReference <@ Nullable VersionedBestSolution <Solution_ >> versionedBestSolutionRef =
45- new AtomicReference <>();
46- private final AtomicReference <SortedMap <BigInteger , List <CompletableFuture <Void >>>> problemChangesPerVersionRef =
47- new AtomicReference <>(createNewProblemChangesMap ());
39+ private final AtomicReference <BigInteger > lastProcessedVersion = new AtomicReference <>(BigInteger .valueOf (-1 ));
40+
41+ // These references are non-final and being accessed from multiple threads,
42+ // therefore they need to be volatile and all access synchronized.
43+ // Both the map and the best solution are based on the current version,
44+ // and therefore access to both needs to be guarded by the same lock.
4845 // The version is BigInteger to avoid long overflow.
4946 // The solver can run potentially forever, so long overflow is a (remote) possibility.
50- private final AtomicReference <BigInteger > currentVersion = new AtomicReference <>(BigInteger .ZERO );
51- private final AtomicReference <BigInteger > lastProcessedVersion = new AtomicReference <>(BigInteger .valueOf (-1 ));
47+ private volatile SortedMap <BigInteger , List <CompletableFuture <Void >>> problemChangesPerVersionMap =
48+ createNewProblemChangesMap ();
49+ private volatile @ Nullable VersionedBestSolution <Solution_ > versionedBestSolution = null ;
50+ private volatile BigInteger currentVersion = BigInteger .ZERO ;
5251
5352 private static SortedMap <BigInteger , List <CompletableFuture <Void >>> createNewProblemChangesMap () {
5453 return createNewProblemChangesMap (Collections .emptySortedMap ());
@@ -59,8 +58,8 @@ private static SortedMap<BigInteger, List<CompletableFuture<Void>>> createNewPro
5958 return new TreeMap <>(map );
6059 }
6160
62- boolean isEmpty () {
63- return versionedBestSolutionRef . get () == null ;
61+ synchronized boolean isEmpty () {
62+ return this . versionedBestSolution == null ;
6463 }
6564
6665 /**
@@ -69,12 +68,12 @@ boolean isEmpty() {
6968 */
7069 @ Nullable
7170 BestSolutionContainingProblemChanges <Solution_ > take () {
72- var versionedBestSolution = versionedBestSolutionRef . getAndSet ( null );
73- if (versionedBestSolution == null ) {
71+ var latestVersionedBestSolution = resetVersionedBestSolution ( );
72+ if (latestVersionedBestSolution == null ) {
7473 return null ;
7574 }
7675
77- var bestSolutionVersion = versionedBestSolution .version ();
76+ var bestSolutionVersion = latestVersionedBestSolution .version ();
7877 var latestProcessedVersion = this .lastProcessedVersion .getAndUpdate (bestSolutionVersion ::max );
7978 if (latestProcessedVersion .compareTo (bestSolutionVersion ) > 0 ) {
8079 // Corner case: The best solution has already been taken,
@@ -84,7 +83,7 @@ BestSolutionContainingProblemChanges<Solution_> take() {
8483 return null ;
8584 }
8685 // The map is replaced by a map containing only the problem changes that are not contained in the best solution.
87- // This is done atomically , so no other thread can access the old map anymore.
86+ // This is fully synchronized , so no other thread can access the old map anymore.
8887 // The old map can then be processed by the current thread without synchronization.
8988 // The copying of maps is possibly expensive, but due to the nature of problem changes,
9089 // we do not expect the map to ever get too big.
@@ -93,7 +92,7 @@ BestSolutionContainingProblemChanges<Solution_> take() {
9392 // The solver also finds new best solutions, which regularly trims the size of the map as well.
9493 var boundaryVersion = bestSolutionVersion .add (BigInteger .ONE );
9594 var oldProblemChangesPerVersion =
96- problemChangesPerVersionRef . getAndUpdate (map -> createNewProblemChangesMap (map .tailMap (boundaryVersion )));
95+ replaceMapSynchronized (map -> createNewProblemChangesMap (map .tailMap (boundaryVersion )));
9796 // At this point, the old map is not accessible to any other thread.
9897 // We also do not need to clear it, because this being the only reference,
9998 // garbage collector will do it for us.
@@ -102,29 +101,41 @@ BestSolutionContainingProblemChanges<Solution_> take() {
102101 .stream ()
103102 .flatMap (Collection ::stream )
104103 .toList ();
105- return new BestSolutionContainingProblemChanges <>(versionedBestSolution .bestSolution (), containedProblemChanges );
104+ return new BestSolutionContainingProblemChanges <>(latestVersionedBestSolution .bestSolution (), containedProblemChanges );
105+ }
106+
107+ private synchronized @ Nullable VersionedBestSolution <Solution_ > resetVersionedBestSolution () {
108+ var oldVersionedBestSolution = this .versionedBestSolution ;
109+ this .versionedBestSolution = null ;
110+ return oldVersionedBestSolution ;
111+ }
112+
113+ private synchronized SortedMap <BigInteger , List <CompletableFuture <Void >>> replaceMapSynchronized (
114+ UnaryOperator <SortedMap <BigInteger , List <CompletableFuture <Void >>>> replaceFunction ) {
115+ var oldMap = problemChangesPerVersionMap ;
116+ problemChangesPerVersionMap = replaceFunction .apply (oldMap );
117+ return oldMap ;
106118 }
107119
108120 /**
109- * Sets the new best solution if all known problem changes have been processed and thus are contained in this
110- * best solution.
121+ * Sets the new best solution if all known problem changes have been processed
122+ * and thus are contained in this best solution.
111123 *
112124 * @param bestSolution the new best solution that replaces the previous one if there is any
113125 * @param isEveryProblemChangeProcessed a supplier that tells if all problem changes have been processed
114126 */
115127 void set (Solution_ bestSolution , BooleanSupplier isEveryProblemChangeProcessed ) {
116- /*
117- * The new best solution can be accepted only if there are no pending problem changes
118- * nor any additional changes may come during this operation.
119- * Otherwise, a race condition might occur
120- * that leads to associating problem changes with a solution that was created later,
121- * but does not contain them yet.
122- * As a result, CompletableFutures representing these changes would be completed too early.
123- */
128+ // The new best solution can be accepted only if there are no pending problem changes
129+ // nor any additional changes may come during this operation.
130+ // Otherwise, a race condition might occur
131+ // that leads to associating problem changes with a solution that was created later,
132+ // but does not contain them yet.
133+ // As a result, CompletableFutures representing these changes would be completed too early.
124134 if (isEveryProblemChangeProcessed .getAsBoolean ()) {
125- // This field is atomic, so we can safely set the new best solution without synchronization.
126- versionedBestSolutionRef .set (
127- new VersionedBestSolution <>(bestSolution , currentVersion .getAndUpdate (old -> old .add (BigInteger .ONE ))));
135+ synchronized (this ) {
136+ versionedBestSolution = new VersionedBestSolution <>(bestSolution , currentVersion );
137+ currentVersion = currentVersion .add (BigInteger .ONE );
138+ }
128139 }
129140 }
130141
@@ -139,23 +150,20 @@ void set(Solution_ bestSolution, BooleanSupplier isEveryProblemChangeProcessed)
139150 CompletableFuture <Void > addProblemChange (Solver <Solution_ > solver , List <ProblemChange <Solution_ >> problemChangeList ) {
140151 var futureProblemChange = new CompletableFuture <Void >();
141152 synchronized (this ) {
142- // This actually needs to be synchronized,
143- // as we want the new problem change and its version to be linked.
144- var futureProblemChangeList =
145- problemChangesPerVersionRef .get ().computeIfAbsent (currentVersion .get (), version -> new ArrayList <>());
153+ var futureProblemChangeList = problemChangesPerVersionMap .computeIfAbsent (currentVersion ,
154+ version -> new ArrayList <>());
146155 futureProblemChangeList .add (futureProblemChange );
147156 solver .addProblemChanges (problemChangeList );
148157 }
149158 return futureProblemChange ;
150159 }
151160
152161 void cancelPendingChanges () {
153- // The map is an atomic reference.
154- // We first replace the reference with a new map atomically, avoiding synchronization issues.
155- // Then we process the old map, which is safe because no one can access it anymore.
162+ // We first replace the reference with a new map, fully synchronized.
163+ // Then we process the old map unsynchronized, which is safe because no one can access it anymore.
156164 // We do not need to clear it, because this being the only reference,
157165 // the garbage collector will do it for us.
158- problemChangesPerVersionRef . getAndSet ( createNewProblemChangesMap ())
166+ replaceMapSynchronized ( map -> createNewProblemChangesMap ())
159167 .values ()
160168 .stream ()
161169 .flatMap (Collection ::stream )
0 commit comments