diff --git a/CHANGELOG.md b/CHANGELOG.md index b8fc67b6b53..758851410db 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,8 @@ ### Performance +- Batch and coalesce scope-persistence disk writes to reduce startup cost ([#5791](https://github.com/getsentry/sentry-java/pull/5791)) + - Scope mutations are now coalesced (latest value per field) and breadcrumbs are appended in batches behind a single fsync, instead of one synchronous disk write per mutation. - Reduce the number of SDK threads: `LifecycleWatcher` now schedules the session-end task on the shared timer executor instead of creating a dedicated `java.util.Timer` thread ([#5819](https://github.com/getsentry/sentry-java/pull/5819)) ## 8.50.1 diff --git a/sentry/api/sentry.api b/sentry/api/sentry.api index c623e71d08f..61d531ea53e 100644 --- a/sentry/api/sentry.api +++ b/sentry/api/sentry.api @@ -4922,6 +4922,7 @@ public abstract class io/sentry/cache/tape/ObjectQueue : java/io/Closeable, java public fun remove ()V public abstract fun remove (I)V public abstract fun size ()I + public fun sync ()V } public abstract interface class io/sentry/cache/tape/ObjectQueue$Converter { @@ -4942,6 +4943,7 @@ public final class io/sentry/cache/tape/QueueFile : java/io/Closeable, java/lang public fun remove ()V public fun remove (I)V public fun size ()I + public fun sync ()V public fun toString ()Ljava/lang/String; } @@ -4949,6 +4951,7 @@ public final class io/sentry/cache/tape/QueueFile$Builder { public fun (Ljava/io/File;)V public fun build ()Lio/sentry/cache/tape/QueueFile; public fun size (I)Lio/sentry/cache/tape/QueueFile$Builder; + public fun synchronousWrites (Z)Lio/sentry/cache/tape/QueueFile$Builder; public fun zero (Z)Lio/sentry/cache/tape/QueueFile$Builder; } diff --git a/sentry/src/main/java/io/sentry/cache/PersistingScopeObserver.java b/sentry/src/main/java/io/sentry/cache/PersistingScopeObserver.java index 420d0d31e22..0c468381b69 100644 --- a/sentry/src/main/java/io/sentry/cache/PersistingScopeObserver.java +++ b/sentry/src/main/java/io/sentry/cache/PersistingScopeObserver.java @@ -7,7 +7,6 @@ import io.sentry.Breadcrumb; import io.sentry.IScope; import io.sentry.ScopeObserverAdapter; -import io.sentry.SentryExecutorService; import io.sentry.SentryLevel; import io.sentry.SentryOptions; import io.sentry.SpanContext; @@ -29,15 +28,28 @@ import java.io.Reader; import java.io.Writer; import java.nio.charset.Charset; +import java.util.ArrayList; import java.util.Collection; import java.util.Map; +import java.util.Queue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; +import org.jetbrains.annotations.TestOnly; public final class PersistingScopeObserver extends ScopeObserverAdapter { private static final Charset UTF_8 = Charset.forName("UTF-8"); + /** Sentinel value marking a file that should be deleted rather than written on the next flush. */ + private static final Object DELETE_MARKER = new Object(); + + /** Sentinel value marking a breadcrumb clear in the pending breadcrumb operations. */ + private static final Object CLEAR_MARKER = new Object(); + public static final String SCOPE_CACHE = ".scope-cache"; public static final String USER_FILENAME = "user.json"; public static final String BREADCRUMBS_FILENAME = "breadcrumbs.json"; @@ -65,7 +77,11 @@ public final class PersistingScopeObserver extends ScopeObserverAdapter { final File file = new File(cacheDir, BREADCRUMBS_FILENAME); try { try { - queueFile = new QueueFile.Builder(file).size(options.getMaxBreadcrumbs()).build(); + queueFile = + new QueueFile.Builder(file) + .size(options.getMaxBreadcrumbs()) + .synchronousWrites(false) + .build(); } catch (IOException e) { // if file is corrupted we simply delete it and try to create it again. We accept // the trade @@ -74,7 +90,11 @@ public final class PersistingScopeObserver extends ScopeObserverAdapter { // update where the new format was introduced file.delete(); - queueFile = new QueueFile.Builder(file).size(options.getMaxBreadcrumbs()).build(); + queueFile = + new QueueFile.Builder(file) + .size(options.getMaxBreadcrumbs()) + .synchronousWrites(false) + .build(); } } catch (IOException e) { options.getLogger().log(ERROR, "Failed to create breadcrumbs queue", e); @@ -106,32 +126,30 @@ public void toStream(Breadcrumb value, OutputStream sink) throws IOException { }); }); + // Latest pending value per file (or DELETE_MARKER), coalesced until the next flush. + private final @NotNull Map pendingWrites = new ConcurrentHashMap<>(); + // Breadcrumb adds and clears buffered since the last flush, applied together behind a single + // fsync. Adds and clears share one queue so their relative order survives batching: a clear + // must not wipe a breadcrumb added after it. + private final @NotNull Queue pendingBreadcrumbs = new ConcurrentLinkedQueue<>(); + private final @NotNull AtomicBoolean hasPendingFlush = new AtomicBoolean(false); + public PersistingScopeObserver(final @NotNull SentryOptions options) { this.options = options; } @Override public void setUser(final @Nullable User user) { - serializeToDisk( - () -> { - if (user == null) { - delete(USER_FILENAME); - } else { - store(user, USER_FILENAME); - } - }); + enqueue(USER_FILENAME, user); } @Override public void addBreadcrumb(@NotNull Breadcrumb crumb) { - serializeToDisk( - () -> { - try { - breadcrumbsQueue.getValue().add(crumb); - } catch (IOException e) { - options.getLogger().log(ERROR, "Failed to add breadcrumb to file queue", e); - } - }); + if (!options.isEnableScopePersistence()) { + return; + } + pendingBreadcrumbs.offer(crumb); + requestFlush(); } @Override @@ -139,110 +157,157 @@ public void setBreadcrumbs(@NotNull Collection breadcrumbs) { if (breadcrumbs.isEmpty()) { // we only clear the queue if the new collection is empty (someone called clearBreadcrumbs) // If it's not empty, we'd add breadcrumbs one-by-one in the method above - serializeToDisk( - () -> { - try { - breadcrumbsQueue.getValue().clear(); - } catch (IOException e) { - options.getLogger().log(ERROR, "Failed to clear breadcrumbs from file queue", e); - } - }); + if (!options.isEnableScopePersistence()) { + return; + } + pendingBreadcrumbs.offer(CLEAR_MARKER); + requestFlush(); } } @Override public void setTags(@NotNull Map tags) { - serializeToDisk(() -> store(tags, TAGS_FILENAME)); + enqueue(TAGS_FILENAME, tags); } @Override public void setExtras(@NotNull Map extras) { - serializeToDisk(() -> store(extras, EXTRAS_FILENAME)); + enqueue(EXTRAS_FILENAME, extras); } @Override public void setRequest(@Nullable Request request) { - serializeToDisk( - () -> { - if (request == null) { - delete(REQUEST_FILENAME); - } else { - store(request, REQUEST_FILENAME); - } - }); + enqueue(REQUEST_FILENAME, request); } @Override public void setFingerprint(@NotNull Collection fingerprint) { - serializeToDisk(() -> store(fingerprint, FINGERPRINT_FILENAME)); + enqueue(FINGERPRINT_FILENAME, fingerprint); } @Override public void setLevel(@Nullable SentryLevel level) { - serializeToDisk( - () -> { - if (level == null) { - delete(LEVEL_FILENAME); - } else { - store(level, LEVEL_FILENAME); - } - }); + enqueue(LEVEL_FILENAME, level); } @Override public void setTransaction(@Nullable String transaction) { - serializeToDisk( - () -> { - if (transaction == null) { - delete(TRANSACTION_FILENAME); - } else { - store(transaction, TRANSACTION_FILENAME); - } - }); + enqueue(TRANSACTION_FILENAME, transaction); } @Override public void setTrace(@Nullable SpanContext spanContext, @NotNull IScope scope) { - serializeToDisk( - () -> { - if (spanContext == null) { - // we always need a trace_id to properly link with traces/replays, so we fallback to - // propagation context values and create a fake SpanContext - store(scope.getPropagationContext().toSpanContext(), TRACE_FILENAME); - } else { - store(spanContext, TRACE_FILENAME); - } - }); + // we always need a trace_id to properly link with traces/replays, so we fallback to + // propagation context values and create a fake SpanContext + enqueue( + TRACE_FILENAME, + spanContext == null ? scope.getPropagationContext().toSpanContext() : spanContext); } @Override public void setContexts(@NotNull Contexts contexts) { - serializeToDisk(() -> store(contexts, CONTEXTS_FILENAME)); + enqueue(CONTEXTS_FILENAME, contexts); } @Override public void setReplayId(@NotNull SentryId replayId) { - serializeToDisk(() -> store(replayId, REPLAY_FILENAME)); + enqueue(REPLAY_FILENAME, replayId); } - @SuppressWarnings("FutureReturnValueIgnored") - private void serializeToDisk(final @NotNull Runnable task) { + private void enqueue(final @NotNull String fileName, final @Nullable Object entity) { if (!options.isEnableScopePersistence()) { return; } - if (SentryExecutorService.isSentryExecutorThread()) { - // we're already on the sentry executor thread, so we can just execute it directly - runSafely(task); + // latest value wins; a null entity means the file should be deleted on the next flush + pendingWrites.put(fileName, entity == null ? DELETE_MARKER : entity); + requestFlush(); + } + + /** + * Queues a flush unless one is already queued. Coalescing comes from the flush task sitting in + * the executor queue: every mutation that arrives before it runs is folded into the same write. + * The executor is single-threaded, so during startup — when mutations are frequent and the queue + * is deep — that window covers many mutations. + */ + private void requestFlush() { + if (!hasPendingFlush.compareAndSet(false, true)) { + // a flush is already queued; it will pick up the latest pending state return; } - try { - options.getExecutorService().submit(() -> runSafely(task)); + final @NotNull Future future = options.getExecutorService().submit(this::flushOnExecutor); + if (future.isCancelled()) { + // the executor rejects tasks without throwing once its queue is full, so clear the flag or + // no later mutation would ever be able to queue a flush again + hasPendingFlush.set(false); + } } catch (Throwable e) { - options.getLogger().log(ERROR, "Serialization task could not be scheduled", e); + hasPendingFlush.set(false); + options.getLogger().log(ERROR, "Scope persistence flush could not be submitted", e); + } + } + + private void flushOnExecutor() { + runSafely(this::flushPending); + // clear the flag before re-checking, otherwise a mutation landing between the drain and the + // clear would see a flush still queued and be left with nobody to write it + hasPendingFlush.set(false); + if (!pendingWrites.isEmpty() || !pendingBreadcrumbs.isEmpty()) { + requestFlush(); } } + /** Writes all coalesced scope state to disk. Does I/O; must run off the caller/main thread. */ + private void flushPending() { + for (final @NotNull String fileName : new ArrayList<>(pendingWrites.keySet())) { + final @Nullable Object entity = pendingWrites.remove(fileName); + if (entity == null) { + // removed by a concurrent flush + continue; + } + if (entity == DELETE_MARKER) { + delete(fileName); + } else { + store(entity, fileName); + } + } + + boolean breadcrumbsChanged = false; + final @NotNull ObjectQueue queue = breadcrumbsQueue.getValue(); + + Object operation; + while ((operation = pendingBreadcrumbs.poll()) != null) { + try { + if (operation == CLEAR_MARKER) { + queue.clear(); + } else { + queue.add((Breadcrumb) operation); + } + breadcrumbsChanged = true; + } catch (IOException e) { + options.getLogger().log(ERROR, "Failed to apply breadcrumb change to file queue", e); + } + } + + if (breadcrumbsChanged) { + try { + // single fsync for the whole batch instead of one per breadcrumb + queue.sync(); + } catch (IOException e) { + options.getLogger().log(ERROR, "Failed to sync breadcrumbs file queue", e); + } + } + } + + /** + * Synchronously writes any pending scope state to disk. Does I/O on the calling thread, so it's + * only meant for tests and shutdown, not the hot path. + */ + @TestOnly + void flush() { + runSafely(this::flushPending); + } + private void runSafely(final @NotNull Runnable task) { try { task.run(); @@ -286,9 +351,15 @@ public static void store( * I/O and should be called from a background thread. */ public void resetCache() { + // NOTE: pending mutations are deliberately left alone. They only ever hold values from the + // current process, which is exactly what this reset is clearing the way for; dropping them + // would lose scope state set during init. // since it keeps a reference to the file and we cannot delete it, breadcrumbs we just clear try { - breadcrumbsQueue.getValue().clear(); + final @NotNull ObjectQueue queue = breadcrumbsQueue.getValue(); + queue.clear(); + // breadcrumbs use buffered writes, so make the clear durable explicitly + queue.sync(); } catch (IOException e) { options.getLogger().log(ERROR, "Failed to clear breadcrumbs from file queue", e); } diff --git a/sentry/src/main/java/io/sentry/cache/tape/FileObjectQueue.java b/sentry/src/main/java/io/sentry/cache/tape/FileObjectQueue.java index 31b111e03ad..d7cf5cd1f98 100644 --- a/sentry/src/main/java/io/sentry/cache/tape/FileObjectQueue.java +++ b/sentry/src/main/java/io/sentry/cache/tape/FileObjectQueue.java @@ -59,6 +59,11 @@ public void add(T entry) throws IOException { queueFile.add(bytes.getArray(), 0, bytes.size()); } + @Override + public void sync() throws IOException { + queueFile.sync(); + } + @Override public @Nullable T peek() throws IOException { byte[] bytes = queueFile.peek(); diff --git a/sentry/src/main/java/io/sentry/cache/tape/ObjectQueue.java b/sentry/src/main/java/io/sentry/cache/tape/ObjectQueue.java index c92cad36218..49a4ae272bb 100644 --- a/sentry/src/main/java/io/sentry/cache/tape/ObjectQueue.java +++ b/sentry/src/main/java/io/sentry/cache/tape/ObjectQueue.java @@ -54,6 +54,12 @@ public boolean isEmpty() { /** Enqueues an entry that can be processed at any time. */ public abstract void add(T entry) throws IOException; + /** + * Flushes any buffered writes to storage. No-op for queues that already write synchronously or + * are purely in-memory. + */ + public void sync() throws IOException {} + /** * Returns the head of the queue, or {@code null} if the queue is empty. Does not modify the * queue. diff --git a/sentry/src/main/java/io/sentry/cache/tape/QueueFile.java b/sentry/src/main/java/io/sentry/cache/tape/QueueFile.java index bc2ed568267..92fbcf8e4b1 100644 --- a/sentry/src/main/java/io/sentry/cache/tape/QueueFile.java +++ b/sentry/src/main/java/io/sentry/cache/tape/QueueFile.java @@ -130,13 +130,22 @@ public final class QueueFile implements Closeable, Iterable { /** A number of elements at which this queue will wrap around (ring buffer). */ private final int maxElements; + /** + * When {@code false}, writes are buffered by the OS and callers must call {@link #sync()} to make + * them durable. This lets callers batch many adds behind a single fsync instead of paying one + * fsync per write. + */ + private final boolean synchronousWrites; + boolean closed; - static RandomAccessFile initializeFromFile(File file) throws IOException { + static RandomAccessFile initializeFromFile(File file, boolean synchronousWrites) + throws IOException { if (!file.exists()) { // Use a temp file so we don't leave a partially-initialized file. File tempFile = new File(file.getPath() + ".tmp"); - RandomAccessFile raf = open(tempFile); + // The one-time initialization is always synchronous so the header is durable before rename. + RandomAccessFile raf = open(tempFile, true); try { raf.setLength(INITIAL_LENGTH); raf.seek(0); @@ -152,23 +161,36 @@ static RandomAccessFile initializeFromFile(File file) throws IOException { } } - return open(file); + return open(file, synchronousWrites); } - /** Opens a random access file that writes synchronously. */ - private static RandomAccessFile open(File file) throws FileNotFoundException { - return new RandomAccessFile(file, "rwd"); + private static RandomAccessFile open(File file, boolean synchronousWrites) + throws FileNotFoundException { + return new RandomAccessFile(file, synchronousWrites ? "rwd" : "rw"); } - QueueFile(File file, RandomAccessFile raf, boolean zero, int maxElements) throws IOException { + QueueFile( + File file, RandomAccessFile raf, boolean zero, int maxElements, boolean synchronousWrites) + throws IOException { this.file = file; this.raf = raf; this.zero = zero; this.maxElements = maxElements; + this.synchronousWrites = synchronousWrites; readInitialData(); } + /** + * Flushes buffered writes to storage. No-op when the queue was opened with synchronous writes, + * since those are already durable. + */ + public void sync() throws IOException { + if (!synchronousWrites) { + raf.getChannel().force(false); + } + } + private void readInitialData() throws IOException { raf.seek(0); raf.readFully(buffer); @@ -196,7 +218,7 @@ private void readInitialData() throws IOException { private void resetFile() throws IOException { raf.close(); file.delete(); - raf = initializeFromFile(file); + raf = initializeFromFile(file, synchronousWrites); readInitialData(); } @@ -767,6 +789,7 @@ public static final class Builder { final File file; boolean zero = true; int size = -1; + boolean synchronousWrites = true; /** Start constructing a new queue backed by the given file. */ public Builder(File file) { @@ -788,15 +811,24 @@ public Builder size(int size) { return this; } + /** + * When {@code false}, writes are buffered and the caller must call {@link QueueFile#sync()} to + * make them durable. Defaults to {@code true} (every write is fsync'd). + */ + public Builder synchronousWrites(boolean synchronousWrites) { + this.synchronousWrites = synchronousWrites; + return this; + } + /** * Constructs a new queue backed by the given builder. Only one instance should access a given * file at a time. */ public QueueFile build() throws IOException { - RandomAccessFile raf = initializeFromFile(file); + RandomAccessFile raf = initializeFromFile(file, synchronousWrites); QueueFile qf = null; try { - qf = new QueueFile(file, raf, zero, size); + qf = new QueueFile(file, raf, zero, size, synchronousWrites); return qf; } finally { if (qf == null) { diff --git a/sentry/src/test/java/io/sentry/cache/PersistingScopeObserverBatchingTest.kt b/sentry/src/test/java/io/sentry/cache/PersistingScopeObserverBatchingTest.kt new file mode 100644 index 00000000000..310113e2859 --- /dev/null +++ b/sentry/src/test/java/io/sentry/cache/PersistingScopeObserverBatchingTest.kt @@ -0,0 +1,143 @@ +package io.sentry.cache + +import com.google.common.truth.Truth.assertThat +import io.sentry.Breadcrumb +import io.sentry.ISentryExecutorService +import io.sentry.ISerializer +import io.sentry.SentryOptions +import io.sentry.cache.PersistingScopeObserver.BREADCRUMBS_FILENAME +import io.sentry.cache.PersistingScopeObserver.TRANSACTION_FILENAME +import io.sentry.test.DeferredExecutorService +import java.io.Writer +import java.util.concurrent.atomic.AtomicBoolean +import kotlin.test.Test +import org.junit.Rule +import org.junit.rules.TemporaryFolder + +class PersistingScopeObserverBatchingTest { + @get:Rule val tmpDir = TemporaryFolder() + + private val options = SentryOptions() + + private fun getSut(executor: ISentryExecutorService): PersistingScopeObserver { + options.executorService = executor + options.cacheDirPath = tmpDir.newFolder().absolutePath + return PersistingScopeObserver(options) + } + + private fun PersistingScopeObserver.readTransaction(): String? = + read(options, TRANSACTION_FILENAME, String::class.java) + + @Suppress("UNCHECKED_CAST") + private fun PersistingScopeObserver.readBreadcrumbs(): List = + read(options, BREADCRUMBS_FILENAME, List::class.java) as List + + @Test + fun `defers writes until the flush runs`() { + val executor = DeferredExecutorService() + val sut = getSut(executor) + + sut.setTransaction("MainActivity") + assertThat(sut.readTransaction()).isNull() + + executor.runAll() + assertThat(sut.readTransaction()).isEqualTo("MainActivity") + } + + @Test + fun `coalesces repeated writes keeping only the latest value`() { + val executor = DeferredExecutorService() + val sut = getSut(executor) + + sut.setTransaction("A") + sut.setTransaction("B") + sut.setTransaction("C") + executor.runAll() + + assertThat(sut.readTransaction()).isEqualTo("C") + } + + @Test + fun `batches breadcrumbs and persists all of them`() { + val executor = DeferredExecutorService() + val sut = getSut(executor) + + sut.addBreadcrumb(Breadcrumb().apply { message = "one" }) + sut.addBreadcrumb(Breadcrumb().apply { message = "two" }) + assertThat(sut.readBreadcrumbs()).isEmpty() + + executor.runAll() + assertThat(sut.readBreadcrumbs().map { it.message }).containsExactly("one", "two").inOrder() + } + + @Test + fun `clearing breadcrumbs drops earlier pending adds but keeps later ones`() { + val executor = DeferredExecutorService() + val sut = getSut(executor) + + sut.addBreadcrumb(Breadcrumb().apply { message = "dropped" }) + sut.setBreadcrumbs(emptyList()) + sut.addBreadcrumb(Breadcrumb().apply { message = "kept" }) + executor.runAll() + + assertThat(sut.readBreadcrumbs().map { it.message }).containsExactly("kept") + } + + @Test + fun `a clear landing mid-flush does not wipe breadcrumbs added after it`() { + val executor = DeferredExecutorService() + lateinit var sut: PersistingScopeObserver + // fires while the flush is draining, i.e. after the first breadcrumb has been written + options.setSerializer( + SerializerHook(options.serializer) { + sut.setBreadcrumbs(emptyList()) + sut.addBreadcrumb(Breadcrumb().apply { message = "kept" }) + } + ) + sut = getSut(executor) + + sut.addBreadcrumb(Breadcrumb().apply { message = "dropped" }) + executor.runAll() + executor.runAll() + + assertThat(sut.readBreadcrumbs().map { it.message }).containsExactly("kept") + } + + /** Delegates to [delegate], running [onFirstBreadcrumb] once, mid-serialization. */ + private class SerializerHook( + private val delegate: ISerializer, + private val onFirstBreadcrumb: () -> Unit, + ) : ISerializer by delegate { + private val fired = AtomicBoolean(false) + + override fun serialize(entity: T, writer: Writer) { + if (entity is Breadcrumb && fired.compareAndSet(false, true)) { + onFirstBreadcrumb() + } + delegate.serialize(entity, writer) + } + } + + @Test + fun `resetCache keeps pending mutations from the current process`() { + val executor = DeferredExecutorService() + val sut = getSut(executor) + + sut.setTransaction("SetDuringInit") + sut.resetCache() + executor.runAll() + + assertThat(sut.readTransaction()).isEqualTo("SetDuringInit") + } + + @Test + fun `flush writes pending state synchronously`() { + val executor = DeferredExecutorService() + val sut = getSut(executor) + + sut.setTransaction("Sync") + sut.flush() + + assertThat(sut.readTransaction()).isEqualTo("Sync") + } +} diff --git a/sentry/src/test/java/io/sentry/cache/tape/QueueFileTest.kt b/sentry/src/test/java/io/sentry/cache/tape/QueueFileTest.kt index c909b48a9b3..db5f3ebfae7 100644 --- a/sentry/src/test/java/io/sentry/cache/tape/QueueFileTest.kt +++ b/sentry/src/test/java/io/sentry/cache/tape/QueueFileTest.kt @@ -49,7 +49,8 @@ class QueueFileTest { @get:Rule val folder = TemporaryFolder() private lateinit var file: File - private fun newQueueFile(raf: RandomAccessFile): QueueFile = QueueFile(this.file, raf, true, -1) + private fun newQueueFile(raf: RandomAccessFile): QueueFile = + QueueFile(this.file, raf, true, -1, true) private fun newQueueFile(zero: Boolean = true, size: Int = -1): QueueFile = Builder(file).zero(zero).size(size).build() @@ -72,6 +73,21 @@ class QueueFileTest { assertArrayEquals(queue.peek(), expected) } + @Test + fun bufferedWritesSurviveReopenAfterSync() { + var queue = Builder(file).synchronousWrites(false).build() + val first = values[253] + val second = values[25] + queue.add(first) + queue.add(second) + queue.sync() + queue.close() + + queue = newQueueFile() + assertEquals(2, queue.size()) + assertArrayEquals(queue.peek(), first) + } + @Test fun testClearErases() { val queue = newQueueFile()