1 /*
2 * Copyright (C) 2015 The Guava Authors
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
5 * in compliance with the License. You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software distributed under the License
10 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
11 * or implied. See the License for the specific language governing permissions and limitations under
12 * the License.
13 */
14
15 package com.google.common.util.concurrent;
16
17 import static com.google.common.collect.Sets.newConcurrentHashSet;
18 import static java.util.Objects.requireNonNull;
19 import static java.util.concurrent.atomic.AtomicIntegerFieldUpdater.newUpdater;
20 import static java.util.concurrent.atomic.AtomicReferenceFieldUpdater.newUpdater;
21
22 import com.google.common.annotations.GwtCompatible;
23 import com.google.j2objc.annotations.ReflectionSupport;
24 import java.util.Set;
25 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
26 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
27 import java.util.logging.Level;
28 import java.util.logging.Logger;
29 import javax.annotation.CheckForNull;
30 import org.checkerframework.checker.nullness.qual.Nullable;
31
32 /**
33 * A helper which does some thread-safe operations for aggregate futures, which must be implemented
34 * differently in GWT. Namely:
35 *
36 * <ul>
37 * <li>Lazily initializes a set of seen exceptions
38 * <li>Decrements a counter atomically
39 * </ul>
40 */
41 @GwtCompatible(emulated = true)
42 @ReflectionSupport(value = ReflectionSupport.Level.FULL)
43 @ElementTypesAreNonnullByDefault
44 abstract class AggregateFutureState<OutputT extends @Nullable Object>
45 extends AbstractFuture.TrustedFuture<OutputT> {
46 // Lazily initialized the first time we see an exception; not released until all the input futures
47 // have completed and we have processed them all.
48 @CheckForNull private volatile Set<Throwable> seenExceptions = null;
49
50 private volatile int remaining;
51
52 private static final AtomicHelper ATOMIC_HELPER;
53
54 private static final Logger log = Logger.getLogger(AggregateFutureState.class.getName());
55
56 static {
57 AtomicHelper helper;
58 Throwable thrownReflectionFailure = null;
59 try {
60 helper =
61 new SafeAtomicHelper(
62 newUpdater(AggregateFutureState.class, Set.class, "seenExceptions"),
63 newUpdater(AggregateFutureState.class, "remaining"));
64 } catch (Throwable reflectionFailure) {
65 // Some Android 5.0.x Samsung devices have bugs in JDK reflection APIs that cause
66 // getDeclaredField to throw a NoSuchFieldException when the field is definitely there.
67 // For these users fallback to a suboptimal implementation, based on synchronized. This will
68 // be a definite performance hit to those users.
69 thrownReflectionFailure = reflectionFailure;
70 helper = new SynchronizedAtomicHelper();
71 }
72 ATOMIC_HELPER = helper;
73 // Log after all static init is finished; if an installed logger uses any Futures methods, it
74 // shouldn't break in cases where reflection is missing/broken.
75 if (thrownReflectionFailure != null) {
76 log.log(Level.SEVERE, "SafeAtomicHelper is broken!", thrownReflectionFailure);
77 }
78 }
79
80 AggregateFutureState(int remainingFutures) {
81 this.remaining = remainingFutures;
82 }
83
84 final Set<Throwable> getOrInitSeenExceptions() {
85 /*
86 * The initialization of seenExceptions has to be more complicated than we'd like. The simple
87 * approach would be for each caller CAS it from null to a Set populated with its exception. But
88 * there's another race: If the first thread fails with an exception and a second thread
89 * immediately fails with the same exception:
90 *
91 * Thread1: calls setException(), which returns true, context switch before it can CAS
92 * seenExceptions to its exception
93 *
94 * Thread2: calls setException(), which returns false, CASes seenExceptions to its exception,
95 * and wrongly believes that its exception is new (leading it to logging it when it shouldn't)
96 *
97 * Our solution is for threads to CAS seenExceptions from null to a Set populated with _the
98 * initial exception_, no matter which thread does the work. This ensures that seenExceptions
99 * always contains not just the current thread's exception but also the initial thread's.
100 */
101 Set<Throwable> seenExceptionsLocal = seenExceptions;
102 if (seenExceptionsLocal == null) {
103 // TODO(cpovirk): Should we use a simpler (presumably cheaper) data structure?
104 /*
105 * Using weak references here could let us release exceptions earlier, but:
106 *
107 * 1. On Android, querying a WeakReference blocks if the GC is doing an otherwise-concurrent
108 * pass.
109 *
110 * 2. We would probably choose to compare exceptions using == instead of equals() (for
111 * consistency with how weak references are cleared). That's a behavior change -- arguably the
112 * removal of a feature.
113 *
114 * Fortunately, exceptions rarely contain references to expensive resources.
115 */
116
117 //
118 seenExceptionsLocal = newConcurrentHashSet();
119 /*
120 * Other handleException() callers may see this as soon as we publish it. We need to populate
121 * it with the initial failure before we do, or else they may think that the initial failure
122 * has never been seen before.
123 */
124 addInitialException(seenExceptionsLocal);
125
126 ATOMIC_HELPER.compareAndSetSeenExceptions(this, null, seenExceptionsLocal);
127 /*
128 * If another handleException() caller created the set, we need to use that copy in case yet
129 * other callers have added to it.
130 *
131 * This read is guaranteed to get us the right value because we only set this once (here).
132 *
133 * requireNonNull is safe because either our compareAndSet succeeded or it failed because
134 * another thread did it for us.
135 */
136 seenExceptionsLocal = requireNonNull(seenExceptions);
137 }
138 return seenExceptionsLocal;
139 }
140
141 /** Populates {@code seen} with the exception that was passed to {@code setException}. */
142 abstract void addInitialException(Set<Throwable> seen);
143
144 final int decrementRemainingAndGet() {
145 return ATOMIC_HELPER.decrementAndGetRemainingCount(this);
146 }
147
148 final void clearSeenExceptions() {
149 seenExceptions = null;
150 }
151
152 private abstract static class AtomicHelper {
153 /** Atomic compare-and-set of the {@link AggregateFutureState#seenExceptions} field. */
154 abstract void compareAndSetSeenExceptions(
155 AggregateFutureState<?> state, @CheckForNull Set<Throwable> expect, Set<Throwable> update);
156
157 /** Atomic decrement-and-get of the {@link AggregateFutureState#remaining} field. */
158 abstract int decrementAndGetRemainingCount(AggregateFutureState<?> state);
159 }
160
161 private static final class SafeAtomicHelper extends AtomicHelper {
162 final AtomicReferenceFieldUpdater<AggregateFutureState<?>, Set<Throwable>>
163 seenExceptionsUpdater;
164
165 final AtomicIntegerFieldUpdater<AggregateFutureState<?>> remainingCountUpdater;
166
167 @SuppressWarnings({"rawtypes", "unchecked"}) // Unavoidable with reflection API
168 SafeAtomicHelper(
169 AtomicReferenceFieldUpdater seenExceptionsUpdater,
170 AtomicIntegerFieldUpdater remainingCountUpdater) {
171 this.seenExceptionsUpdater =
172 (AtomicReferenceFieldUpdater<AggregateFutureState<?>, Set<Throwable>>)
173 seenExceptionsUpdater;
174 this.remainingCountUpdater =
175 (AtomicIntegerFieldUpdater<AggregateFutureState<?>>) remainingCountUpdater;
176 }
177
178 @Override
179 void compareAndSetSeenExceptions(
180 AggregateFutureState<?> state, @CheckForNull Set<Throwable> expect, Set<Throwable> update) {
181 seenExceptionsUpdater.compareAndSet(state, expect, update);
182 }
183
184 @Override
185 int decrementAndGetRemainingCount(AggregateFutureState<?> state) {
186 return remainingCountUpdater.decrementAndGet(state);
187 }
188 }
189
190 private static final class SynchronizedAtomicHelper extends AtomicHelper {
191 @Override
192 void compareAndSetSeenExceptions(
193 AggregateFutureState<?> state, @CheckForNull Set<Throwable> expect, Set<Throwable> update) {
194 synchronized (state) {
195 if (state.seenExceptions == expect) {
196 state.seenExceptions = update;
197 }
198 }
199 }
200
201 @Override
202 int decrementAndGetRemainingCount(AggregateFutureState<?> state) {
203 synchronized (state) {
204 return --state.remaining;
205 }
206 }
207 }
208 }